那些遇到过的问题

配置 Kafka 客户端以使用颁发的 SSL 密钥/证书进行连接

Rob*_*rty 3 apache-kafka

我正在使用 Heroku Kafka，它运行 0.10.1.1 并使用 SSL。他们只支持最新的协议。

Heroku Kafka 使用 SSL 进行身份验证并颁发客户端证书和密钥，并提供 CA 证书。我将它们分别放置在client_cert.pem、client_key.pem和trusted_cert.pem中，并运行以下命令来构建密钥库：

openssl pkcs12 -export -in client_cert.pem -inkey client_key.pem -certfile client_cert.pem -out client.p12
keytool -importkeystore -srckeystore client.p12 -srcstoretype pkcs12 -destkeystore kafka.keystore.jks -deststoretype JKS
keytool -keystore kafka.truststore.jks -alias CARoot -import -file trusted_cert.pem
Run Code Online (Sandbox Code Playgroud)

然后我创建了client-ssl.properties包含以下内容的内容：

ssl.protocol=SSL
security.protocol=SSL
ssl.truststore.location=kafka.truststore.jks
ssl.truststore.type=JKS
ssl.truststore.password=xxxx
ssl.keystore.location=kafka.keystore.jks
ssl.keystore.type=JKS
ssl.keystore.password=xxxx
ssl.key.password=xxxx
Run Code Online (Sandbox Code Playgroud)

然后我将kafka-console-producer（版本 0.10.1.1）与以下内容一起使用：

kafka-console-producer --broker-list kafka+ssl://a.a.a.a:9096,kafka+ssl://b.b.b.b:9096,kafka+ssl://c.c.c.c:9096 --producer.config client-ssl.properties --topic robintest
Run Code Online (Sandbox Code Playgroud)

（robintest主题已创建。）

