英文:
Kafka Java Producer is not able to send message to kafka instance
问题
我正在运行一个在Docker容器中的Kafka实例,使用以下docker-compose.yml文件。
version: "3"
services:
zookeeper:
image: 'bitnami/zookeeper:latest'
ports:
- '2181:2181'
environment:
- ALLOW_ANONYMOUS_LOGIN=yes
kafka:
image: 'bitnami/kafka:latest'
ports:
- '9092:9092'
- '9093:9093'
environment:
- KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
- ALLOW_PLAINTEXT_LISTENER=yes
- KAFKA_BROKER_ID=1
- KAFKA_CREATE_TOPICS="topic_name:1:3"
- KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CLIENT:PLAINTEXT,EXTERNAL:PLAINTEXT
- KAFKA_CFG_LISTENERS=CLIENT://:9092,EXTERNAL://:9093
- KAFKA_CFG_ADVERTISED_LISTENERS=CLIENT://127.0.0.1:9092,EXTERNAL://127.0.0.1:9093
- KAFKA_INTER_BROKER_LISTENER_NAME=EXTERNAL
depends_on:
- zookeeper
它运行得相当好。我通过使用kafkacat发送数据来测试Kafka,没有问题,我能够通过Kafka消费者接收数据。请检查以下kafkacat命令。
kafkacat -P -b 127.0.0.1:9092 -t topic_name
kafkacat -C -b 127.0.0.1:9092 -t topic_name
然而,当我尝试使用Java生产者代码发送数据时,我无法从kafkacat消费者那里接收到数据。请检查下面的Java生产者代码。我想听听您的建议?提前感谢。
public class DataProducer {
public static void main(String[] args) {
KafkaTemplate<String,String> kafkaTemplate = new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(kafkaConfig()));
kafkaTemplate.send("topic_name", "test");
}
public static Map<String, Object> kafkaConfig() {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
return props;
}}
此外,以下是包含元数据的kafkacat命令输出。
kafkacat -b 127.0.0.1:9092 -L
Metadata for all topics (from broker 1: 127.0.0.1:9092/1):
1 brokers:
broker 1 at 127.0.0.1:9092 (controller)
2 topics:
topic "topic_name" with 1 partitions:
partition 0, leader 1, replicas: 1, isrs: 1
topic "__consumer_offsets" with 50 partitions:
partition 0, leader 1, replicas: 1, isrs: 1
partition 1, leader 1, replicas: 1, isrs: 1
..
partition 48, leader 1, replicas: 1, isrs: 1
partition 49, leader 1, replicas: 1, isrs: 1
英文:
I am running a kafka instance in a docker container with following docker-compose.yml file.
version: "3"
services:
zookeeper:
image: 'bitnami/zookeeper:latest'
ports:
- '2181:2181'
environment:
- ALLOW_ANONYMOUS_LOGIN=yes
kafka:
image: 'bitnami/kafka:latest'
ports:
- '9092:9092'
- '9093:9093'
environment:
- KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
- ALLOW_PLAINTEXT_LISTENER=yes
- KAFKA_BROKER_ID=1
- KAFKA_CREATE_TOPICS="topic_name:1:3"
- KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CLIENT:PLAINTEXT,EXTERNAL:PLAINTEXT
- KAFKA_CFG_LISTENERS=CLIENT://:9092,EXTERNAL://:9093
- KAFKA_CFG_ADVERTISED_LISTENERS=CLIENT://127.0.0.1:9092,EXTERNAL://127.0.0.1:9093
- KAFKA_INTER_BROKER_LISTENER_NAME=EXTERNAL
depends_on:
- zookeeper
It is running pretty much good. I tested kafka by sending data via kafkacat. no problem i am able to receive the data via kafka consumer. Please check following kafkacat commands.
kafkacat -P -b 127.0.0.1:9092 -t topic_name
kafkacat -C -b 127.0.0.1:9092 -t topic_name
However when i tried to send it by java producer code, I am not able to receive from kafkacat consumer. Please check java producer code below. I would like to hear your suggestions? Thanks in advance
public class DataProducer {
public static void main(String[] args) {
KafkaTemplate<String,String> kafkaTemplate = new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(kafkaConfig()));
kafkaTemplate.send("topic_name", "test");
}
public static Map<String, Object> kafkaConfig() {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
return props;
}}
Adding also metadata following kafkacat command output.
kafkacat -b 127.0.0.1:9092 -L
Metadata for all topics (from broker 1: 127.0.0.1:9092/1):
1 brokers:
broker 1 at 127.0.0.1:9092 (controller)
2 topics:
topic "topic_name" with 1 partitions:
partition 0, leader 1, replicas: 1, isrs: 1
topic "__consumer_offsets" with 50 partitions:
partition 0, leader 1, replicas: 1, isrs: 1
partition 1, leader 1, replicas: 1, isrs: 1
..
partition 48, leader 1, replicas: 1, isrs: 1
partition 49, leader 1, replicas: 1, isrs: 1
答案1
得分: 3
经纪人配置似乎正常,因为您收到了正确的元数据。
我认为问题出在您的代码中。kafkaTemplate.send()
是一个异步操作,很可能在生产者成功发送消息之前,您的进程已经结束。尝试在发送方法中添加 .get()
以强制它成为同步操作。
kafkaTemplate.send("topic_name", "test").get();
英文:
Broker configuration seems to be fine since you get back the correct metadata.
I think the problem is in your code. kafkaTemplate.send()
is an asynchronous operation and most likely your process ends before the producer manages to actually send the message. Try adding a .get()
to that send method to force it in being synchronous.
kafkaTemplate.send("topic_name", "test").get();
答案2
得分: 0
我遇到过类似的问题。很奇怪,但将 kafka-clients 版本的Maven依赖从latest(2.6.0)
更改为2.0.0
(或升级到2.5.0
)对我有帮助。
英文:
I had similar issue. It's strange, but change in maven dependency of kafka-clients version from latest(2.6.0)
to 2.0.0
(or up to 2.5.0
) helped me.
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库,让每个人都能够通过互相帮助和分享经验来进步。
评论