如何使用 Beam KafkaIO 配置客户端身份验证

Question

我一直在浏览 Beam KafkaIO 教程，并一直在尝试查找有关 kafka 客户端身份验证的文档，但到目前为止只找到了非常基本的示例。我需要为 Kafkaio 客户端提供以下配置才能成功进行身份验证：

bootstrap.servers=kafka1:9093
security.protocol=SSL
ssl.truststore.location=/var/private/ssl/kafka.client.truststore.jks
ssl.truststore.password=test1234
ssl.keystore.location=/var/private/ssl/kafka.client.keystore.jks
ssl.keystore.password=test1234
ssl.key.password=test1234

我将如何指定此配置？

到目前为止，我在示例中找到的都是这样配置的：

p.apply(KafkaIO.<Long, String>read()
.withBootstrapServers("kafka1:9022")
.withTopic("test-topic")
.withKeyDeserializer(LongDeserializer.class)
.withValueDeserializer(StringDeserializer.class)

Answer 1

您可以使用updateConsumerProperties(properties)方法设置ssl配置。
为此，您需要设置以下消费者属性。

Properties props = new Properties();
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, "/var/private/ssl/kafka.client.truststore.jks");    
props.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, resourcePath.get("keystore.jks"));
props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG,  "test1234");
props.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG,  "test1234"); 
props.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG,  "test1234");

在方法中传递上述属性，如下所示：

p.apply(KafkaIO.<Long, String>read()
.withBootstrapServers("kafka1:9022")
.withTopic("test-topic")
.withKeyDeserializer(LongDeserializer.class)
.withValueDeserializer(StringDeserializer.class)
.updateConsumerProperties(props)

您可以在此处找到有关如何在 KafkaIO 中设置自定义属性的更多文档：https://beam.apache.org/releases/javadoc/2.5.0/org/apache/beam/sdk/io/kafka/KafkaIO.html

如何使用 Beam KafkaIO 配置客户端身份验证

How to Configure Client Authentication with Beam KafkaIO

java

apache-kafka

apache-beam