[2017-01-31 10:06:50,385] INFO ProducerConfig values: 
    acks = 1
    batch.size = 16384
    block.on.buffer.full = false
    bootstrap.servers = [kafka+ssl://a.a.a.a:9096, kafka+ssl://b.b.b.b:9096, kafka+ssl://c.c.c.c:9096]
    buffer.memory = 33554432
    client.id = console-producer
    compression.type = none
    connections.max.idle.ms = 540000
    interceptor.classes = null
    key.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer
    linger.ms = 1000
    max.block.ms = 60000
    max.in.flight.requests.per.connection = 5
    max.request.size = 1048576
    metadata.fetch.timeout.ms = 60000
    metadata.max.age.ms = 300000
    metric.reporters = []
    metrics.num.samples = 2
    metrics.sample.window.ms = 30000
    partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
    receive.buffer.bytes = 32768
    reconnect.backoff.ms = 50
    request.timeout.ms = 1500
    retries = 3
    retry.backoff.ms = 100
    sasl.kerberos.kinit.cmd = /usr/bin/kinit
    sasl.kerberos.min.time.before.relogin = 60000
    sasl.kerberos.service.name = null
    sasl.kerberos.ticket.renew.jitter = 0.05
    sasl.kerberos.ticket.renew.window.factor = 0.8
    sasl.mechanism = GSSAPI
    security.protocol = SSL
    send.buffer.bytes = 102400
    ssl.cipher.suites = null
    ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
    ssl.endpoint.identification.algorithm = null
    ssl.key.password = [hidden]
    ssl.keymanager.algorithm = SunX509
    ssl.keystore.location = kafka.keystore.jks
    ssl.keystore.password = [hidden]
    ssl.keystore.type = JKS
    ssl.protocol = SSL
    ssl.provider = null
    ssl.secure.random.implementation = null
    ssl.trustmanager.algorithm = PKIX
    ssl.truststore.location = kafka.truststore.jks
    ssl.truststore.password = [hidden]
    ssl.truststore.type = JKS
    timeout.ms = 30000
    value.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer
 (org.apache.kafka.clients.producer.ProducerConfig)
[2017-01-31 10:06:50,390] INFO ProducerConfig values: 
    acks = 1
    batch.size = 16384
    block.on.buffer.full = false
    bootstrap.servers = [kafka+ssl://a.a.a.a:9096, kafka+ssl://b.b.b.b:9096, kafka+ssl://c.c.c.c:9096]
    buffer.memory = 33554432
    client.id = console-producer
    compression.type = none
    connections.max.idle.ms = 540000
    interceptor.classes = null
    key.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer
    linger.ms = 1000
    max.block.ms = 60000
    max.in.flight.requests.per.connection = 5
    max.request.size = 1048576
    metadata.fetch.timeout.ms = 60000
    metadata.max.age.ms = 300000
    metric.reporters = []
    metrics.num.samples = 2
    metrics.sample.window.ms = 30000
    partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
    receive.buffer.bytes = 32768
    reconnect.backoff.ms = 50
    request.timeout.ms = 1500
    retries = 3
    retry.backoff.ms = 100
    sasl.kerberos.kinit.cmd = /usr/bin/kinit
    sasl.kerberos.min.time.before.relogin = 60000
    sasl.kerberos.service.name = null
    sasl.kerberos.ticket.renew.jitter = 0.05
    sasl.kerberos.ticket.renew.window.factor = 0.8
    sasl.mechanism = GSSAPI
    security.protocol = SSL
    send.buffer.bytes = 102400
    ssl.cipher.suites = null
    ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
    ssl.endpoint.identification.algorithm = null
    ssl.key.password = [hidden]
    ssl.keymanager.algorithm = SunX509
    ssl.keystore.location = kafka.keystore.jks
    ssl.keystore.password = [hidden]
    ssl.keystore.type = JKS
    ssl.protocol = SSL
    ssl.provider = null
    ssl.secure.random.implementation = null
    ssl.trustmanager.algorithm = PKIX
    ssl.truststore.location = kafka.truststore.jks
    ssl.truststore.password = [hidden]
    ssl.truststore.type = JKS
    timeout.ms = 30000
    value.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer
 (org.apache.kafka.clients.producer.ProducerConfig)
[2017-01-31 10:06:50,396] DEBUG Added sensor with name bufferpool-wait-time (org.apache.kafka.common.metrics.Metrics)
[2017-01-31 10:06:50,398] DEBUG Added sensor with name buffer-exhausted-records (org.apache.kafka.common.metrics.Metrics)
[2017-01-31 10:06:50,399] DEBUG Updated cluster metadata version 1 to Cluster(id = null, nodes = [b.b.b.b:9096 (id: -2 rack: null), c.c.c.c:9096 (id: -3 rack: null), a.a.a.a:9096 (id: -1 rack: null)], partitions = []) (org.apache.kafka.clients.Metadata)
[2017-01-31 10:06:50,457] DEBUG Added sensor with name connections-closed: (org.apache.kafka.common.metrics.Metrics)
[2017-01-31 10:06:50,457] DEBUG Added sensor with name connections-created: (org.apache.kafka.common.metrics.Metrics)
[2017-01-31 10:06:50,457] DEBUG Added sensor with name bytes-sent-received: (org.apache.kafka.common.metrics.Metrics)
[2017-01-31 10:06:50,457] DEBUG Added sensor with name bytes-sent: (org.apache.kafka.common.metrics.Metrics)
[2017-01-31 10:06:50,458] DEBUG Added sensor with name bytes-received: (org.apache.kafka.common.metrics.Metrics)
[2017-01-31 10:06:50,458] DEBUG Added sensor with name select-time: (org.apache.kafka.common.metrics.Metrics)
[2017-01-31 10:06:50,459] DEBUG Added sensor with name io-time: (org.apache.kafka.common.metrics.Metrics)
[2017-01-31 10:06:50,462] DEBUG Added sensor with name batch-size (org.apache.kafka.common.metrics.Metrics)
[2017-01-31 10:06:50,462] DEBUG Added sensor with name compression-rate (org.apache.kafka.common.metrics.Metrics)
[2017-01-31 10:06:50,462] DEBUG Added sensor with name queue-time (org.apache.kafka.common.metrics.Metrics)
[2017-01-31 10:06:50,462] DEBUG Added sensor with name request-time (org.apache.kafka.common.metrics.Metrics)
[2017-01-31 10:06:50,464] DEBUG Added sensor with name produce-throttle-time (org.apache.kafka.common.metrics.Metrics)
[2017-01-31 10:06:50,465] DEBUG Added sensor with name records-per-request (org.apache.kafka.common.metrics.Metrics)
[2017-01-31 10:06:50,465] DEBUG Added sensor with name record-retries (org.apache.kafka.common.metrics.Metrics)
[2017-01-31 10:06:50,465] DEBUG Added sensor with name errors (org.apache.kafka.common.metrics.Metrics)
[2017-01-31 10:06:50,465] DEBUG Added sensor with name record-size-max (org.apache.kafka.common.metrics.Metrics)
[2017-01-31 10:06:50,467] DEBUG Starting Kafka producer I/O thread. (org.apache.kafka.clients.producer.internals.Sender)
[2017-01-31 10:06:50,468] INFO Kafka version : 0.10.1.1 (org.apache.kafka.common.utils.AppInfoParser)
[2017-01-31 10:06:50,468] INFO Kafka commitId : f10ef2720b03b247 (org.apache.kafka.common.utils.AppInfoParser)
[2017-01-31 10:06:50,468] DEBUG Kafka producer started (org.apache.kafka.clients.producer.KafkaProducer)
Run Code Online (Sandbox Code Playgroud)

此时，我发送一条记录并按 Enter 键。

[2017-01-31 10:06:53,194] DEBUG Initialize connection to node -2 for sending metadata request (org.apache.kafka.clients.NetworkClient)
[2017-01-31 10:06:53,194] DEBUG Initiating connection to node -2 at b.b.b.b:9096. (org.apache.kafka.clients.NetworkClient)
[2017-01-31 10:06:53,457] DEBUG Added sensor with name node--2.bytes-sent (org.apache.kafka.common.metrics.Metrics)
[2017-01-31 10:06:53,457] DEBUG Added sensor with name node--2.bytes-received (org.apache.kafka.common.metrics.Metrics)
[2017-01-31 10:06:53,458] DEBUG Added sensor with name node--2.latency (org.apache.kafka.common.metrics.Metrics)
[2017-01-31 10:06:53,460] DEBUG Created socket with SO_RCVBUF = 33304, SO_SNDBUF = 102808, SO_TIMEOUT = 0 to node -2 (org.apache.kafka.common.network.Selector)
[2017-01-31 10:06:53,463] DEBUG Completed connection to node -2 (org.apache.kafka.clients.NetworkClient)
[2017-01-31 10:06:53,692] DEBUG Sending metadata request {topics=[robintest]} to node -2 (org.apache.kafka.clients.NetworkClient)
[2017-01-31 10:06:53,724] DEBUG Connection with ec2-34-194-25-39.compute-1.amazonaws.com/b.b.b.b disconnected (org.apache.kafka.common.network.Selector)
java.io.EOFException
    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.network.KafkaChannel.receive(KafkaChannel.java:154)
    at org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:135)
    at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:343)
    at org.apache.kafka.common.network.Selector.poll(Selector.java:291)
    at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:260)
    at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:236)
    at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:135)
    at java.lang.Thread.run(Thread.java:745)
