无法从springboot应用程序连接kafka

tjvv9vkg  于 2021-06-04  发布在  Kafka
关注(0)|答案(1)|浏览(595)

我正在尝试从pc上运行的springboot java应用程序连接docker中远程主机上运行的kafka。该应用程序尝试创建kafka使用者。从下面的日志中我可以看到,应用程序正确地识别了kafka引导服务器并尝试建立连接。但由于某种原因,它开始在本地主机上寻找Kafka代理,显然失败了。你知道为什么会这样吗?当与kafka在同一个主机上运行时,应用程序工作正常。
日志说明:
第1行:在设置引导服务器之前,我从java代码中打印出来。它表明已从配置文件中正确读取配置。
第2行:使用者配置显示它将使用正确的kafka主机:bootstrap.servers=[my\u kafka_host:9092]
第3-11行:应用程序尝试连接正确的kafka主机
第14-20行:由于某种原因,应用程序试图连接localhost,我不知道为什么。
第21行:与localhost的连接失败,如预期的那样。

1  2020-03-13 14:24:34.135  INFO 9732 --- [           main] c.p.controller.KafkaConfiguration  : kafkaBootstrapServer=MY_KAFKA_HOST:9092

2  2020-03-13 14:24:36.956  INFO 9732 --- [           main] o.a.k.clients.consumer.ConsumerConfig    : ConsumerConfig values: 
    allow.auto.create.topics = true
    auto.commit.interval.ms = 5000
    auto.offset.reset = latest
    bootstrap.servers = [MY_KAFKA_HOST:9092]
    check.crcs = true
    client.dns.lookup = default

