如何在 Quarkus 中使用 Kafka 的 TLS？

Question

Quarkus 的 Kafka 指南在Docker 本地运行 Kafka 时效果很好。我试图通过将本地 Kafka 服务替换为需要 TLS 的云中托管 Kafka 服务来更改此示例。

有谁知道我该如何配置这个？在Quarkus 文档和Smallrye 文档中，我没有看到任何相关属性。

我想使用 IBM Cloud 中的 Kafka 服务。根据文档，我在 application.properties 中尝试了以下配置：

kafka.bootstrap.servers=broker-0-8c8cph49mx2p2wqy.kafka.svc01.us-south.eventstreams.cloud.ibm.com:9093,broker-4-8c8cph49mx2p2wqy.kafka.svc01.us-south.eventstreams.cloud.ibm.com:9093,broker-3-8c8cph49mx2p2wqy.kafka.svc01.us-south.eventstreams.cloud.ibm.com:9093,broker-5-8c8cph49mx2p2wqy.kafka.svc01.us-south.eventstreams.cloud.ibm.com:9093,broker-2-8c8cph49mx2p2wqy.kafka.svc01.us-south.eventstreams.cloud.ibm.com:9093,broker-1-8c8cph49mx2p2wqy.kafka.svc01.us-south.eventstreams.cloud.ibm.com:9093
kafka.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="token" password="...";
kafka.sasl.mechanism=PLAIN
kafka.security.protocol=SASL_SSL
kafka.ssl.protocol=TLSv1.2

更新：我也尝试了下面 Gunnar 的建议，但它不起作用。当我使用以下 application.properties 时...

mp.messaging.outgoing.generated-price.connector=smallrye-kafka
mp.messaging.outgoing.generated-price.topic=prices
mp.messaging.outgoing.generated-price.value.serializer=org.apache.kafka.common.serialization.IntegerSerializer

mp.messaging.outgoing.generated-price.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="token" password="...";
mp.messaging.outgoing.generated-price.sasl.mechanism=PLAIN
mp.messaging.outgoing.generated-price.security.protocol=SASL_SSL
mp.messaging.outgoing.generated-price.ssl.protocol=TLSv1.2

mp.messaging.incoming.prices.connector=smallrye-kafka
mp.messaging.incoming.prices.topic=prices
mp.messaging.incoming.prices.value.deserializer=org.apache.kafka.common.serialization.IntegerDeserializer

mp.messaging.outgoing.prices.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="token" password="...";
mp.messaging.outgoing.prices.sasl.mechanism=PLAIN
mp.messaging.outgoing.prices.security.protocol=SASL_SSL
mp.messaging.outgoing.prices.ssl.protocol=TLSv1.2

kafka.bootstrap.servers=broker-0-8c8cph49mx2p2wqy.kafka.svc01.us-south.eventstreams.cloud.ibm.com:9093,broker-4-8c8cph49mx2p2wqy.kafka.svc01.us-south.eventstreams.cloud.ibm.com:9093,broker-3-8c8cph49mx2p2wqy.kafka.svc01.us-south.eventstreams.cloud.ibm.com:9093,broker-5-8c8cph49mx2p2wqy.kafka.svc01.us-south.eventstreams.cloud.ibm.com:9093,broker-2-8c8cph49mx2p2wqy.kafka.svc01.us-south.eventstreams.cloud.ibm.com:9093,broker-1-8c8cph49mx2p2wqy.kafka.svc01.us-south.eventstreams.cloud.ibm.com:9093

...我收到错误： javax.enterprise.inject.spi.DeploymentException: java.lang.IllegalArgumentException: 无效的通道配置 -connector必须为prices io.quarkus.smallrye.reactivemessaging.runtime.SmallRyeReactiveMessagingLifecycle.onApplicationStart( SmallRyeReactiveMessagingLifecycle.java:22)

Quarkus 中的 Kafka 目前是否可以使用 TLS？

谢谢

Answer 1

您是否尝试过在通道级别指定相关属性？例如

mp.messaging.outgoing.generated-price.connector=smallrye-kafka
mp.messaging.outgoing.generated-price.topic=mytopic
mp.messaging.outgoing.generated-price.ssl.protocol=...
mp.messaging.outgoing.generated-price.ssl.keystore.location=...
mp.messaging.outgoing.generated-price.ssl.keystore.password=...

当多个主题需要相同的值时，您还可以引用变量。