Kafka Consumer is not receiving message

Question

I have this in my docker-compose.yml

version '3'

services:
  app:
    build:
      context: .
      dockerfile: app/Dockerfile
    ports:
      - 8000:8000
    environment:
      DB_NAME: postgres
      DB_USER: postgres
      DB_PASS: postgres
      DB_HOST: db
      DB_PORT: 5432
      SECRET_KEY: my_secret_key
      DEBUG: 'True'
    depends_on:
      - db
  db:
    image: postgres:11.1-alpine
    environment:
      DB_NAME: postgres
      DB_USER: postgres
      DB_PASS: postgres
      DB_PORT: 5432
    ports:
      - 5432:5432
    volumes:
      - db:/var/lib/postgresql/data
    restart: always
  zookeeper:
    image: wurstmeister/zookeeper:latest
    ports:
      - "2181:2181"
  kafka:
    image: wurstmeister/kafka:latest
    ports:
      - "9092:9092"
    expose:
      - "9093"
    environment:
      KAFKA_ADVERTISED_HOST_NAME: kafka
      KAFKA_ADVERTISED_LISTENERS: INSIDE://kafka:9093,OUTSIDE://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INSIDE:PLAINTEXT,OUTSIDE:PLAINTEXT
      KAFKA_LISTENERS: INSIDE://0.0.0.0:9093,OUTSIDE://0.0.0.0:9092
      KAFKA_INTER_BROKER_LISTENER_NAME: INSIDE
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_CREATE_TOPICS: "msg_topic:1:1"
    volumes:
     - /var/run/docker.sock:/var/run/docker.sock
  listener:
    build:
      context: .
      dockerfile: listener/Dockerfile
    depends_on:
      - kafka
      - zookeeper
      - app

volumes:
  db:

this is the producer

def prod(msg):
    producer = KafkaProducer(bootstrap_servers='kafka:9092')
    data = {
        'msg': msg.text
    }
    serialized_data = pickle.dumps(data, pickle.HIGHEST_PROTOCOL)
    producer.send('msg_topic', serialized_data)
    print('message was sent')
    return HttpResponse(200)

and consumer

def cons():
    consumer = KafkaConsumer('msg_topic',
                             bootstrap_servers=['kafka:9092'])
    for message in consumer:
        deserialized_data = pickle.loads(message.value)
        result = deserialized_data['msg']

all images are running and when I try send message from producer, consumer doesn't receive it. What's the problem? I spent a lot of time on this, can anyone help please? Could this be a port issue? To be honest, I don't understand what ports should i use

Try to change port in consumer method to 9093 and make network to be able to listen to kafka. — Abdelrahman Elayashy, Apr 15 '22 at 16:22
how can i watch it? And how to make network to be able to listen to kafka? — Timur, Apr 15 '22 at 16:37
The below answer is correct, however, your Python code needs to wait for the Kafka server to actually start — OneCricketeer, Apr 15 '22 at 17:57

score 1 · Answer 1 · edited Apr 15 '22 at 21:46

Could this be a port issue? To be honest, I don't understand what ports should i use

let us fix ports. We define network to let containers that they can listen to each other or send to each other.

version '3'

services:
  app:
    build:
      context: .
      dockerfile: app/Dockerfile
    ports:
      - 8000:8000
    networks:
      - spring
    environment:
      DB_NAME: postgres
      DB_USER: postgres
      DB_PASS: postgres
      DB_HOST: db
      DB_PORT: 5432
      SECRET_KEY: my_secret_key
      DEBUG: 'True'
    depends_on:
      - db
  db:
    image: postgres:11.1-alpine
    environment:
      DB_NAME: postgres
      DB_USER: postgres
      DB_PASS: postgres
      DB_PORT: 5432
    ports:
      - 5432:5432
    volumes:
      - db:/var/lib/postgresql/data
    restart: always
  zookeeper:
    image: wurstmeister/zookeeper:latest
    ports:
      - "2181:2181" 
    network:
      - spring
  kafka:
    image: wurstmeister/kafka:latest
    networks:
      - spring
 
    environment:
      KAFKA_ADVERTISED_HOST_NAME: kafka
      KAFKA_ADVERTISED_LISTENERS: INSIDE://kafka:9093,OUTSIDE://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INSIDE:PLAINTEXT,OUTSIDE:PLAINTEXT
      KAFKA_LISTENERS: INSIDE://0.0.0.0:9093,OUTSIDE://0.0.0.0:9092
      KAFKA_INTER_BROKER_LISTENER_NAME: INSIDE
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_CREATE_TOPICS: "msg_topic:1:1"
    volumes:
     - /var/run/docker.sock:/var/run/docker.sock
  listener:
    build:
      context: .
      dockerfile: listener/Dockerfile
    networks:
      - spring
    depends_on:
      - kafka
      - zookeeper
      - app

volumes:
  db:

networks:
  spring:
    driver: bridge

what I have changed in composer is that I create a network and only let Kafka expose port in that network so that other can send through it or listen from it.

then change kafka ports in producer and consumer to 9093 producer

def prod(msg):
    producer = KafkaProducer(bootstrap_servers='kafka:9093')
    data = {
        'msg': msg.text
    }
    serialized_data = pickle.dumps(data, pickle.HIGHEST_PROTOCOL)
    producer.send('msg_topic', serialized_data)
    print('message was sent')
    return HttpResponse(200)

Listener

def cons():
    consumer = KafkaConsumer('msg_topic',
                             bootstrap_servers=['kafka:9093'])
    for message in consumer:
        deserialized_data = pickle.loads(message.value)
        result = deserialized_data['msg']

now i have an error in consumer: kafka.errors.NoBrokersAvailable: NoBrokersAvailable — Timur, Apr 15 '22 at 16:56
try this one https://stackoverflow.com/questions/38854957/nobrokersavailable-nobrokersavailable-kafka-error — Abdelrahman Elayashy, Apr 15 '22 at 17:04
it didn't work for me.. ok thanks for taking the time to help me — Timur, Apr 15 '22 at 17:20
Try to debug it. First download https://www.kafkatool.com/ to monitor kafka. See first if topic is created or not. Then see if message sent or not. And later debug consumer. Maybe the problem is not with consumer — Abdelrahman Elayashy, Apr 15 '22 at 17:23
Note: You don't need to expose the port to the host in order for containers to communicate over the bridge network. You also need to put Zookeeper on the spring network (which not clear why you added) — OneCricketeer, Apr 15 '22 at 17:58
@OneCricketeer thanks for note. I have edited the answer. What was not clear about adding what?. — Abdelrahman Elayashy, Apr 15 '22 at 21:12
There's no spring code used and the network was not part of the question. This appears to be something that you added — OneCricketeer, Apr 15 '22 at 21:45
My thought was that I did not use the default network because he has used container's name not ip in python code. So I have defined spring network. The name spring that what came to mind — Abdelrahman Elayashy, Apr 16 '22 at 06:39

Kafka Consumer is not receiving message

1 Answers1