[2017-01-31 10:06:53,728] DEBUG Node -2 disconnected. (org.apache.kafka.clients.NetworkClient)
[2017-01-31 10:06:53,728] WARN Bootstrap broker b.b.b.b:9096 disconnected (org.apache.kafka.clients.NetworkClient)
[2017-01-31 10:06:53,729] DEBUG Initialize connection to node -1 for sending metadata request (org.apache.kafka.clients.NetworkClient)
[2017-01-31 10:06:53,729] DEBUG Initiating connection to node -1 at a.a.a.a:9096. (org.apache.kafka.clients.NetworkClient)
[2017-01-31 10:06:53,791] DEBUG Added sensor with name node--1.bytes-sent (org.apache.kafka.common.metrics.Metrics)
[2017-01-31 10:06:53,792] DEBUG Added sensor with name node--1.bytes-received (org.apache.kafka.common.metrics.Metrics)
[2017-01-31 10:06:53,792] DEBUG Added sensor with name node--1.latency (org.apache.kafka.common.metrics.Metrics)
[2017-01-31 10:06:53,792] DEBUG Created socket with SO_RCVBUF = 33304, SO_SNDBUF = 102808, SO_TIMEOUT = 0 to node -1 (org.apache.kafka.common.network.Selector)
[2017-01-31 10:06:53,792] DEBUG Completed connection to node -1 (org.apache.kafka.clients.NetworkClient)
[2017-01-31 10:06:53,994] DEBUG Sending metadata request {topics=[robintest]} to node -1 (org.apache.kafka.clients.NetworkClient)
[2017-01-31 10:06:54,025] DEBUG Connection with ec2-34-194-39-35.compute-1.amazonaws.com/a.a.a.a disconnected (org.apache.kafka.common.network.Selector)
java.io.EOFException
    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.network.KafkaChannel.receive(KafkaChannel.java:154)
    at org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:135)
    at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:343)
    at org.apache.kafka.common.network.Selector.poll(Selector.java:291)
    at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:260)
    at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:236)
    at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:135)
    at java.lang.Thread.run(Thread.java:745)
