0

Goal: I want to analyse the headers from a topic and I am looking for some straighforward way to see the header. So I don't want to develop an extra application or extense code just for that. Any straighforward tool to see the header will be usefull.

From this question I read it is possible with kafkacat that way

kafkacat -b kafka-broker:9092 -t my_topic_name -C \
  -f '\nKey (%K bytes): %k
  Value (%S bytes): %s
  Timestamp: %T
  Partition: %p
  Offset: %o
  Headers: %h\n'

So I desire to start kafkacat as part of my docker-compose and my first try is

version: '3'

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:5.4.0
    hostname: zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  broker:
    image: confluentinc/cp-server:5.4.0
    hostname: broker
    container_name: broker
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_METRIC_REPORTERS: io.confluent.metrics.reporter.ConfluentMetricsReporter
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_CONFLUENT_LICENSE_TOPIC_REPLICATION_FACTOR: 1
      CONFLUENT_METRICS_REPORTER_BOOTSTRAP_SERVERS: broker:29092
      CONFLUENT_METRICS_REPORTER_ZOOKEEPER_CONNECT: zookeeper:2181
      CONFLUENT_METRICS_REPORTER_TOPIC_REPLICAS: 1
      CONFLUENT_METRICS_ENABLE: 'true'
      CONFLUENT_SUPPORT_CUSTOMER_ID: 'anonymous'

  kafka-tools:
    image: confluentinc/cp-kafka:5.4.0
    hostname: kafka
    container_name: kafka
    command: ["tail", "-f", "/dev/null"]
    network_mode: "host"

  kafkacat:
    image: confluentinc/cp-kafkacat
    command: 
      - bash 
      - -c 
    links:
      - broker

And I got

broker         | [2020-12-11 10:46:13,084] DEBUG [Controller id=1] Topics not in preferred replica for broker 1 Map() (kafka.controller.KafkaController)
broker         | [2020-12-11 10:46:13,084] TRACE [Controller id=1] Leader imbalance ratio for broker 1 is 0.0 (kafka.controller.KafkaController)
confluentinc_kafkacat_1 exited with code 2
broker         | [2020-12-11 10:51:13,085] INFO [Controller id=1] Processing automatic preferred replica leader election (kafka.controller.KafkaController)

Second tentative base on kafkacat tutorial

C:\Users\DEMETRC>docker run --tty confluentinc/cp-kafkacat kafkacat -b localhost:9092 -L
% ERROR: Failed to acquire metadata: Local: Broker transport failure

PS: my docker kafka is available in localhost:9092 without docker network. In the example from tutorial its kafka is available in kafka:29092 on the Docker network docker-compose_default

Any clue how add kafkacat to my docker-compose? Any other suggestion to investigate the message header?

*** first edit after Robin Moffatt's answer

I edited docker-compose adding Robin's suggestion

version: '3'

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:5.4.0
    hostname: zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
  broker:
    image: confluentinc/cp-server:5.4.0
    hostname: broker
    container_name: broker
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_METRIC_REPORTERS: io.confluent.metrics.reporter.ConfluentMetricsReporter
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_CONFLUENT_LICENSE_TOPIC_REPLICATION_FACTOR: 1
      CONFLUENT_METRICS_REPORTER_BOOTSTRAP_SERVERS: broker:29092
      CONFLUENT_METRICS_REPORTER_ZOOKEEPER_CONNECT: zookeeper:2181
      CONFLUENT_METRICS_REPORTER_TOPIC_REPLICAS: 1
      CONFLUENT_METRICS_ENABLE: 'true'
      CONFLUENT_SUPPORT_CUSTOMER_ID: 'anonymous'

  kafka-tools:
    image: confluentinc/cp-kafka:5.4.0
    hostname: kafka
    container_name: kafka
    command: ["tail", "-f", "/dev/null"]
    network_mode: "host"

  kafkacat:
    image: edenhill/kafkacat:1.6.0
    container_name: kafkacat
    links:
      - broker
    entrypoint: 
      - /bin/sh 
      - -c 
      - |
        apk add jq; 
        while [ 1 -eq 1 ];do sleep 60;done

Here is my topic description

# kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic demotopic
Created topic demotopic.
# kafka-topics --describe --bootstrap-server localhost:9092 --topic demotopic
Topic: demotopic        PartitionCount: 1       ReplicationFactor: 1    Configs:
        Topic: demotopic        Partition: 0    Leader: 1       Replicas: 1     Isr: 1
#

And the error I am getting trying to use kafkacat

/ # kafkacat -b localhost:9092 -t demotopic -C -f '\nKey (%K bytes): %k Value (%S bytes): %s Timestamp: %T Partition: %p
 Offset: %o Headers: %h\n'
%3|1607709622.155|FAIL|rdkafka#consumer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#127.0.0.1:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1607709623.155|FAIL|rdkafka#consumer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#127.0.0.1:9092 failed: Connection refused (after 0ms in state CONNECT, 1 identical error(s) suppressed)
% ERROR: Failed to query metadata for topic demotopic: Local: Broker transport failure
/ #

*** Second edit

Now I can use kafkacat for listing all topics but I am still blocked to get the message headers

/ #  kafkacat -L -b broker:9092
Metadata for all topics (from broker -1: broker:9092/bootstrap):
 1 brokers:
  broker 1 at localhost:9092 (controller)
 4 topics:
  topic "demotopic" with 1 partitions:
    partition 0, leader 1, replicas: 1, isrs: 1
  topic "_confluent-license" with 1 partitions:
