我无法发送Kafka制作人的消息。我的配置不起作用，看起来是这样的：
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
1条答案
按热度按时间mitkmikd1#
默认情况下，kafka producer中启用了16k字节大小的批处理。但是，在代码中，您只发送一条可能不满足批处理大小的记录。
因此，要使代码正常工作，请尝试将以下属性“batch.size”添加到kafka producer属性的值“0”中。
这将禁用批处理机制，并允许生产者将记录写入kafka代理。
注意：在实时情况下，禁用批处理将增加对代理的写入请求数，并降低生产者和服务器的i/o吞吐量和性能。