New client access for new versions of Kafka

[hhj1988]

I tried to modify your code, but failed to ask for help! for example： Properties props = new Properties(); props.put("bootstrap.servers", "locahost:9094"); props.put("group.id", "groupId"); props.put("enable.auto.commit", "true"); props.put("auto.commit.interval.ms", "1000"); props.put("session.timeout.ms", "30000"); props.put("auto.offset.reset", "earliest"); props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); this.consumer = new KafkaConsumer<>(props); consumer.subscribe(Arrays.asList(topics)); while (true) { ConsumerRecords<String, String> records = consumer.poll(200); for (ConsumerRecord<String, String> record : records) { System.out.println(Thread.currentThread().getName() + " partition: "

• record.partition() + " offset: " + record.offset()
• " key:" + record.key() + " value:"
• record.value()); } }

[NicolasReyrolle]

I am planning to work on this in the next days/weeks. Nevertheless, I would like to know the best approach for that. @spektom, do you plan to maintain multiple versions of your plugin, depending on the Kafka version? Because the current 0.11 is very different from your implementation of 0.8 and they will not be compatible? I would tend to upgrade to the latest, but that would mean dropping previous support. Please let me know.

[spektom]

@NicolasReyrolle as far us I understand, having multiple plug-ins is the more straightforward approach here.

[qin4zhang]

How to set Kafka broker to connect, not only zk?