CODEWITHSUNDEEP
javaapache-kafkakafka-consumer-apikafka-producer-api
luckystones
luckystones

Reputation: 371

Kafka Java Producer is not able to send message to kafka instance

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

Upvotes: 1

Views: 772

Answers (2)

Greg K
Greg K

Reputation: 1

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.

Upvotes: 0

BogdanSucaciu
BogdanSucaciu

Reputation: 904

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();

Upvotes: 3

Related Questions