Spring集成KafkaProducerMessageHandler TimeoutException異常上發送元數據

Question

我試圖實現與Spring集成和卡夫卡在泊塢窗以下情形：

• 投票HTTP端點，導致字符串
• 將結果發送到卡夫卡

輪詢工作正常，我的應用程序設置似乎是正確的，因爲該應用程序在沒有任何連接異常的情況下啓動。

正如您在輸出中看到的那樣，KafkaProducerMessageHandler接收到該消息，但掛起了'試圖向節點-1發送元數據請求'超過一分鐘，然後拋出一個TimeoutException。

下面是輸出和代碼，需要鏈接：

OUTPUT

kafka-broker.log

zookeeper.log

這裏我的應用程序設置：

docker-compose.yml

application.yml

KafkaIntegration.java

的KafkaConfig IST基於從https://github.com/spring-projects/spring-integration-samples和從https://github.com/wurstmeister/kafka-docker卡夫卡多克爾容器卡夫卡例子。

感謝您的任何想法和建議！

Answer 1

我傻啊！

我的搬運工，撰寫了：

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 

其中192.168.59.103是我的IP。我必須將其更改爲localhost，現在看起來很好。

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}] 

Answer 2

我們看到了你的日誌是這樣的：

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 

看起來你真的應該檢查代理日誌。所有的

首先，它可能是：

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

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

所以，在你的主題名稱的.是非法的。您可能會在客戶端使用不兼容的Kafka版本。 Spring集成Kafka 1.3基於Kafka-0.8.2.2。

從對方可能是一些網絡問題：Kafka 0.8.2.2 - Unable to publish messages