[2017-01-31 10:06:54,026] DEBUG Node -1 disconnected. (org.apache.kafka.clients.NetworkClient)
[2017-01-31 10:06:54,026] WARN Bootstrap broker a.a.a.a:9096 disconnected (org.apache.kafka.clients.NetworkClient)
[2017-01-31 10:06:54,027] DEBUG Initialize connection to node -3 for sending metadata request (org.apache.kafka.clients.NetworkClient)
[2017-01-31 10:06:54,027] DEBUG Initiating connection to node -3 at c.c.c.c:9096. (org.apache.kafka.clients.NetworkClient)
[2017-01-31 10:06:54,102] DEBUG Added sensor with name node--3.bytes-sent (org.apache.kafka.common.metrics.Metrics)
[2017-01-31 10:06:54,103] DEBUG Added sensor with name node--3.bytes-received (org.apache.kafka.common.metrics.Metrics)
[2017-01-31 10:06:54,103] DEBUG Added sensor with name node--3.latency (org.apache.kafka.common.metrics.Metrics)
[2017-01-31 10:06:54,104] DEBUG Created socket with SO_RCVBUF = 33304, SO_SNDBUF = 102808, SO_TIMEOUT = 0 to node -3 (org.apache.kafka.common.network.Selector)
[2017-01-31 10:06:54,104] DEBUG Completed connection to node -3 (org.apache.kafka.clients.NetworkClient)
[2017-01-31 10:06:54,309] DEBUG Sending metadata request {topics=[robintest]} to node -3 (org.apache.kafka.clients.NetworkClient)
[2017-01-31 10:06:54,342] DEBUG Connection with ec2-34-194-45-119.compute-1.amazonaws.com/c.c.c.c disconnected (org.apache.kafka.common.network.Selector)
java.io.EOFException
    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.network.KafkaChannel.receive(KafkaChannel.java:154)
    at org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:135)
    at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:343)
    at org.apache.kafka.common.network.Selector.poll(Selector.java:291)
    at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:260)
    at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:236)
    at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:135)
    at java.lang.Thread.run(Thread.java:745)
