How can I get my Kafka brokers to communicate with each other?

Question

I'm trying to create two kafka docker images to communicate with each other but the first kafka broker to stand up cannot locate the second kafka broker. Here is my docker-compose.yml:

version: '3.8'
services:
  zookeeper:
    image: "bitnami/zookeeper:latest"
    ports:
      - "2181:2181"
    environment:
      - ALLOW_ANONYMOUS_LOGIN=yes
      - ZOOKEEPER_CLIENT_PORT=2181
      - ZOOKEEPER_TICK_TIME=2000
    tmpfs: "/datalog"
  kafka-1:
    image: "bitnami/kafka:latest"
    container_name: "kafka-1"
    ports:
      - "9094:9094"
    depends_on:
      - zookeeper
    restart: always
    environment:
      - KAFKA_BROKER_ID=1
      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
      - KAFKA_LISTENERS=PLAINTEXT://:9094
      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://127.0.0.1:9094
      - KAFKA_DEFAULT_REPLICATION_FACTOR=2
      - KAFKA_NUM_PARTITIONS=2
      - ALLOW_PLAINTEXT_LISTENER=yes
  kafka-2:
    image: "bitnami/kafka:latest"
    container_name: "kafka-2"
    ports:
      - "9095:9095"
    depends_on:
      - zookeeper
    restart: always
    environment:
      - KAFKA_BROKER_ID=2
      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
      - KAFKA_LISTENERS=PLAINTEXT://:9095
      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://127.0.0.1:9095
      - KAFKA_DEFAULT_REPLICATION_FACTOR=2
      - KAFKA_NUM_PARTITIONS=2
      - ALLOW_PLAINTEXT_LISTENER=yes

the kafka container repeatedly prints out:

kafka-2                      | [[2022-10-24 15:41:47,131] INFO [Controller id=2, targetBrokerId=1] Client requested connection close from node 1 (org.apache.kafka.clients.NetworkClient)
kafka-2                      | [2022-10-24 15:41:47,232] INFO [Controller id=2, targetBrokerId=1] Node 1 disconnected. (org.apache.kafka.clients.NetworkClient)
kafka-2                      | [2022-10-24 15:41:47,232] WARN [Controller id=2, targetBrokerId=1] Connection to node 1 (/127.0.0.1:9094) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
kafka-2                      | [2022-10-24 15:41:47,232] WARN [RequestSendThread controllerId=2] Controller 2's connection to broker 127.0.0.1:9094 (id: 1 rack: null) was unsuccessful (kafka.controller.RequestSendThread)
kafka-2                      | java.io.IOException: Connection to 127.0.0.1:9094 (id: 1 rack: null) failed.
kafka-2                      |  at org.apache.kafka.clients.NetworkClientUtils.awaitReady(NetworkClientUtils.java:70)
kafka-2                      |  at kafka.controller.RequestSendThread.brokerReady(ControllerChannelManager.scala:291)
kafka-2                      |  at kafka.controller.RequestSendThread.doWork(ControllerChannelManager.scala:245)
kafka-2                      |  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:96)

when I try and use kafkajs to send a message I get the error message: KafkaJSNonRetriableError: Replication-factor is invalid

EDIT:

Using the suggestion below:

