无法将消息发布到消息中心

Question

// Asynchronous response from Message Hub / Kafka.
kafkaProducer.send(record,
new Callback() {
   public void onCompletion(RecordMetadata m, Exception e) {
       if(e != null) {
       e.printStackTrace();
       } else { 
    log.debug(" **** Message sent, offset: " + m.offset() + 
    " @ partition " + m.partition());
    log.debug(" <<<< " +
    " document_id " + key +
    " @ " + account.getActivityId());
    }
   }
});

当尝试使用上述代码向 Message Hub 发布消息时，我们总是收到以下错误。

2016-06-21 18:38:22-[信息] com.ibm.cloudant.streaming.messageHub.Client.send(476): >>>> 发送 document_id julia30 @ my_database 2016-06-21 18:38:22-[调试] org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater.maybeUpdate(623)： 初始化与节点 -1 的连接以发送元数据请求 2016-06-21 18:38:22-[调试] org.apache.kafka.clients.NetworkClient.initiateConnect(487)： 启动与节点 -1 的连接 kafka01-prod01.messagehub.services.us-south.bluemix.net:9093。 2016-06-21 18:38:22-[调试] org.apache.kafka.common.security.authenticator.SaslClientAuthenticator$1.run(105)： 创建 SaslClient： client=multiuser-adapter@multiuser.messagehub.ibm.com;service=kafka;serviceHostname=kafka01-prod01.messagehub.services.us-south.bluemix.net;mechs=[GSSAPI] 2016-06-21 18:38:22-[调试] com.ibm.cloudant.streaming.messageHub.AccountManager.(53)： 进程号：15825 2016-06-21 18:38:22-[INFO] org.apache.kafka.common.network.SaslChannelBuilder.buildChannel(92)： 由于无法创建频道 org.apache.kafka.common.KafkaException：无法配置 SaslClientAuthenticator 在 org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.configure(SaslClientAuthenticator.java:96) 在 org.apache.kafka.common.network.SaslChannelBuilder.buildChannel(SaslChannelBuilder.java:89) 在 org.apache.kafka.common.network.Selector.connect(Selector.java:162) 在 org.apache.kafka.clients.NetworkClient.initiateConnect(NetworkClient.java:489) 在 org.apache.kafka.clients.NetworkClient.access$400(NetworkClient.java:47) 在 org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater.maybeUpdate(NetworkClient.java:624) 在 org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater.maybeUpdate(NetworkClient.java:543) 在 org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:254) 在 org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:216) 在 org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:128) 在 java.lang.Thread.run(Thread.java:745) 引起：org.apache.kafka.common.KafkaException：创建SaslClient失败 在 org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.createSaslClient(SaslClientAuthenticator.java:112) 在 org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.configure(SaslClientAuthenticator.java:94) ... 10 更多 原因：javax.security.sasl.SaslException：PLAIN：必须指定授权 ID 和密码 在 com.sun.security.sasl.PlainClient.(PlainClient.java:58) 在 com.sun.security.sasl.ClientFactoryImpl.createSaslClient(ClientFactoryImpl.java:97) 在 javax.security.sasl.Sasl.createSaslClient(Sasl.java:384) 在 com.ibm.messagehub.login.MessageHubSaslClientFactory.createSaslClient(MessageHubSaslClientFactory.java:77) 在 javax.security.sasl.Sasl.createSaslClient(Sasl.java:384) 在 org.apache.kafka.common.security.authenticator.SaslClientAuthenticator$1.run(SaslClientAuthenticator.java:107) 在 org.apache.kafka.common.security.authenticator.SaslClientAuthenticator$1.run(SaslClientAuthenticator.java:102) 在 java.security.AccessController.doPrivileged（本机方法） 在 javax.security.auth.Subject.doAs(Subject.java:422) 在 org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.createSaslClient(SaslClientAuthenticator.java:102) ... 11 更多 2016-06-21 18:38:22-[错误] org.apache.kafka.clients.producer.internals.Sender.run(130)：未捕获 kafka 生产者 I/O 线程中的错误： org.apache.kafka.common.KafkaException: org.apache.kafka.common.KafkaException: 配置失败 SaslClientAuthenticator 在 org.apache.kafka.common.network.SaslChannelBuilder.buildChannel(SaslChannelBuilder.java:93) 在 org.apache.kafka.common.network.Selector.connect(Selector.java:162) 在 org.apache.kafka.clients.NetworkClient.initiateConnect(NetworkClient.java:489) 在 org.apache.kafka.clients.NetworkClient.access$400(NetworkClient.java:47) 在 org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater.maybeUpdate(NetworkClient.java:624) 在 org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater.maybeUpdate(NetworkClient.java:543) 在 org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:254) 在 org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:216) 在 org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:128) 在 java.lang.Thread.run(Thread.java:745) 引起：org.apache.kafka.common.KafkaException：配置SaslClientAuthenticator失败 在 org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.configure(SaslClientAuthenticator.java:96) 在 org.apache.kafka.common.network.SaslChannelBuilder.buildChannel(SaslChannelBuilder.java:89) ... 9 更多 引起：org.apache.kafka.common.KafkaException：创建SaslClient失败 在 org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.createSaslClient(SaslClientAuthenticator.java:112) 在 org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.configure(SaslClientAuthenticator.java:94) ... 10 更多 原因：javax.security.sasl.SaslException：PLAIN：必须指定授权 ID 和密码 在 com.sun.security.sasl.PlainClient.(PlainClient.java:58) 在 com.sun.security.sasl.ClientFactoryImpl.createSaslClient(ClientFactoryImpl.java:97) 在 javax.security.sasl.Sasl.createSaslClient(Sasl.java:384) 在 com.ibm.messagehub.login.MessageHubSaslClientFactory.createSaslClient(MessageHubSaslClientFactory.java:77) 在 javax.security.sasl.Sasl.createSaslClient(Sasl.java:384) 在 org.apache.kafka.common.security.authenticator.SaslClientAuthenticator$1.run(SaslClientAuthenticator.java:107) 在 org.apache.kafka.common.security.authenticator.SaslClientAuthenticator$1.run(SaslClientAuthenticator.java:102) 在 java.security.AccessController.doPrivileged（本机方法） 在 javax.security.auth.Subject.doAs(Subject.java:422) 在 org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.createSaslClient(SaslClientAuthenticator.java:102) ... 11 更多 2016-06-21 18:38:22-[错误] org.apache.kafka.clients.producer.internals.Sender.run(130)：未捕获 kafka 生产者 I/O 线程中的错误： java.lang.NullPointerException 在 org.apache.kafka.common.network.Selector.poll(Selector.java:268) 在 org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:256) 在 org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:216) 在 org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:128) 在 java.lang.Thread.run(Thread.java:745)