/ # kafkacat -L -b broker:9092
Metadata for all topics (from broker -1: broker:9092/bootstrap):
 1 brokers:
  broker 1 at localhost:9092 (controller)
 4 topics:
  topic "demotopic" with 1 partitions:
    partition 0, leader 1, replicas: 1, isrs: 1
  topic "_confluent-license" with 1 partitions:
    partition 0, leader 1, replicas: 1, isrs: 1
  topic "_confluent-metrics" with 12 partitions:
    partition 0, leader 1, replicas: 1, isrs: 1
    partition 1, leader 1, replicas: 1, isrs: 1
    partition 2, leader 1, replicas: 1, isrs: 1
    partition 3, leader 1, replicas: 1, isrs: 1
    partition 4, leader 1, replicas: 1, isrs: 1
    partition 5, leader 1, replicas: 1, isrs: 1
    partition 6, leader 1, replicas: 1, isrs: 1
    partition 7, leader 1, replicas: 1, isrs: 1
    partition 8, leader 1, replicas: 1, isrs: 1
    partition 9, leader 1, replicas: 1, isrs: 1
    partition 10, leader 1, replicas: 1, isrs: 1
    partition 11, leader 1, replicas: 1, isrs: 1
  topic "__confluent.support.metrics" with 1 partitions:
    partition 0, leader 1, replicas: 1, isrs: 1
/ # kafkacat -b broker:9092 -t demotopic -C -f '\nKey (%K bytes): %k Value (%S bytes): %s Timestamp: %T Partition: %p Of
fset: %o Headers: %h\n'
%3|1607719862.133|FAIL|rdkafka#consumer-1| [thrd:localhost:9092/1]: localhost:9092/1: Connect to ipv4#127.0.0.1:9092 failed: Connection refused (after 0ms in state CONNECT)
% ERROR: Local: Broker transport failure: localhost:9092/1: Connect to ipv4#127.0.0.1:9092 failed: Connection refused (after 0ms in state CONNECT)

Any thing wrong with this kafkacat script?

kafkacat -b broker:9092 -t demotopic -C -f '\nKey (%K bytes): %k Value (%S bytes): %s Timestamp: %T Partition: %p Offset: %o Headers: %h\n'
Jim C
  • 3,957
  • 25
  • 85
  • 162
  • 1
    Why are you trying to add kafkacat to docker-compose? This is overcomplicated. First start kafka broker, then produce some messages and last consume them via kafkacat. For latter I propose a simple local installation. Also, instead using kafkacat you could simply use the confluent UI which is part of the confluent stack: https://www.confluent.de/product/confluent-platform/gui-driven-management-and-monitoring/ – Aydin K. Dec 11 '20 at 11:28
  • @AydinK. this is my second tentative failing with % ERROR: Failed to acquire metadata: Local: Broker transport failure (I added above). Do you know what I am missing? – Jim C Dec 11 '20 at 11:29
  • 1
    You are still running kafkacat as a docker container there. Forget docker for kafkacat, it adds unnecessary overhead as you see with networking. Use a local kafkacat instead – Aydin K. Dec 11 '20 at 11:31
  • I can't install anything in my Company Windows laptop with very restrictive permissions to me. Do you know if it is possible with kafka-tools? Kafka-tools has been very helpful in my case but I couldn't figure out how use it to analyse message headers. – Jim C Dec 11 '20 at 11:53

1 Answers1

1

confluentinc_kafkacat_1 exited with code 2 is because the container exited, since you overrode the startup command to just launch bash; which it does, and then exits

   command: 
      - bash 
      - -c

This file shows a working example of what you need, in which the container will stay running

 entrypoint: 
      - /bin/sh 
      - -c 
      - |
        apk add jq; 
        while [ 1 -eq 1 ];do sleep 60;done

To Aydin's point that you don't need kafkacat in Docker - you can do it locally, but I find it easier to include in the Docker Compose so that you're not dependent on a local install.

If you want to run kafkacat outside of Docker Compose but still in Docker with docker run you can, but bear in mind the networking implications. If you try to use localhost then that's relative to the container itself, i.e. where kafkacat is running. Instead you need to make the Kafka broker accessible to the kafkacat container, e.g. by adding to the host network:

docker run --network host --interactive --rm edenhill/kafkacat:1.6.0 -b localhost:9092 -L
Robin Moffatt
  • 30,382
  • 3
  • 65
  • 92
  • 1
    You should reuse the network that compose creates by default, not host – OneCricketeer Dec 11 '20 at 15:12
  • @RobinMoffatt, thanks. I can see I am close to get the kafkacat up and running. I am getting this error now "Connect to ipv4#127.0.0.1:9092 failed: Connection refused". I added above my whole docker-compose. Any clue? – Jim C Dec 11 '20 at 18:08
  • @RobinMoffatt, kafkacat is up and running now after I replaced localhost by broker and I can connect to broker and list all topics. But when I try what I really want, get the headers, I get connection refused. Please see above the whole logs for kafkacat -b broker:9092 -t demotopic -C -f '\nKey (%K bytes): %k Value (%S bytes): %s Timestamp: %T Partition: %p Offset: %o Headers: %h\n' – Jim C Dec 11 '20 at 20:53