CODEWITHSUNDEEP
spring-integrationapache-kafka
Markus Lamm
Markus Lamm

Reputation: 391

Spring Integration KafkaProducerMessageHandler TimeoutException on metadata sending

I'm trying to achieve the following scenario with Spring Integration and Kafka on Docker:

The polling works fine, and my application setup seems to be correct, because the app is starting without any connection exceptions.

As you can see in the output, the KafkaProducerMessageHandler receives the message, but hangs with 'Trying to send metadata request to node -1' for over a minute, then throws a TimeoutException.

Here is the output and code, needed to link:

OUTPUT

kafka-broker.log

zookeeper.log

Here my application setup:

docker-compose.yml

application.yml

KafkaIntegration.java

The KafkaConfig ist based and the Kafka example from https://github.com/spring-projects/spring-integration-samples and the Kafka Docker container from https://github.com/wurstmeister/kafka-docker.

Thanks for any ideas and suggestions!!!!

Upvotes: 1

Views: 620

Answers (2)

Markus Lamm
Markus Lamm

Reputation: 391

I'm stupid!

My docker-compose was:

kafka:
    image: wurstmeister/kafka
    ports:
        - "9092:9092"
    links:
        - zookeeper:zk
    environment:
        KAFKA_ADVERTISED_HOST_NAME: 192.168.59.103
    volumes:
        - /var/run/docker.sock:/var/run/docker.sock

where 192.168.59.103 is my IP. I had to change it to localhost, now it seems fine.