3  2020-03-13 14:24:37.299 TRACE 9732 --- [ntainer#0-0-C-1] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-1, groupId=my_project_group_id] Found least loaded node MY_KAFKA_HOST:9092 (id: -1 rack: null) with no active connection

4  [mongod output] 2020-03-13 14:24:37.842 DEBUG 9732 --- [ntainer#0-0-C-1] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-1, groupId=my_project_group_id] Initiating connection to node MY_KAFKA_HOST:9092 (id: -1 rack: null) using address /MY_KAFKA_HOST
5  2020-03-13 14:24:37.865 TRACE 9732 --- [ntainer#0-0-C-1] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-1, groupId=my_project_group_id] Found least loaded connecting node MY_KAFKA_HOST:9092 (id: -1 rack: null)
6  2020-03-13 14:24:37.868 TRACE 9732 --- [ntainer#0-0-C-1] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-1, groupId=my_project_group_id] Found least loaded connecting node MY_KAFKA_HOST:9092 (id: -1 rack: null)
7  2020-03-13 14:24:37.869 DEBUG 9732 --- [ntainer#0-0-C-1] org.apache.kafka.common.metrics.Metrics  : Added sensor with name node--1.bytes-sent
8  2020-03-13 14:24:37.870 TRACE 9732 --- [ntainer#0-0-C-1] org.apache.kafka.common.metrics.Metrics  : Registered metric named MetricName [name=outgoing-byte-total, group=consumer-node-metrics, description=The total number of outgoing bytes, tags={client-id=consumer-1, node-id=node--1}]
9  2020-03-13 14:24:37.870 TRACE 9732 --- [ntainer#0-0-C-1] org.apache.kafka.common.metrics.Metrics  : Registered metric named MetricName [name=outgoing-byte-rate, group=consumer-node-metrics, description=The number of outgoing bytes per second, tags={client-id=consumer-1, node-id=node--1}]
10 2020-03-13 14:24:37.925 TRACE 9732 --- [ntainer#0-0-C-1] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-1, groupId=my_project_group_id] Found least loaded node MY_KAFKA_HOST:9092 (id: -1 rack: null) connected with no in-flight requests
11 2020-03-13 14:24:37.932 DEBUG 9732 --- [ntainer#0-0-C-1] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-1, groupId=my_project_group_id] Sending metadata request MetadataRequestData(topics=[MetadataRequestTopic(name='my_test_topic_name')], allowAutoTopicCreation=true, includeClusterAuthorizedOperations=false, includeTopicAuthorizedOperations=false) to node MY_KAFKA_HOST:9092 (id: -1 rack: null)
12 2020-03-13 14:24:37.935 TRACE 9732 --- [ntainer#0-0-C-1] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-1, groupId=my_project_group_id] Sending METADATA {topics=[{name=my_test_topic_name}],allow_auto_topic_creation=true,include_cluster_authorized_operations=false,include_topic_authorized_operations=false} with correlation id 2 to node -1
groupId=my_project_group_id] Sending METADATA {topics=[{name=my_test_topic_name}],allow_auto_topic_creation=true,include_cluster_authorized_operations=false,include_topic_authorized_operations=false} with correlation id 2 to node -1
13 2020-03-13 14:24:37.938 TRACE 9732 --- [ntainer#0-0-C-1] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-1, groupId=my_project_group_id] Sending FIND_COORDINATOR {key=my_project_group_id,key_type=0} with correlation id 0 to node -1
14 2020-03-13 14:24:37.949 TRACE 9732 --- [ntainer#0-0-C-1] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-1, groupId=my_project_group_id] Completed receive from node -1 for METADATA with correlation id 2, received {throttle_time_ms=0,brokers=[{node_id=1,host=localhost,port=9092,rack=null}],cluster_id=kjU7kDUlSKq9UodK7-T6Wg,controller_id=1,topics=[{error_code=0,name=my_test_topic_name,is_internal=false,partitions=[{error_code=0,partition_index=0,leader_id=1,leader_epoch=0,replica_nodes=[1],isr_nodes=[1],offline_replicas=[]}],topic_authorized_operations=0}],cluster_authorized_operations=0}
2
15 2020-03-13 14:24:37.984 DEBUG 9732 --- [ntainer#0-0-C-1] org.apache.kafka.clients.Metadata        : [Consumer clientId=consumer-1, groupId=my_project_group_id] Updated cluster metadata updateVersion 2 to MetadataCache{cluster=Cluster(id = kjU7kDUlSKq9UodK7-T6Wg, nodes = [localhost:9092 (id: 1 rack: null)], partitions = [Partition(topic = my_test_topic_name, partition = 0, leader = 1, replicas = [1], isr = [1], offlineReplicas = [])], controller = localhost:9092 (id: 1 rack: null))}
16 2020-03-13 14:24:37.985 TRACE 9732 --- [ntainer#0-0-C-1] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-1, groupId=my_project_group_id] Completed receive from node -1 for FIND_COORDINATOR with correlation id 0, received {throttle_time_ms=0,error_code=0,error_message=NONE,node_id=1,host=localhost,port=9092}
17 2020-03-13 14:24:37.991 DEBUG 9732 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-1, groupId=my_project_group_id] Received FindCoordinator response ClientResponse(receivedTimeMs=1584102277985, latencyMs=163, disconnected=false, requestHeader=RequestHeader(apiKey=FIND_COORDINATOR, apiVersion=2, clientId=consumer-1, correlationId=0), responseBody=FindCoordinatorResponseData(throttleTimeMs=0, errorCode=0, errorMessage='NONE', nodeId=1, host='localhost', port=9092))
18 2020-03-13 14:24:37.992  INFO 9732 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-1, groupId=my_project_group_id] Discovered group coordinator localhost:9092 (id: 2147483646 rack: null)
19 2020-03-13 14:24:37.993 DEBUG 9732 --- [ntainer#0-0-C-1] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-1, groupId=my_project_group_id] Initiating connection to node localhost:9092 (id: 2147483646 rack: null) using address localhost/127.0.0.1
20 2020-03-13 14:24:38.002  INFO 9732 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-1, groupId=my_project_group_id] Revoking previously assigned partitions []
21 2020-03-13 14:24:42.167  WARN 9732 --- [ntainer#0-0-C-1] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-1, groupId=my_project_group_id] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.

这是我的Kafka Docker 作品:

kafka:
    image: my_artifactory:5555/wurstmeister/kafka:latest
    ports:
      - "9092:9092"
      - "9093:9093"
    environment:
      KAFKA_LISTENERS: INTERNAL://kafka:9093,EXTERNAL://:9092
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:9093,EXTERNAL://my_kafka_host_ip:9092
      KAFKA_ADVERTISED_HOST_NAME: my_kafka_host_ip
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_BROKER_ID: 1
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
bis0qfac

bis0qfac1#

欢迎使用stackoverflow:-)
您没有正确配置您的kafka代理。当客户端连接到代理时,它会请求所有可用代理及其侦听器的详细信息。这就是所谓的播发侦听器(因为它是代理“播发”侦听器的内容)。
您可以在客户端日志中看到这种情况:

Completed receive from node -1 for METADATA with correlation id 2
received … node_id=1,host=localhost,port=9092…

您需要配置 advertised.hosts 在您的代理上作为主机和端口,您的客户机可以在其上访问它。这意味着docker容器必须向主机公开其端口,并且您的主机没有防火墙阻止该端口。
有关详细信息,请参阅https://rmoff.net/2018/08/02/kafka-listeners-explained/

相关问题