Kafka Java Producer 无法将消息发送到 Kafka 实例。

huangapple go评论66阅读模式
英文:

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: &quot;3&quot;
services:
  zookeeper:
    image: &#39;bitnami/zookeeper:latest&#39;
    ports:
      - &#39;2181:2181&#39;
    environment:
     - ALLOW_ANONYMOUS_LOGIN=yes
  kafka:
    image: &#39;bitnami/kafka:latest&#39;
    ports:
      - &#39;9092:9092&#39;
      - &#39;9093:9093&#39;
    environment:
      - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
      - ALLOW_PLAINTEXT_LISTENER=yes
      - KAFKA_BROKER_ID=1
      - KAFKA_CREATE_TOPICS=&quot;topic_name:1:3&quot;
      - 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&lt;String,String&gt; kafkaTemplate = new KafkaTemplate&lt;&gt;(new DefaultKafkaProducerFactory&lt;&gt;(kafkaConfig()));
    kafkaTemplate.send(&quot;topic_name&quot;, &quot;test&quot;);
}

public static Map&lt;String, Object&gt; kafkaConfig() {
    Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();

    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, &quot;127.0.0.1:9092&quot;);
    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 &quot;topic_name&quot; with 1 partitions:
   partition 0, leader 1, replicas: 1, isrs: 1
topic &quot;__consumer_offsets&quot; 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(&quot;topic_name&quot;, &quot;test&quot;).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.

huangapple
  • 本文由 发表于 2020年8月10日 04:59:49
  • 转载请务必保留本文链接:https://go.coder-hub.com/63331265.html
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定