17:25:26.797 DEBUG 25436 --- [ask-scheduler-9] o.s.integration.channel.DirectChannel    : preSend on channel 'kafkaChannel', message: GenericMessage [payload={"login":"markuslamm","id":1075906,"avatar_url":"https://avatars.githubusercontent.com/u/1075906?v=3","gravatar_id":"","url":"https://api.github.com/users/markuslamm","html_url":"https://github.com/markuslamm","followers_url":"https://api.github.com/users/markuslamm/followers","following_url":"https://api.github.com/users/markuslamm/following{/other_user}","gists_url":"https://api.github.com/users/markuslamm/gists{/gist_id}","starred_url":"https://api.github.com/users/markuslamm/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/markuslamm/subscriptions","organizations_url":"https://api.github.com/users/markuslamm/orgs","repos_url":"https://api.github.com/users/markuslamm/repos","events_url":"https://api.github.com/users/markuslamm/events{/privacy}","received_events_url":"https://api.github.com/users/markuslamm/received_events","type":"User","site_admin":false,"name":"Markus Lamm","company":null,"blog":null,"location":null,"email":null,"hireable":null,"bio":null,"public_repos":21,"public_gists":0,"followers":0,"following":0,"created_at":"2011-09-24T07:04:34Z","updated_at":"2016-01-26T15:19:00Z"}, headers={Server=GitHub.com, Cache-Control=public, max-age=60, s-maxage=60, ETag="c06cdef5b902bb019cced50412d4b913", Vary=[Accept, Accept-Encoding], Last-Modified=1453821540000, id=6bffbe99-0e6d-55a5-65a1-208c7be7542d, Content-Length=1135, contentType=application/json;charset=utf-8, http_statusCode=200, Date=1453825526000, timestamp=1453825526797}]
2016-01-26 17:25:26.797 DEBUG 25436 --- [ask-scheduler-9] ssor$ReplyProducingMessageHandlerWrapper : kafkaIntegration.kafkaHandler.serviceActivator.handler received message: GenericMessage [payload={"login":"markuslamm","id":1075906,"avatar_url":"https://avatars.githubusercontent.com/u/1075906?v=3","gravatar_id":"","url":"https://api.github.com/users/markuslamm","html_url":"https://github.com/markuslamm","followers_url":"https://api.github.com/users/markuslamm/followers","following_url":"https://api.github.com/users/markuslamm/following{/other_user}","gists_url":"https://api.github.com/users/markuslamm/gists{/gist_id}","starred_url":"https://api.github.com/users/markuslamm/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/markuslamm/subscriptions","organizations_url":"https://api.github.com/users/markuslamm/orgs","repos_url":"https://api.github.com/users/markuslamm/repos","events_url":"https://api.github.com/users/markuslamm/events{/privacy}","received_events_url":"https://api.github.com/users/markuslamm/received_events","type":"User","site_admin":false,"name":"Markus Lamm","company":null,"blog":null,"location":null,"email":null,"hireable":null,"bio":null,"public_repos":21,"public_gists":0,"followers":0,"following":0,"created_at":"2011-09-24T07:04:34Z","updated_at":"2016-01-26T15:19:00Z"}, headers={Server=GitHub.com, Cache-Control=public, max-age=60, s-maxage=60, ETag="c06cdef5b902bb019cced50412d4b913", Vary=[Accept, Accept-Encoding], Last-Modified=1453821540000, id=6bffbe99-0e6d-55a5-65a1-208c7be7542d, Content-Length=1135, contentType=application/json;charset=utf-8, http_statusCode=200, Date=1453825526000, timestamp=1453825526797}]
2016-01-26 17:25:26.797 DEBUG 25436 --- [ask-scheduler-9] o.s.i.k.o.KafkaProducerMessageHandler    : kafkaHandler received message: GenericMessage [payload={"login":"markuslamm","id":1075906,"avatar_url":"https://avatars.githubusercontent.com/u/1075906?v=3","gravatar_id":"","url":"https://api.github.com/users/markuslamm","html_url":"https://github.com/markuslamm","followers_url":"https://api.github.com/users/markuslamm/followers","following_url":"https://api.github.com/users/markuslamm/following{/other_user}","gists_url":"https://api.github.com/users/markuslamm/gists{/gist_id}","starred_url":"https://api.github.com/users/markuslamm/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/markuslamm/subscriptions","organizations_url":"https://api.github.com/users/markuslamm/orgs","repos_url":"https://api.github.com/users/markuslamm/repos","events_url":"https://api.github.com/users/markuslamm/events{/privacy}","received_events_url":"https://api.github.com/users/markuslamm/received_events","type":"User","site_admin":false,"name":"Markus Lamm","company":null,"blog":null,"location":null,"email":null,"hireable":null,"bio":null,"public_repos":21,"public_gists":0,"followers":0,"following":0,"created_at":"2011-09-24T07:04:34Z","updated_at":"2016-01-26T15:19:00Z"}, headers={Server=GitHub.com, Cache-Control=public, max-age=60, s-maxage=60, ETag="c06cdef5b902bb019cced50412d4b913", Vary=[Accept, Accept-Encoding], Last-Modified=1453821540000, id=6bffbe99-0e6d-55a5-65a1-208c7be7542d, Content-Length=1135, contentType=application/json;charset=utf-8, http_statusCode=200, Date=1453825526000, timestamp=1453825526797}]
2016-01-26 17:25:26.798 DEBUG 25436 --- [ask-scheduler-9] ssor$ReplyProducingMessageHandlerWrapper : handler 'kafkaIntegration.kafkaHandler.serviceActivator.handler' produced no reply for request Message: GenericMessage [payload={"login":"markuslamm","id":1075906,"avatar_url":"https://avatars.githubusercontent.com/u/1075906?v=3","gravatar_id":"","url":"https://api.github.com/users/markuslamm","html_url":"https://github.com/markuslamm","followers_url":"https://api.github.com/users/markuslamm/followers","following_url":"https://api.github.com/users/markuslamm/following{/other_user}","gists_url":"https://api.github.com/users/markuslamm/gists{/gist_id}","starred_url":"https://api.github.com/users/markuslamm/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/markuslamm/subscriptions","organizations_url":"https://api.github.com/users/markuslamm/orgs","repos_url":"https://api.github.com/users/markuslamm/repos","events_url":"https://api.github.com/users/markuslamm/events{/privacy}","received_events_url":"https://api.github.com/users/markuslamm/received_events","type":"User","site_admin":false,"name":"Markus Lamm","company":null,"blog":null,"location":null,"email":null,"hireable":null,"bio":null,"public_repos":21,"public_gists":0,"followers":0,"following":0,"created_at":"2011-09-24T07:04:34Z","updated_at":"2016-01-26T15:19:00Z"}, headers={Server=GitHub.com, Cache-Control=public, max-age=60, s-maxage=60, ETag="c06cdef5b902bb019cced50412d4b913", Vary=[Accept, Accept-Encoding], Last-Modified=1453821540000, id=6bffbe99-0e6d-55a5-65a1-208c7be7542d, Content-Length=1135, contentType=application/json;charset=utf-8, http_statusCode=200, Date=1453825526000, timestamp=1453825526797}]
2016-01-26 17:25:26.798 DEBUG 25436 --- [ask-scheduler-9] o.s.integration.channel.DirectChannel    : postSend (sent=true) on channel 'kafkaChannel', message: GenericMessage [payload={"login":"markuslamm","id":1075906,"avatar_url":"https://avatars.githubusercontent.com/u/1075906?v=3","gravatar_id":"","url":"https://api.github.com/users/markuslamm","html_url":"https://github.com/markuslamm","followers_url":"https://api.github.com/users/markuslamm/followers","following_url":"https://api.github.com/users/markuslamm/following{/other_user}","gists_url":"https://api.github.com/users/markuslamm/gists{/gist_id}","starred_url":"https://api.github.com/users/markuslamm/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/markuslamm/subscriptions","organizations_url":"https://api.github.com/users/markuslamm/orgs","repos_url":"https://api.github.com/users/markuslamm/repos","events_url":"https://api.github.com/users/markuslamm/events{/privacy}","received_events_url":"https://api.github.com/users/markuslamm/received_events","type":"User","site_admin":false,"name":"Markus Lamm","company":null,"blog":null,"location":null,"email":null,"hireable":null,"bio":null,"public_repos":21,"public_gists":0,"followers":0,"following":0,"created_at":"2011-09-24T07:04:34Z","updated_at":"2016-01-26T15:19:00Z"}, headers={Server=GitHub.com, Cache-Control=public, max-age=60, s-maxage=60, ETag="c06cdef5b902bb019cced50412d4b913", Vary=[Accept, Accept-Encoding], Last-Modified=1453821540000, id=6bffbe99-0e6d-55a5-65a1-208c7be7542d, Content-Length=1135, contentType=application/json;charset=utf-8, http_statusCode=200, Date=1453825526000, timestamp=1453825526797}]
2016-01-26 17:25:26.799 DEBUG 25436 --- [ask-scheduler-9] o.s.integration.channel.DirectChannel    : postSend (sent=true) on channel 'httpPollChannel', message: GenericMessage [payload=, headers={id=55aecfc4-ea57-d769-4275-801b39a0416d, timestamp=1453825526420}]

Upvotes: 1

Artem Bilan
Artem Bilan

Reputation: 121542

We see something like this in your logs:

227 --- [kafka-producer-network-thread | producer-1] org.apache.kafka.clients.NetworkClient   : Sending metadata request ClientRequest(expectResponse=true, payload=null, request=RequestSend(header={api_key=3,api_version=0,correlation_id=23,client_id=producer-1}, body={topics=[test.topic]})) to node -1
2016-01-26 15:22:24.986 DEBUG 4227 --- [kafka-producer-network-thread | producer-1] org.apache.kafka.clients.NetworkClient   : Trying to send metadata request to node -1

Looks like you really should check the Broker logs.

First of all it might be :

kafka.common.InvalidTopicException: topic name TOPIC= is illegal, contains a character other than ASCII alphanumerics, '.', '_' and '-'

https://issues.apache.org/jira/browse/KAFKA-1884

So, the . in your topic name is illegal. You might use incompatible Kafka version with its client. The Spring Integration Kafka 1.3 is based on the Kafka-0.8.2.2.

From other side it might be some network problem: Kafka 0.8.2.2 - Unable to publish messages

Upvotes: 1

Related Questions