生产者的设置如下：

```

 compression.type = none
 metric.reporters = []
 metadata.max.age.ms = 300000
 metadata.fetch.timeout.ms = 60000
 reconnect.backoff.ms = 50
 sasl.kerberos.ticket.renew.window.factor = 0.8
 bootstrap.servers = [kafka01-prod01.messagehub.services.us-south.bluemix.net:9093]
 retry.backoff.ms = 100
 sasl.kerberos.kinit.cmd = /usr/bin/kinit
 buffer.memory = 33554432
 timeout.ms = 30000
 key.serializer = class org.apache.kafka.common.serialization.StringSerializer
 sasl.kerberos.service.name = null
 sasl.kerberos.ticket.renew.jitter = 0.05
 ssl.keystore.type = JKS
 ssl.trustmanager.algorithm = PKIX
 block.on.buffer.full = false
 ssl.key.password = null
 max.block.ms = 60000
 sasl.kerberos.min.time.before.relogin = 60000
 connections.max.idle.ms = 540000
 ssl.truststore.password = [hidden]
 max.in.flight.requests.per.connection = 5
 metrics.num.samples = 2
 client.id = kafka01-prod01.messagehub.services.us-south.bluemix.net%3A9093_8qp87X32V6PK5epv.1
 ssl.endpoint.identification.algorithm = HTTPS
 ssl.protocol = TLSv1.2
 request.timeout.ms = 30000
 ssl.provider = null
 ssl.enabled.protocols = [TLSv1.2]
 acks = -1
 batch.size = 16384
 ssl.keystore.location = null
 receive.buffer.bytes = 32768
 ssl.cipher.suites = null
 ssl.truststore.type = JKS
 security.protocol = SASL_SSL
 retries = 1
 max.request.size = 1048576
 value.serializer = class  org.apache.kafka.common.serialization.StringSerializer
 ssl.truststore.location = /Users/jiangph/tools/liberty/usr/shared/resources/keystore.jks
 ssl.keystore.password = null
 ssl.keymanager.algorithm = SunX509
 metrics.sample.window.ms = 30000
 send.buffer.bytes = 131072
 linger.ms = 0

```

使用上述设置，使用 Message Hub REST API 创建主题没有问题。尝试发布消息时会出现问题。

非常感谢任何想法。

Answer 1

MessageHub REST 客户端 API 的身份验证方式与 Java Kafka 客户端不同。

我发现您的日志中有一个身份验证错误： javax.security.sasl.SaslException: PLAIN: authorization ID and password must be specified at com.sun.security.sasl.PlainClient.

为 MessageHub SASL 身份验证配置 java 客户端， 请在以下位置查看 java 示例：

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

请注意，您的生产者属性应包括以下内容：

https://github.com/ibm-messaging/message-hub-samples/blob/master/java/message-hub-kafka-ssl/resources/producer.properties

你的 jaas.conf 文件应该是这样的

KafkaClient { com.ibm.messagehub.login.MessageHubLoginModule required serviceName="kafka" username="your-username" password="your-password"; }; 并且您的类路径中必须有 MessageHub 登录 jar。

HTH, 江户

Answer 2

原来问题与身份验证完全无关，而与键/值序列化有关。变化自

    props.setProperty("key.serializer", 
            "org.apache.kafka.common.serialization.StringSerializer");
    props.setProperty("value.serializer", 
            "org.apache.kafka.common.serialization.StringSerializer")

到

    props.setProperty("key.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
    props.setProperty("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");

解决了 javax.security.sasl.SaslException

奇怪的是，一旦你成功使用了 ByteArraySerialization，你甚至可以切换回 StringSerialization，一切都会继续工作。

Answer 3

我可以使用 StringSerializer 运行消费者或生产者样本，并且无需切换。 序列化程序不适用于 SASL 身份验证。