Kafka版本升级(0.10.00.10.2)

 1 Properties props = new Properties();
 2         props.put("bootstrap.servers", "localhost:9092,localhost:9093");
 3         props.put("acks", "all");
 4         props.put("retries", Integer.MAX_VALUE);
 5         props.put("batch.size", 16384);
 6         props.put("linger.ms", 1);
 7         props.put("buffer.memory", 33554432);
 8         props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 9         props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
10 
11         Producer producer = new KafkaProducer<>(props);
12         final AtomicInteger success = new AtomicInteger(0);
13         final AtomicInteger failed = new AtomicInteger(0);
14         try {
15             while (true) {
16                 producer.send(new ProducerRecord("test", "a message"), new Callback() {
17                     @Override
18                     public void onCompletion(RecordMetadata metadata, Exception exception) {
19                         if (exception != null) {
20                             System.out.println("Current failed count: " + failed.incrementAndGet());
21                         } else
22                             System.out.println("Current success count: " + success.incrementAndGet()
23                                     + ", failed: " + failed.get());
24                     }
25                 });
26                 Thread.sleep(2000);
27 
28             }
29         } finally {
30             producer.close();
31         }