连接到 Bluemix 上的消息中心的 Kafka Producer 超时问题

Question

我正在尝试连接到 IBM Bluemix 消息中心并按照示例

使用 java 生成一条消息

https://github.com/ibm-messaging/message-hub-samples/tree/master/java/message-hub-kafka-ssl

producer.properties

key.serializer=org.apache.kafka.common.serialization.ByteArraySerializer
value.serializer=org.apache.kafka.common.serialization.ByteArraySerializer
acks=-1
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
ssl.protocol=TLSv1.2
ssl.enabled.protocols=TLSv1.2
ssl.truststore.password=changeit
ssl.truststore.type=JKS
ssl.endpoint.identification.algorithm=HTTPS
ssl.truststore.location=/Library/Java/JavaVirtualMachines/jdk1.8.0_66.jdk/Contents/Home/jre/lib/security/cacerts

jaas.conf.template

KafkaClient {
    com.ibm.messagehub.login.MessageHubLoginModule required
    serviceName="kafka"
    user="$USERNAME"
    password="$PASSWORD";
};

Producer.java 片段

ProducerRecord<byte[], byte[]> record = new ProducerRecord<byte[], byte[]>(
                "MyTopic",
                KEY.getBytes("UTF-8"),
                "MESSAGE".getBytes("UTF-8"));

        // Synchronously wait for a response from Message Hub / Kafka.
        RecordMetadata m = kafkaProducer.send(record).get();

问题是，当我尝试获取 Future RecordMetadata 时出现超时异常

java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 60000 ms.
at org.apache.kafka.clients.producer.KafkaProducer$FutureFailure.<init>(KafkaProducer.java:730)
at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:483)
at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:430)
at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:353)

阅读关于同一主题的早期 post

提到的可能原因是主题不是 created.I 可以在 bluemix 控制台中看到主题并进行验证，我在推送消息之前调用了 rest 服务来获取主题列表

RESTRequest restApi = new RESTRequest(getRestHost(),getApiKey());

String topics = restApi.get("/admin/topics", false);

logger.info("Topics present in the system: " + topics);

它 returns 我正在尝试推送消息但出现超时错误的主题。

有人可以帮我调试问题吗

更新

根据评论，我为kafka启用了调试日志，这是日志的顺序

2016-11-23 08:48:20.906 DEBUG 72885 --- [sage-hub-sample]    org.apache.kafka.clients.NetworkClient   : Initialize connection to node -5 for sending metadata request
2016-11-23 08:48:20.906 DEBUG 72885 --- [sage-hub-sample] org.apache.kafka.clients.NetworkClient   : Initiating connection to node -5 at kafka05-prod01.messagehub.services.us-south.bluemix.net:9093.
2016-11-23 08:48:20.914 DEBUG 72885 --- [sage-hub-sample] o.a.k.c.s.a.SaslClientAuthenticator      : Set SASL client state to SEND_HANDSHAKE_REQUEST
2016-11-23 08:48:20.915 DEBUG 72885 --- [sage-hub-sample] o.a.k.c.s.a.SaslClientAuthenticator      : Creating SaslClient: client=messagehub/ignorehost@messagehub.ibm.com;service=kafka;serviceHostname=kafka05-prod01.messagehub.services.us-south.bluemix.net;mechs=[PLAIN]
2016-11-23 08:48:20.979 DEBUG 72885 --- [sage-hub-sample] org.apache.kafka.common.metrics.Metrics  : Added sensor with name node--5.bytes-sent
2016-11-23 08:48:20.980 DEBUG 72885 --- [sage-hub-sample] org.apache.kafka.common.metrics.Metrics  : Added sensor with name node--5.bytes-received
2016-11-23 08:48:20.980 DEBUG 72885 --- [sage-hub-sample] org.apache.kafka.common.metrics.Metrics  : Added sensor with name node--5.latency
2016-11-23 08:48:20.982 DEBUG 72885 --- [sage-hub-sample] org.apache.kafka.clients.NetworkClient   : Completed connection to node -5
2016-11-23 08:48:21.080 DEBUG 72885 --- [sage-hub-sample] o.a.k.c.s.a.SaslClientAuthenticator      : Set SASL client state to RECEIVE_HANDSHAKE_RESPONSE

2016-11-23 08:48:21.264 DEBUG 72885 --- [sage-hub-sample] o.a.k.c.s.a.SaslClientAuthenticator      : Set SASL client state to INITIAL
2016-11-23 08:48:21.265 DEBUG 72885 --- [sage-hub-sample] o.a.k.c.s.a.SaslClientAuthenticator      : Set SASL client state to INTERMEDIATE
2016-11-23 08:48:21.284 DEBUG 72885 --- [sage-hub-sample]  o.apache.kafka.common.network.Selector   : Connection with kafka05-prod01.messagehub.services.us-south.bluemix.net/23.246.202.55 disconnected

java.io.EOFException: null
at org.apache.kafka.common.network.SslTransportLayer.read(SslTransportLayer.java:488)
at org.apache.kafka.common.network.NetworkReceive.readFromReadableChannel(NetworkReceive.java:81)
at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:71)
at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.receiveResponseOrToken(SaslClientAuthenticator.java:239)
at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.authenticate(SaslClientAuthenticator.java:182)
at org.apache.kafka.common.network.KafkaChannel.prepare(KafkaChannel.java:64)
at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:318)
at org.apache.kafka.common.network.Selector.poll(Selector.java:283)
at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:260)
at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:229)
at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:134)
at java.lang.Thread.run(Thread.java:745)

以上所有内容均针对 5 个经纪人中的任何一个进行了记录。此外，这些都是调试语句，所以我不确定它们是否是错误的。

-塔塔

Answer 1

请启用 log4j DEBUG 级别，Kafka 客户端跟踪可能有助于识别任何连接问题。尝试将最后一行更改为 log4j.logger.org.apache.kafka=DEBUG 在 log4j.properties 文件中，重建并查看详细输出。欢迎转发。

Answer 2

我发现您的 JAAS 文件存在一些问题：

您所指的示例使用的是 Kafka 0.10。0.X 因此您不应使用 Kafka 0.9 的旧 Message Hub 登录模块。所以将 "com.ibm.messagehub.login.MessageHubLoginModule" 替换为 "org.apache.kafka.common.security.plain.PlainLoginModule"
用户名字段称为 "username" 而不是 "user"。应该是 username="$USERNAME"

连接到 Bluemix 上的消息中心的 Kafka Producer 超时问题

Kafka Producer Timeout Issue connecting to message-hub on Bluemix

message-hub

ibm-cloud