如果消息是由生产者产生的，如何从Kafka经纪人那里得到确认？

Question

有没有其他方法可以得到认可？

@Override
    public void onCompletion(RecordMetadata metadata, Exception exception) {
        long elapsedTime = System.currentTimeMillis() - startTime;
        System.out.println("Called Callback method");
        if (metadata != null) {
            System.out.println("message(" + key + ", " + message
                    + ") sent to partition(" + metadata.partition() + "), "
                    + "offset(" + metadata.offset() + ") in " + elapsedTime
                    + " ms");
        } else {
            exception.printStackTrace();
        }

    }

props.put("bootstrap.servers", "localhost:9092");
props.put("client.id", "mytopic");
props.put("key.serializer", org.apache.kafka.common.serialization.StringSerializer.class);
props.put("value.serializer", org.apache.kafka.common.serialization.ByteArraySerializer.class);

KafkaProducer<String, byte[]> producer = new KafkaProducer<String,byte[]>(props);
long runtime = new Date().getTime(); 
String ip = "192.168.2."+ rnd.nextInt(255); 
String msg = runtime + ".www.ppop.com," + ip;
producer.send(new ProducerRecord<String, byte[]>("mytopic", msg.getBytes()), `new TransCallBack(Calendar.getInstance().getTimeInMillis(), key, msg));`

我使用kafka-client api 0.9.1和broker版本0.8.2。

Answer 1

所以我不能百分之百地确定Kafaka的哪种版本与哪种版本相一致。目前我使用的是0.8.2，我知道0.9引入了一些突破性的变化，但我不能肯定地告诉你现在什么是可行的/不可行的。

一个非常强烈的建议是，我会使用对应于您的代理版本的Kafka-Client版本。如果您使用的是Broker0.8.2，那么我也会使用kakfa-client0.8.2。

您从未提供过如何使用它的代码，所以我只是在暗中猜测。但是我已经通过在Producer中使用这个方法在Kafka0.8.2中实现了回调特性。下面是方法签名。

public java.util.concurrent.Future<RecordMetadata> send(ProducerRecord<K,V> record, Callback callback)

KafkaProducer<String, String> prod = new KafkaProducer<String, String>(props);
ProducerRecord<String, String> record = //data to send to kafka
prod.send(record, new Callback() {
  @Override
  public void onCompletion(RecordMetadata metadata, Exception e) {
    if (e != null) {
      e.printStackTrace();
    } else {
      //implement logic here, or call another method to process metadata
      System.out.println("Callback");
    }
  }
});