[2017-01-31 10:06:54,342] DEBUG Node -3 disconnected. (org.apache.kafka.clients.NetworkClient)
[2017-01-31 10:06:54,343] WARN Bootstrap broker c.c.c.c:9096 disconnected (org.apache.kafka.clients.NetworkClient)
[2017-01-31 10:06:54,343] DEBUG Initialize connection to node -1 for sending metadata request (org.apache.kafka.clients.NetworkClient)
[2017-01-31 10:06:54,343] DEBUG Initiating connection to node -1 at a.a.a.a:9096. (org.apache.kafka.clients.NetworkClient)
[2017-01-31 10:06:54,348] DEBUG Initialize connection to node -2 for sending metadata request (org.apache.kafka.clients.NetworkClient)
[2017-01-31 10:06:54,348] DEBUG Initiating connection to node -2 at b.b.b.b:9096. (org.apache.kafka.clients.NetworkClient)
[2017-01-31 10:06:54,376] DEBUG Created socket with SO_RCVBUF = 33304, SO_SNDBUF = 102808, SO_TIMEOUT = 0 to node -2 (org.apache.kafka.common.network.Selector)
[2017-01-31 10:06:54,377] DEBUG Completed connection to node -2 (org.apache.kafka.clients.NetworkClient)
[2017-01-31 10:06:54,379] DEBUG Created socket with SO_RCVBUF = 33304, SO_SNDBUF = 102808, SO_TIMEOUT = 0 to node -1 (org.apache.kafka.common.network.Selector)
[2017-01-31 10:06:54,379] DEBUG Completed connection to node -1 (org.apache.kafka.clients.NetworkClient)
Run Code Online (Sandbox Code Playgroud)

这些条目将永远持续下去，直到我终止该过程。

我尝试了每种配置组合，包括在属性文件中为所有配置添加前缀producer.、删除整个配置（这似乎没有什么区别）、将密码设置为不正确的值（这似乎没有区别）。我还尝试使用他们的凭据连接到不同的提供商 (www.cloudkarafka.com)，并且得到相同的结果。所以这看起来肯定是一个配置问题。

Rob*_*rty 5

事实证明，我的 Kafka 集群（Heroku 附加组件）实际上并未运行 0.10.1.1，它运行的是 0.10.0.1。两者的消费者 API 似乎不兼容。（我不得不说，“这正是语义版本控制存在的原因。”）

要升级 Heroku 上运行的 Kafka，请使用：heroku kafka:upgrade --version 0.10 升级到最新的 0.10.X 版本。因此，如果您使用的是 0.9 并且想要 0.10.0.1，那么祝您好运。

归档时间：

查看次数：

18544 次

最近记录：

6 年，4 月 前
相关归档
平衡Kafka消费者 9
whatsapp 或 FB Messenger 中使用的队列 7
卡夫卡对数压缩消耗数据 6
保留配置offsets.retention.minutes和log.retention.minutes之间的区别 6
如何在默认情况下从 Kafka Spring Cloud Stream 消费并同时消费由 Confluent API 生成的 Kafka 消息？ 5
我可以使用我的服务网格进行事件驱动的消息传递吗？ 4
Kafka Stream聚合器 - 如何在发送消息之前设置在聚合中等待的时间？ 3
来自kafka消费者的InstanceAlreadyExistsException 2
kafka生产者API中的头信息 2
使用kafka接收器重命名elasticsearch中的索引 1
难疑归档
JavaScript中使用"严格"做什么,背后的原因是什么？ 7339
C#中字符串和字符串有什么区别？ 6250
如何确定数组是否包含Java中的特定值？ 2194
2048游戏的最佳算法是什么？ 1893
轻松获取最新的git子模块 1748
如何在Linux中更改echo的输出颜色 1582
在JavaScript中停止setInterval调用 1332
是否有"以前的兄弟"CSS选择器？ 1253
Objective-C中的快捷方式用于连接NSStrings 1114
致命错误:Python.h:没有这样的文件或目录 1042