问无法使用kafka-python从另一个容器请求Kafka容器
EN

Stack Overflow用户

提问于 2019-06-23 22:07:14

回答 2查看 677关注 0票数 1

环境：

services:  
  zookeeper:
      image: wurstmeister/zookeeper
      ports:
        - 2181
  kafka:
    image: wurstmeister/kafka
    ports:
      - 9092:9092
      #- 8004:8004
    links:
      - zookeeper
    environment:
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_HOST_NAME: kafka
      KAFKA_ADVERTISED_PORT: 9092
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
      KAFKA_CREATE_TOPICS: "foo:10:1"
      # JMX_PORT: 8004
  clickhouse-01:
      image: yandex/clickhouse-server
      hostname: clickhouse-01
      container_name: clickhouse-01
      ports:
          - 9001:9000
      volumes:
          - ./config/config.xml:/etc/clickhouse-server/config.xml
          - ./config/metrika.xml:/etc/clickhouse-server/metrika.xml
          - ./config/macros/macros-01.xml:/etc/clickhouse-server/config.d/macros.xml

      ulimits:
          nofile:
              soft: 262144
              hard: 262144
      depends_on:
          - "zookeeper"

  clickhouse-02:
      image: yandex/clickhouse-server
      hostname: clickhouse-02
      container_name: clickhouse-02
      ports:
          - 9002:9000
      volumes:
          - ./config/config.xml:/etc/clickhouse-server/config.xml
          - ./config/metrika.xml:/etc/clickhouse-server/metrika.xml
          - ./config/macros/macros-02.xml:/etc/clickhouse-server/config.d/macros.xml

      ulimits:
          nofile:
              soft: 262144
              hard: 262144
      depends_on:
          - "zookeeper"

  clickhouse-03:
      image: yandex/clickhouse-server
      hostname: clickhouse-03
      container_name: clickhouse-03
      ports:
          - 9003:9000
      volumes:
          - ./config/config.xml:/etc/clickhouse-server/config.xml
          - ./config/metrika.xml:/etc/clickhouse-server/metrika.xml
          - ./config/macros/macros-03.xml:/etc/clickhouse-server/config.d/macros.xml

      ulimits:
          nofile:
              soft: 262144
              hard: 262144
      depends_on:
          - "zookeeper"

通过Zookeeper容器查询Kafka：

bash-4.4# /opt/kafka/bin/kafka-topics.sh --list --zookeeper zookeeper:2181
__consumer_offsets
foo
raw_trap

来自zookeeper容器的Netstat结果：

root@0a5f9a441da3:/opt/zookeeper-3.4.13# netstat
Active Internet connections (w/o servers)
Proto Recv-Q Send-Q Local Address           Foreign Address         State      
tcp        0      0 0a5f9a441da3:2181       kafka_1:58622 ESTABLISHED
tcp        0      0 0a5f9a441da3:2181       clickhouse-02.cli:60728 ESTABLISHED
tcp        0      0 0a5f9a441da3:2181       clickhouse-01.cli:56448 ESTABLISHED
tcp        0      0 0a5f9a441da3:2181       clickhouse-03.cli:39656 ESTABLISHED

从运行kafka-python的容器Telnet到broker：

root@f10fe1b58fa9:~# telnet kafka 9092
Trying 172.18.0.8...
Connected to kafka.
Escape character is '^]'.

来自telnet的Kafka错误：

kafka_1           | [2019-06-23 13:38:05,350] WARN [SocketServer brokerId=1019] Unexpected error from /172.18.0.5; closing connection (org.apache.kafka.common.network.Selector)
kafka_1           | org.apache.kafka.common.network.InvalidReceiveException: Invalid receive (size = 1903520116 larger than 104857600)
kafka_1           |     at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:104)
kafka_1           |     at org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:424)
kafka_1           |     at org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:385)
kafka_1           |     at org.apache.kafka.common.network.Selector.attemptRead(Selector.java:651)
kafka_1           |     at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:572)
kafka_1           |     at org.apache.kafka.common.network.Selector.poll(Selector.java:483)
kafka_1           |     at kafka.network.Processor.poll(SocketServer.scala:830)
kafka_1           |     at kafka.network.Processor.run(SocketServer.scala:730)
kafka_1           |     at java.lang.Thread.run(Thread.java:748)

尝试使用python向kafka topic发送数据时出错：

>>> from kafka import KafkaProducer
>>> producer = KafkaProducer(bootstrap_servers=['kafka:9092'])
>>> producer
<kafka.producer.kafka.KafkaProducer object at 0x7ff84417b320>
>>> producer.send('foo', b'raw_bytes')
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/usr/local/lib/python3.7/site-packages/kafka/producer/kafka.py", line 564, in send
    self._wait_on_metadata(topic, self.config['max_block_ms'] / 1000.0)
  File "/usr/local/lib/python3.7/site-packages/kafka/producer/kafka.py", line 691, in _wait_on_metadata
    "Failed to update metadata after %.1f secs." % (max_wait,))
kafka.errors.KafkaTimeoutError: KafkaTimeoutError: Failed to update metadata after 60.0 secs.

我已经在网上上上下下找了几次，试图找到一个解决方案。我开始确保容器中的KAFKA_ADVERTISED_HOST_NAME是正确的，并尝试更改它，但一无所获。当我更改bootstrap_servers=['kafka:9092']条目的端点时，我得到一个错误：

>>> consumer = KafkaConsumer('foo', 
...                          group_id='test-group',
...                          bootstrap_servers=['localhost:9092'])
Traceback (most recent call last):
  File "<stdin>", line 3, in <module>
  File "/usr/local/lib/python3.7/site-packages/kafka/consumer/group.py", line 353, in __init__
    self._client = KafkaClient(metrics=self._metrics, **self.config)
  File "/usr/local/lib/python3.7/site-packages/kafka/client_async.py", line 239, in __init__
    self.config['api_version'] = self.check_version(timeout=check_timeout)
  File "/usr/local/lib/python3.7/site-packages/kafka/client_async.py", line 865, in check_version
    raise Errors.NoBrokersAvailable()
kafka.errors.NoBrokersAvailable: NoBrokersAvailable

因此，看起来我可能建立了联系，但可能从根本上误解了我试图与生产者进行的请求的某些方面。

下面是我用来测试https://kafka-python.readthedocs.io/en/master/usage.html#kafkaconsumer的python库的文档和示例

编辑:我已经成功地从我们使用消费者的裸机上运行的生产kafka环境返回了消息。

python