version: '3.8'
services:
  dynamodb-local-express-bp:
    command: "-jar DynamoDBLocal.jar -sharedDb -dbPath ./data"
    image: "amazon/dynamodb-local:latest"
    container_name: dynamodb-local-express-bp
    ports:
      - "8000:8000"
    volumes:
      - "./docker/dynamodb:/home/dynamodblocal/data"
    working_dir: /home/dynamodblocal
  zookeeper:
    image: "bitnami/zookeeper:latest"
    ports:
      - "2181:2181"
    environment:
      - ALLOW_ANONYMOUS_LOGIN=yes
      - ZOOKEEPER_CLIENT_PORT=2181
      - ZOOKEEPER_TICK_TIME=2000
    tmpfs: "/datalog"
  kafka-1:
    image: "bitnami/kafka:latest"
    container_name: "kafka-1"
    expose:
      - 9092
    ports:
      - "9094:9094"
    depends_on:
      - zookeeper
    restart: always
    environment:
      - KAFKA_BROKER_ID=1
      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9094,INTERNAL://:9092
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://127.0.0.1:9094,INTERNAL://kafka-1:9092
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXT:PLAINTEXT,INTERNAL:PLAINTEXT
      - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=INTERNAL
      - KAFKA_CFG_DEFAULT_REPLICATION_FACTOR=2
      - KAFKA_CFG_NUM_PARTITIONS=2
      - ALLOW_PLAINTEXT_LISTENER=yes
  kafka-2:
    image: "bitnami/kafka:latest"
    container_name: "kafka-2"
    ports:
      - "9095:9095"
    depends_on:
      - zookeeper
    restart: always
    environment:
      - KAFKA_BROKER_ID=2
      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9095,INTERNAL://:9092
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://127.0.0.1:9095,INTERNAL://kafka-1:9092
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXT:PLAINTEXT,INTERNAL:PLAINTEXT
      - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=INTERNAL
      - KAFKA_CFG_DEFAULT_REPLICATION_FACTOR=2
      - KAFKA_CFG_NUM_PARTITIONS=2
      - ALLOW_PLAINTEXT_LISTENER=yes
      # - KAFKA_BROKER_ID=2
      # - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
      # - KAFKA_LISTENERS=PLAINTEXT://:9095
      # - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://127.0.0.1:9095
      # - KAFKA_DEFAULT_REPLICATION_FACTOR=2
      # - KAFKA_NUM_PARTITIONS=2
      # - ALLOW_PLAINTEXT_LISTENER=yes

i get the following error message:

[2022-10-24 18:59:06,286] ERROR [KafkaServer id=2] Fatal error during KafkaServer startup. Prepare to shutdown (kafka.server.KafkaServer)
kafka-2                      | java.lang.IllegalArgumentException: requirement failed: Configured end points kafka-1:9092 in advertised listeners are already registered by broker 1
kafka-2                      |  at scala.Predef$.require(Predef.scala:281)
kafka-2                      |  at kafka.server.KafkaServer.$anonfun$createBrokerInfo$3(KafkaServer.scala:512)
kafka-2                      |  at kafka.server.KafkaServer.$anonfun$createBrokerInfo$3$adapted(KafkaServer.scala:510)
kafka-2                      |  at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)
kafka-2                      |  at scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)
kafka-2                      |  at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49)
kafka-2                      |  at kafka.server.KafkaServer.createBrokerInfo(KafkaServer.scala:510)
kafka-2                      |  at kafka.server.KafkaServer.startup(KafkaServer.scala:331)
kafka-2                      |  at kafka.Kafka$.main(Kafka.scala:109)
kafka-2                      |  at kafka.Kafka.main(Kafka.scala)

BTW, 2 brokers on one host is still a single point of failure, and you are needlessly duplicating records on the same hard-drive with replication factor of more than 1... — OneCricketeer, Oct 24 '22 at 17:57

maow · Accepted Answer · 2022-10-24T16:44:31.680

The problem is caused by the bridge network docker creates for you.

Here is an excellent article what happens and how exactly the Kafka brokers discover each other.

You need an additional advertised listener. 127.0.0.1 is the one that is advertised to your host machine. You need one in additional for inside the docker network.

Add/change these configs to your dockerfile:

kafka-1:
   ...
   expose: 
      - 9092
   environment:
      - KAFKA_BROKER_ID=1
      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9094,INTERNAL://:9092
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://127.0.0.1:9094,INTERNAL://kafka-1:9092
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXT:PLAINTEXT,INTERNAL:PLAINTEXT
      - KAFKA_CFG_INTER_BROKER_LISTENER_NAME: INTERNAL
      - KAFKA_CFG_DEFAULT_REPLICATION_FACTOR=2
      - KAFKA_CFG_NUM_PARTITIONS=2
      - ALLOW_PLAINTEXT_LISTENER=yes

and analogeously for kafka-2

Additionally you are using the bitnami image and they prefix all Kafka environment variables with KAFKA_CFG. Thats also a gotcha

didn't work. Both kafka instances throw the error in my edited question. — blu, Oct 24 '22 at 19:01
You need to replace `INTERNAL://kafka-1:9092` by `INTERNAL://kafka-2:9092` for broker 2. Eachbroker must advertise itself. — maow, Oct 24 '22 at 19:21

How can I get my Kafka brokers to communicate with each other?

1 Answers1