☰

Kafka 消费者工具未连接到 Docker 中的集群

tag-icon • tag-icon kafka
Kafka 消费者工具未连接到 Docker 中的集群

Kafka 消费者工具未连接到 Docker 中的集群

命令

docker-compose -f docker-compose.yaml up --force-recreate --remove-orphans

Docker Compose 命令会启动 3 个虚拟机：zookeeper、kafka、kafka-create-topics。第三个虚拟机生成一个新主题topic并将消息推送到该主题并尝试使用。前两个命令似乎可以工作并且没有发出任何错误。然而kafka-console-consumer失败了，看起来无论使用什么配置参数。

  • cub kafka-ready- 成功
  • kafka-topics- 成功
  • kafka-console-producer- 成功
  • kafka-console-consumer- 每次都失败

错误

kafka-create-topics_1  | [2019-10-07 21:03:00,556] WARN [Consumer clientId=consumer-1, groupId=console-consumer-58118] Connection to node -1 (/172.21.0.3:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)     
kafka-create-topics_1  | [2019-10-07 20:58:24,047] ERROR Error processing message, terminating consumer process:  (kafka.tools.ConsoleConsumer$)                                                         
kafka-create-topics_1  | org.apache.kafka.common.errors.TimeoutException                                                                                                                                 
kafka-create-topics_1  | Processed a total of 0 messages

两个文件

kafka-sample-generator.sh
docker-compose.yaml

Dockerfiledocker-compose.yaml

version: '2'
services:

  zookeeper:
    image: confluentinc/cp-zookeeper:5.3.1
    hostname: zookeeper
    ports:
      - '2181:2181'
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181

  kafka:
    image: confluentinc/cp-enterprise-kafka:5.3.1
    hostname: kafka
    ports:
      - '9092:9092'
    depends_on:
      - zookeeper
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false"
      KAFKA_METRIC_REPORTERS: io.confluent.metrics.reporter.ConfluentMetricsReporter
      CONFLUENT_METRICS_REPORTER_BOOTSTRAP_SERVERS: kafka:9092
      CONFLUENT_METRICS_ENABLE: 'true'
      CONFLUENT_SUPPORT_CUSTOMER_ID: 'anonymous'

  kafka-create-topics:
    image: confluentinc/cp-enterprise-kafka:5.3.1
    depends_on:
      - zookeeper
      - kafka
    hostname: kafka-create-topics
    volumes:
      - './kafka-sample-generator.sh:/bin/kafka-sample-generator.sh'
    command: "/bin/kafka-sample-generator.sh"
    environment:
      KAFKA_TOPIC: topic
      KAFKA_BROKER: kafka

示例生成器脚本kafka-sample-generator.sh

#!/bin/sh

HOST=`host $KAFKA_BROKER | awk '/has address/ { print $4 ; exit }'`

## Wait until Kafka is ready then create demo topic
echo 'Waiting for Kafka to be ready...'
cub kafka-ready -b $HOST:9092 1 20 && \
sleep 1

echo "Creating Topic [$HOST:9092 <topic:'$KAFKA_TOPIC'>]"
kafka-topics --create                   \
             --topic $KAFKA_TOPIC       \ 
             --if-not-exists            \
             --zookeeper zookeeper:2181 \
             --partitions 1             
             #--replication-factor 1
sleep 1

echo "Availalbe Topics"
kafka-topics --list --zookeeper zookeeper:2181
sleep 1

## Emit sample data stream
while true
    do echo "Sending Data [$HOST:9092 <topic:'$KAFKA_TOPIC'>]"
    for i in `seq 1 10`;
        do echo "$HOST"
        DATA="{\"data\":\"sample-data-$i\"}"
        echo "$DATA"
        kafka-console-producer       \
            --broker-list $HOST:9092 \
            --topic $KAFKA_TOPIC     \
            "$DATA"
    done
    sleep 1.0


    echo ''
    echo "Receiving Data [$HOST:9092 <topic:'$KAFKA_TOPIC'>]"

    ## FAILES HERE
    ## FAILES HERE
    ## FAILES HERE
    kafka-console-consumer                     \
        --bootstrap-server $HOST:9092          \
        --topic $KAFKA_TOPIC                  #\
        #--partition 0                         \
        #--from-beginning                      \
        #--max-messages 1                      \
        #--timeout-ms 45000                    \
        #--skip-message-on-error
done

答案1

这些是我为使其正常工作所做的更改：

  • 禁用指标（它坚持尝试创建复制因子为 3 的主题）
  • 更改了 kafka-topics create 命令以使用复制因子 1。
  • 通过 STDIN 将消息数据传递给 kafka-console- Producer
  • kafka-console-consumer 从一开始最多接收 10 条消息，因此它不会阻塞等待新消息
version: '2'
services:

  zookeeper:
    image: confluentinc/cp-zookeeper:5.3.1
    hostname: zookeeper
    ports:
      - '2181:2181'
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181

  kafka:
    image: confluentinc/cp-enterprise-kafka:5.3.1
    hostname: kafka
    ports:
      - '9092:9092'
    depends_on:
      - zookeeper
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: LISTENER_BOB://kafka:9092
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false"
      CONFLUENT_METRICS_ENABLE: 'false'
      CONFLUENT_SUPPORT_CUSTOMER_ID: 'anonymous'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: LISTENER_BOB:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: LISTENER_BOB  
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  kafka-create-topics:
    image: confluentinc/cp-enterprise-kafka:5.3.1
    depends_on:
      - zookeeper
      - kafka
    hostname: kafka-create-topics
    volumes:
      - './kafka-sample-generator.sh:/bin/kafka-sample-generator.sh'
    command: "/bin/kafka-sample-generator.sh"
    environment:
      KAFKA_TOPIC: topic
      KAFKA_BROKER: kafka

#!/bin/sh

HOST=`host $KAFKA_BROKER | awk '/has address/ { print $4 ; exit }'`

## Wait until Kafka is ready then create demo topic
echo 'Waiting for Kafka to be ready...'
cub kafka-ready -b $HOST:9092 1 20 && \
sleep 1

echo "Creating Topic [$HOST:9092 <topic:'$KAFKA_TOPIC'>]"
kafka-topics --create  --topic $KAFKA_TOPIC   --if-not-exists --zookeeper zookeeper:2181 --partitions 1  --replication-factor 1
sleep 1

echo "Availalbe Topics"
kafka-topics --list --zookeeper zookeeper:2181
sleep 1

## Emit sample data stream
while true
    do echo "Sending Data [$HOST:9092 <topic:'$KAFKA_TOPIC'>]"
    for i in `seq 1 10`;
    do

        echo "$HOST"
        echo "$DATA"

        DATA="{\"data\":\"sample-data-$i\"}"

        echo "$DATA" | kafka-console-producer   \
            --broker-list $HOST:9092            \
            --topic $KAFKA_TOPIC     
    done
    sleep 1.0

    echo ''
    echo "Receiving Data [$HOST:9092 <topic:'$KAFKA_TOPIC'>]"

kafka-console-consumer              \
    --bootstrap-server $HOST:9092   \
    --topic $KAFKA_TOPIC            \
    --from-beginning                \
    --max-messages 10

done

相关内容