发布完所有消息后，关闭producer，如下所示：

producer.close();

这样可以确保始终调用回调中的以下方法：

onCompletion(RecordMetadata metadata, Exception exception)

注：我已经测试了这一点，通过添加该行到这个示例生产者类，它的作品。

在kafkaproducer使用kafka0.9版本发布消息之后，有一种从代理获取信息的简单方法。您可以调用get（）方法，该方法将返回一个recordmetadata对象，您可以在代码段下面获取诸如offset、topicpartition等信息，例如：

RecordMetadata m = kafkaProducer.send(new ProducerRecord<byte[], byte[]>(
                        topic, key.getBytes("UTF-8"), message
                                .getBytes("UTF-8"))).get();
System.out.println("Message produced, offset: " + m.offset());
System.out.println("Message produced, partition : " + m.partition());
System.out.println("Message produced, topic: " + m.topic());

所以我不能百分之百确定Kafka中的哪个版本适用于哪个版本。目前我使用的是0.8.2，我知道0.9引入了一些突破性的变化，但我不能肯定地告诉您现在什么是有效的/无效的。
一个非常强烈的建议是，我将使用与您的代理版本相对应的kafka客户机版本。如果您使用的是broker0.8.2，那么我也会使用kakfa客户端0.8.2。
你从来没有提出过你是如何使用它的任何代码，所以我只是有点猜在黑暗中。但是我已经在kafka0.8.2中通过在producer中使用这个方法实现了回调特性。下面是方法签名。

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");
    }
  }
});

我想有一种方法也可以像你那样做。但是你必须提供代码来显示你是如何将记录发送给Kafka的。除此之外我只是猜测。