Docker Community Forums

Share and learn in the Docker community.

Connecting Kafka producer to Kafka broker in Docker through Java

I want to make a Kafka producer send messages to a topic in a Kafka broker, runned in Docker, using org.apache.kafka Java library. Here’s what I’ve done:

I started a Kafka and a Zookeeper server using docker-compose up -d with this docker-compose.yml file:

version: '2'

services:
  zookeeper:
    image: 'bitnami/zookeeper:3'
    ports:
      - '2181:2181'
    volumes:
      - 'zookeeper_data:/bitnami'
    environment:
      - ALLOW_ANONYMOUS_LOGIN=yes
  kafka:
    image: 'bitnami/kafka:2'
    ports:
      - '9092:9092'
      - '29092:29092'
    volumes:
      - 'kafka_data:/bitnami'
    environment:
      - KAFKA_BROKER_ID=1
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,PLAINTEXT_HOST://:29092
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
      - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
      - ALLOW_PLAINTEXT_LISTENER=yes
    depends_on:
      - zookeeper

volumes:
  zookeeper_data:
    driver: local
  kafka_data:
    driver: local

Output of docker-compose ps:

          Name                     Command           State                          Ports
-----------------------------------------------------------------------------------------------------------------
dockertoolbox_kafka_1       /entrypoint.sh /run.sh   Up      0.0.0.0:29092->29092/tcp, 0.0.0.0:9092->9092/tcp
dockertoolbox_zookeeper_1   /entrypoint.sh /run.sh   Up      0.0.0.0:2181->2181/tcp, 2888/tcp, 3888/tcp, 8080/tcp

Then, I successfully created a topic via Docker terminal:

docker exec -it dockertoolbox_kafka_1 kafka-topics.sh --create --zookeeper zookeeper:2181 --replication-factor 1 --partitions 1 --topic testTopic

The problem is when I try to connect a KafkaProducer from an external Java app (code taken from https://www.tutorialkart.com/apache-kafka/producer-example-in-apache-kafka/ and edited the port):

public class SampleProducer extends Thread {
    private final KafkaProducer<Integer, String> producer;
    private final String topic;
    private final Boolean isAsync;
 
    public static final String KAFKA_SERVER_URL = "localhost";
    public static final int KAFKA_SERVER_PORT = 29092;
    public static final String CLIENT_ID = "SampleProducer";
 
    public SampleProducer(String topic, Boolean isAsync) {
        final Properties properties = new Properties();
        properties.put("bootstrap.servers", KAFKA_SERVER_URL + ":" + KAFKA_SERVER_PORT);
        properties.put("client.id", CLIENT_ID);
        properties.put("key.serializer", "org.apache.kafka.common.serialization.IntegerSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        producer = new KafkaProducer<>(properties);
        this.topic = topic;
        this.isAsync = isAsync;
    }
 
    @Override
    public void run() {
        int messageNo = 1;
        while (true) {
            final String messageStr = "Message_" + messageNo;
            final long startTime = System.currentTimeMillis();
            if (isAsync) { // Send asynchronously
                producer.send(new ProducerRecord<>(topic,
                        messageNo,
                        messageStr), new DemoCallBack(startTime, messageNo, messageStr));
            } else { // Send synchronously
                try {
                    producer.send(new ProducerRecord<>(topic,
                            messageNo,
                            messageStr)).get();
                    System.out.println("Sent message: (" + messageNo + ", " + messageStr + ")");
                } catch (InterruptedException | ExecutionException e) {
                    e.printStackTrace();
                    // handle the exception
                }
            }
            ++messageNo;
        }
    }
}
 
class DemoCallBack implements Callback {
 
    private final long startTime;
    private final int key;
    private final String message;
 
    public DemoCallBack(long startTime, int key, String message) {
        this.startTime = startTime;
        this.key = key;
        this.message = message;
    }
 
    /**
     * onCompletion method will be called when the record sent to the Kafka Server has been acknowledged.
     *
     * @param metadata  The metadata contains the partition and offset of the record. Null if an error occurred.
     * @param exception The exception thrown during processing of this record. Null if no error occurred.
     */
    @Override
    public void onCompletion(RecordMetadata metadata, Exception exception) {
        final long elapsedTime = System.currentTimeMillis() - startTime;
        if (metadata != null) {
            System.out.println(
                    "message(" + key + ", " + message + ") sent to partition(" + metadata.partition() +
                    "), " +
                    "offset(" + metadata.offset() + ") in " + elapsedTime + " ms");
        } else {
            exception.printStackTrace();
        }
    }
}
public class KafkaProducerDemo {
    public static final String TOPIC = "testTopic";

    public static void main(String[] args) {
        final boolean isAsync = true;
        final SampleProducer producerThread = new SampleProducer(TOPIC, isAsync);
        // start the producer
        producerThread.start();

    }
}

When I run KafkaProducerDemo, it prints this at every attempt to send a message to the docker:

[kafka-producer-network-thread | SampleProducer] WARN org.apache.kafka.clients.NetworkClient - [Producer clientId=SampleProducer] Connection to node -1 (localhost/127.0.0.1:29092) could not be established. Broker may not be available.

What’s wrong? I’m desperate

Try KAFKA_SERVER_PORT = 9092. The external Java app must be executed on the same host as your docker containers, otherwise localhost won’t work.

What is your port mapping 29092:29092 supposed to do?

Thanks, but nothing changed.

About 29092:29092, I honestly just take it from here https://github.com/bitnami/bitnami-docker-kafka/issues/29#issuecomment-435216430 It should be for the host port mapping

Ah, okay, I missed the configuration for PLAINTEXT_HOST.
though, there is no need for it to be port 29092. Usualy it is set to 9092 as well.

Ok, now the docker-compose.yml is:

version: '2'

services:
  zookeeper:
    image: 'bitnami/zookeeper:3'
    ports:
      - '2181:2181'
    volumes:
      - 'zookeeper_data:/bitnami'
    environment:
      - ALLOW_ANONYMOUS_LOGIN=yes
  kafka:
    image: 'bitnami/kafka:2'
    ports:
      - '9092:9092'
    volumes:
      - 'kafka_data:/bitnami'
    environment:
      - KAFKA_BROKER_ID=1
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,PLAINTEXT_HOST://:9092
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:9092
      - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
      - KAFKA_SERVER_PORT=9092
      - ALLOW_PLAINTEXT_LISTENER=yes
    depends_on:
      - zookeeper

volumes:
  zookeeper_data:
    driver: local
  kafka_data:
    driver: local

And obviously I changed the port used in the app to 9092.

But still, I have the same problem as before.