无法从 KafkaProducer 发布消息答案

【问题标题】：Unable to publish message from KafkaProducer无法从 KafkaProducer 发布消息
【发布时间】：2018-07-30 14:20:09
【问题描述】：

我无法从 Kafka Producer 发送消息。我的配置不起作用，它看起来像这样：

Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "127.0.0.1:9092");
properties.setProperty("key.serializer", StringSerializer.class.getName());
properties.setProperty("value.serializer", StringSerializer.class.getName());
properties.setProperty("acks", "1");
properties.setProperty("retries", "3");
properties.setProperty("linger.ms", "1");

Producer<String, String> producer =
    new org.apache.kafka.clients.producer.KafkaProducer<String, String>(properties);
ProducerRecord<String, String> producerRecord =
    new ProducerRecord<String, String>("second_topic", "3", "messagtest");
Future<RecordMetadata> s = producer.send(producerRecord);

producer.flush();
producer.close();

执行 s.get(); 后出现错误

 java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Expiring 1 record(s) for second_topic-0: 30021 ms has passed since batch creation plus linger time
at org.apache.kafka.clients.producer.internals.FutureRecordMetadata.valueOrError(FutureRecordMetadata.java:94)
at org.apache.kafka.clients.producer.internals.FutureRecordMetadata.get(FutureRecordMetadata.java:64)
at org.apache.kafka.clients.producer.internals.FutureRecordMetadata.get(FutureRecordMetadata.java:29)
at ai.sys.producer.Test.main(Test.java:33)
Caused by: org.apache.kafka.common.errors.TimeoutException: Expiring 1 record(s) for second_topic-0: 30021 ms has passed since batch creation plus linger time

【问题讨论】：

会发生什么？有没有错误？
在关闭生产者或发送多条消息之前尝试运行s.get()
@MickaelMaison 没有任何错误。
@MickaelMaison 这是我遇到的错误。
@cricket_007 我在你的指导下得到了这个

标签： java apache-kafka kafka-producer-api

【解决方案1】：

默认情况下，Kafka Producer 中启用了批处理，大小为 16K 字节。但是，在您的代码中，您只发送一条可能不满足批量大小的记录。

因此，为了使您的代码正常工作，请尝试将“batch.size”的以下属性添加到 Kafka Producer 属性中的值“0”。

properties.setProperty("batch.size", "0");

这将禁用批处理机制并允许您的生产者将记录写入 Kafka 代理。

注意：实时禁用批处理会增加对代理的写入请求数量，并降低生产者和服务器的 I/O 吞吐量和性能。

【讨论】：