CODEWITHSUNDEEP
dockerapache-kafka
vasyaz
vasyaz

Reputation: 11

Cannot connect Kafka from Springboot application

I'm trying to connect Kafka running on a remote host in a docker from a Springboot java application running on my PC. The application tries to create a Kafka consumer. As far as I can see from logs (below) the application correctly identifies the kafka bootstrap server and tries to establish a connection. But then for some reason it starts looking for Kafka broker on the local host and obviously fails. Any ideas why this happens? The application works fine when running from the same host as Kafka.

Explanations to the Log:

line 1: my printout from java code just before setting the bootstrap server. It shows that the configuration has been correctly read from the configuration file.

line 2: The consumer configuration shows it's going to use the correct kafka host: bootstrap.servers = [MY_KAFKA_HOST:9092]

lines 3-11: The application tries to connect the correct kafka host

lines 14-20: The application tries to connect the localhost for some reason, I do not know why.

line 21: The connection to the localhost fails, as expected.

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.

This is my Kafka docker compose:

 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

Upvotes: 1

Views: 4323

Answers (1)

Robin Moffatt
Robin Moffatt

Reputation: 32090

You've not configured your Kafka broker correctly. When a client connects to the broker, it requests the details of all the available brokers and their listeners. This is known as the advertised listener (since it is what the broker "advertises" the listener as being).

You can see in the client log here this happening:

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

You need to configure the advertised.hosts on your broker to be a host and port on which your client can access it. That means the Docker container must expose its port to the host machine and your host machine not have a firewall blocking that port.

For more details see my blog https://rmoff.net/2018/08/02/kafka-listeners-explained/

Upvotes: 2

Related Questions