Kafka in Docker not working
DockerApache KafkaDocker ComposeKafka PythonDocker Problem Overview
I am trying to use wurstmeister\kafka-docker image with docker-compose, but I am having real problems with connecting everything.
All the posts or questions that I check, seems not to have any problems, but I am frankly lost. (And there are at least two questions in SO that try to address the problem)
I believe that the problem is my poor understanding of the networking of docker. So the problem:
I can consume and produce from the same container of kafka, but, when I try to create another container (or use my laptop with a python client) I got several errors related to the advertised.host.name parameter (in the image this parameter is KAFKA_ADVERTISED_HOST_NAME)
I already try setting this variable in lot of ways, but it simply don't work.
So I am looking for a authorative answer (i.e. how to set automatically those parameters and what is it meaning) how to set the docker-compose.yml
This is mine:
zookeeper:
image: wurstmeister/zookeeper
ports:
- "2181:2181"
kafka:
image: wurstmeister/kafka
# hostname: kafka
ports:
- "9092"
links:
- zookeeper:zk
environment:
KAFKA_ADVERTISED_HOST_NAME: "kafka"
KAFKA_ADVERTISED_PORT: "9092"
KAFKA_ZOOKEEPER_CONNECT: "zk:2181"
UPDATE
Following the advise of @dnephin, I modified the start-kafka.sh in the following lines:
...
if [[ -z "$KAFKA_ADVERTISED_PORT" ]]; then
export KAFKA_ADVERTISED_PORT=$(hostname -i)
fi
...
and remove KAFKA_ADVERTISED_HOST_NAME: "kafka" from the docker-compose.yml
I started the containers in the canonical way:
docker-compose up -d
Both of the containers are running:
$ docker-compose ps
Name Command State Ports
-----------------------------------------------------------------------------------------------------------------
infraestructura_kafka_1 start-kafka.sh Up 0.0.0.0:32768->9092/tcp
infraestructura_zookeeper_1 /opt/zookeeper/bin/zkServe ... Up 0.0.0.0:2181->2181/tcp, 2888/tcp, 3888/tcp
Afterwards I did:
docker-compose logs
And everything run smoothly.
For checking the ip addresses:
$ KAFKA_IP=$(docker inspect --format '{{ .NetworkSettings.IPAddress }}' infraestructura_kafka_1)
$ echo $KAFKA_IP
172.17.0.4
and
$ ZK_IP=$(docker inspect --format '{{ .NetworkSettings.IPAddress }}' infraestructura_zookeeper_1)
$ echo $ZK_IP
172.17.0.3
Then I execute in two differents consoles:
A producer:
$ docker run --rm --interactive wurstmeister/kafka /opt/kafka_2.11-0.9.0.1/bin/kafka-console-producer.sh --topic grillo --broker-list 171.17.0.4:9092
A consumer:
$ docker run --rm --interactive wurstmeister/kafka /opt/kafka_2.11-0.9.0.1/bin/kafka-console-consumer.sh --topic grillo --from-beginning --zookeeper 172.17.0.3:2181
Almost immediately, warnings start flying all over the screen:
[2016-03-11 00:39:17,010] WARN Fetching topic metadata with correlation id 0 for topics [Set(grillo)] from broker [BrokerEndPoint(1001,ba53d4fd7595,9092)] failed (kafka.client.ClientUtils$)
java.nio.channels.ClosedChannelException
at kafka.network.BlockingChannel.send(BlockingChannel.scala:110)
at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:75)
at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:74)
at kafka.producer.SyncProducer.send(SyncProducer.scala:119)
at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:59)
at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:94)
at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:66)
at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63)
[2016-03-11 00:39:17,013] WARN [console-consumer-79688_9dd5f575d557-1457656747003-f1ed369d-leader-finder-thread], Failed to find leader for Set([grillo,0]) (kafka.consumer.ConsumerFetcherManager$LeaderFin
derThread)
kafka.common.KafkaException: fetching topic metadata for topics [Set(grillo)] from broker [ArrayBuffer(BrokerEndPoint(1001,ba53d4fd7595,9092))] failed
at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:73)
at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:94)
at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:66)
at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63)
Caused by: java.nio.channels.ClosedChannelException
at kafka.network.BlockingChannel.send(BlockingChannel.scala:110)
at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:75)
at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:74)
at kafka.producer.SyncProducer.send(SyncProducer.scala:119)
at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:59)
... 3 more
and so on
In the console of the producer, I wrote some sentences:
$ docker run --rm --interactive klustera/kafka /opt/kafka_2.11-0.9.0.1/bin/kafka-console-producer.sh --topic grillo --broker-list 171.17.0.4:9092
Hola
¿Cómo estáń?
¿Todo bien?
And a few moments later, I got this response:
[2016-03-11 00:39:28,955] ERROR Error when sending message to topic grillo with key: null, value: 4 bytes with error: Failed to update metadata after 60000 ms. (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
[2016-03-11 00:40:28,956] ERROR Error when sending message to topic grillo with key: null, value: 16 bytes with error: Failed to update metadata after 60000 ms. (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
[2016-03-11 00:41:28,956] ERROR Error when sending message to topic grillo with key: null, value: 12 bytes with error: Failed to update metadata after 60000 ms. (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
And in the docker-compose logs
...
zookeeper_1 | 2016-03-11 00:39:07,072 [myid:] - INFO [ProcessThread(sid:0 cport:2181)::PrepRequestProcessor@651] - Got user-level KeeperException when processing sessionid:0x153631368b1000b type:create c
xid:0x2 zxid:0x47 txntype:-1 reqpath:n/a Error Path:/consumers Error:KeeperErrorCode = NodeExists for /consumers
zookeeper_1 | 2016-03-11 00:39:07,243 [myid:] - INFO [ProcessThread(sid:0 cport:2181)::PrepRequestProcessor@651] - Got user-level KeeperException when processing sessionid:0x153631368b1000b type:create c
xid:0x19 zxid:0x4b txntype:-1 reqpath:n/a Error Path:/consumers/console-consumer-79688/owners/grillo Error:KeeperErrorCode = NoNode for /consumers/console-consumer-79688/owners/grillo
zookeeper_1 | 2016-03-11 00:39:07,247 [myid:] - INFO [ProcessThread(sid:0 cport:2181)::PrepRequestProcessor@651] - Got user-level KeeperException when processing sessionid:0x153631368b1000b type:create $xid:0x1a zxid:0x4c txntype:-1 reqpath:n/a Error Path:/consumers/console-consumer-79688/owners Error:KeeperErrorCode = NoNode for /consumers/console-consumer-79688/owners
...
UPDATE 2
I made it work, at least, in docker-machine:
First, I defined a variable with the name of the docker-machine:
DOCKER_VM=kafka_test
Then, I modify the docker-compose.yml as follows:
KAFKA_ADVERTISED_HOST_NAME: "${DOCKER_MACHINE_IP}"
Lastly, in the environment of the docker-machine, I execute:
DOCKER_MACHINE_IP=$(docker-machine ip $DOCKER_VM) docker-compose up -d
But in the laptop (I mean, without using a virtual machine, it doesn't work)
Docker Solutions
Solution 1 - Docker
My solution to this issue is slightly different. I configure Kafka to advertise on kafka host and, because it's exposed on the host machine on localhost:9092, I add an entry in /etc/hosts for kafka to resolve to localhost. By doing this Kafka can be accessed from both other Docker containers and from localhost.
docker-compose.yml:
my-web-service:
build: ./my-web-service
ports:
- "8000:8000"
links:
- kafka
kafka:
image: "wurstmeister/kafka:0.10.2.0"
ports:
- "9092:9092"
hostname: kafka
links:
- zookeeper
environment:
- KAFKA_ADVERTISED_HOST_NAME=kafka
- KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
- KAFKA_ADVERTISED_PORT=9092
zookeeper:
image: wurstmeister/zookeeper
ports:
- "2181:2181"
Updated hosts file:
more /etc/hosts
127.0.0.1 localhost kafka
Solution 2 - Docker
For developing app in localhost, there is a solution in the documentation: "HOSTNAME_COMMAND"
kafka:
image: wurstmeister/kafka
ports:
- 9092:9092
environment:
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
HOSTNAME_COMMAND: "route -n | awk '/UG[ \t]/{print $$2}'"
Hope this help others...
Solution 3 - Docker
I believe the value you use for KAFKA_ADVERTISED_HOST_NAME will change depending on how the container can be reached.
If you're trying to connect from another container, using kafka should be correct (as long as you use set that name as the link alias).
If you're trying to connect from the host, that name isn't going to work. You'd need to use the container IP address, which you can get using docker inspect. However the container IP address will change, so it might be better to set this from inside the container using $(hostname -i) to retrieve it.
Solution 4 - Docker
here's an improved version of @radek1st answer.
links is the old docker way, networks is the current method.
imo, making any sort of system change isn't good and should never be needed. it also kind of defeats the purpose of using Docker.
version: '2.1'
networks:
sb:
driver: bridge
services:
zookeeper:
image: confluentinc/cp-zookeeper:latest
container_name: zookeeper
hostname: zookeeper
networks:
- sb
ports:
- "2181:2181"
environment:
ZOOKEEPER_CLIENT_PORT: 2181
ZOOKEEPER_TICK_TIME: 2000
kafka:
image: confluentinc/cp-kafka:latest
container_name: kafka
hostname: ${KAFKA_HOSTNAME:-kafka}
depends_on:
- zookeeper
networks:
- sb
ports:
- "9092:9092"
environment:
KAFKA_BROKER_ID: 1
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_ADVERTISED_HOST_NAME: ${KAFKA_HOSTNAME:-kafka}
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://${KAFKA_HOSTNAME:-kafka}:9092
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
Then I use the follow bash script to start things. This allows me to override the Kafka hostname for local development. ./startup.sh localhost
#!/bin/bash
echo KAFKA_HOSTNAME=${1:-kafka} > .env
docker-compose up -d
Read more -
Solution 5 - Docker
Just try the following & use service discovery for example this one.
zookeeper:
image: wurstmeister/zookeeper
ports:
- "2181:2181"
kafka:
build: .
ports:
- "9092:9092"
links:
- zookeeper:zk
environment:
KAFKA_ADVERTISED_HOST_NAME: 192.168.59.103
KAFKA_ADVERTISED_PORT: 9092
KAFKA_CREATE_TOPICS: "test:1:1"
volumes:
- /var/run/docker.sock:/var/run/docker.sock
Or you use this one:
zookeeper:
image: wurstmeister/zookeeper
ports:
- "2181"
kafka:
build: .
ports:
- "9092"
links:
- zookeeper:zk
environment:
KAFKA_ADVERTISED_HOST_NAME: 192.168.59.103
DOCKER_HOST: 192.168.59.103:2375
volumes:
- /var/run/docker.sock:/var/run/docker.sock
Solution 6 - Docker
Not direct answer, but if anyone is trying to understand the kafka docker networking wurstmeister\kafka-docker author has written an amazing wiki write up on network connectivity.
It explains the three main requirements for configuring Kafka networking through docker-compose.
- Each Broker must be able to talk to Zookeeper - for leader election etc.
- Each Broker must be able to talk to every other Broker - for replication etc.
- Each Consumer/Producer must be able to talk to every Broker - for reading/writing data etc.
Few Gotcha's:
> Because it is only possible to bind to each unique port once on a > single interface, we can no longer publish the Broker port (9092). > Instead, we simply expose the port. > > ports: > - "9092"
> Since Kafka 0.9.0 - it has been possible to specify multiple ports for > listening on. This is to facilitate support for multiple protocols > (i.e. PLAINTEXT,SASL,SSL etc) and separate internal and external > traffic. With this change, host.name and port have been deprecated in > favour of listeners. advertised.host.name and advertised.port have > been deprecated in favour of advertised.listeners.
Solution 7 - Docker
I resolve this issue use below code：
zookeeper:
image: wurstmeister/zookeeper
ports:
- "2181:2181"
kafka:
image: wurstmeister/kafka
ports:
- "9092:9092"
depends_on:
- zookeeper
environment:
HOSTNAME_COMMAND: "ifconfig eth0 | grep 'inet addr' | awk '{ print $$2}' | awk -F: '{print $$2}''"
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
volumes:
- /var/run/docker.sock:/var/run/docker.sock
Solution 8 - Docker
Personally, I had the issue because the KAFKA_ADVERTISED_PORT: "9092" was missing in the kafka environment.
kafka:
image : wurstmeister/kafka
ports:
- "9092:9092"
environment:
KAFKA_ADVERTISED_HOST_NAME: 127.0.0.1
KAFKA_ADVERTISED_PORT: "9092"
KAFKA_CREATE_TOPICS: "test:1:1"
KAFKA_ZOOKEEPER_CONNECT: zookeeper
Solution 9 - Docker
In my case I forgot to update the docker-compose.yml environment configuration for kafka
Previous has localhost
environment:
- KAFKA_CFG_ADVERTISED_LISTENERS=INTERNAL://kafka:29092,EXTERNAL://localhost:9092
Updated replaces localhost with kafka:
environment:
- KAFKA_CFG_ADVERTISED_LISTENERS=INTERNAL://kafka:29092,EXTERNAL://kafka:9092
full:
networks:
kafka-net:
driver: bridge
volumes:
kafka:
zookeeper_data:
zookeeper_txns:
services:
kafka:
image: "bitnami/kafka:2.7.0"
networks:
- kafka-net
ports:
- "9092:9092"
- "29092:29092"
environment:
- KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
- KAFKA_CFG_LISTENERS=INTERNAL://kafka:29092,EXTERNAL://kafka:9092
- KAFKA_CFG_ADVERTISED_LISTENERS=INTERNAL://kafka:29092,EXTERNAL://kafka:9092
- KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
- KAFKA_CFG_INTER_BROKER_LISTENER_NAME=INTERNAL
- KAFKA_CFG_NUM_PARTITIONS=10
- KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=false
- ALLOW_PLAINTEXT_LISTENER=yes
# 10MB max message size (vs 1MB default)
- KAFKA_CFG_MESSAGE_MAX_BYTES=10485760
- KAFKA_CFG_REPLICA_FETCH_MAX_BYTES=10485760
- KAFKA_CFG_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1
- KAFKA_CFG_TRANSACTION_STATE_LOG_MIN_ISR=1
depends_on:
- zookeeper
zookeeper:
image: "bitnami/zookeeper:3.6.2"
networks:
- kafka-net
ports:
- "2181:2181"
environment:
- ALLOW_ANONYMOUS_LOGIN=yes
Solution 10 - Docker
This is in regards to kafka bailing on the following error type:
> [2022-05-27 15:53:41,940] ERROR Error while creating ephemeral at /brokers/ids/1, node already exists and owner '72062330391560192' does not match current session '72058027900010496' (kafka.zk.KafkaZkClient$CheckedEphemeral)
[2022-05-27 15:53:41,944] ERROR [KafkaServer id=1] Fatal error during KafkaServer startup. Prepare to shutdown (kafka.server.KafkaServer)
org.apache.zookeeper.KeeperException$NodeExistsException: KeeperErrorCode = NodeExists
at org.apache.zookeeper.KeeperException.create(KeeperException.java:126)
at kafka.zk.KafkaZkClient$CheckedEphemeral.getAfterNodeExists(KafkaZkClient.scala:1904)
at kafka.zk.KafkaZkClient$CheckedEphemeral.create(KafkaZkClient.scala:1842)
at kafka.zk.KafkaZkClient.checkedEphemeralCreate(KafkaZkClient.scala:1809)
at kafka.zk.KafkaZkClient.registerBroker(KafkaZkClient.scala:96)
at kafka.server.KafkaServer.startup(KafkaServer.scala:324)
at kafka.Kafka$.main(Kafka.scala:109)
at kafka.Kafka.main(Kafka.scala)
This is what worked for me:
identify which volumes to remove
docker volume ls
remove them
docker volume rm beepboop_zookeeper_data
docker volume rm beepboop_zookeeper_txns
restart kafka
docker-compose restart kafka
Solution 11 - Docker
I just update my host file and add:
127.0.0.1 localhost kafkaserver
It works fine for me. I used the same docker image on windows 10.