如何使用Java用SASL/SCRAM连接到MSK？

Question

我正试图找出如何创建一个工作Kafka生产者对一个AWS MSK集群。集群启用了TLS和SASL/scram。这是我第一次使用Kafka，关于这方面的文档相当少。我目前拥有的针对未加密本地群集的代码是：

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, serverBootstrap);
props.put(ProducerConfig.CLIENT_ID_CONFIG, CLIENT_ID);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
producer = new KafkaProducer<>(props);

我有字符串类型变量username和password，它们是从AWS秘密管理器中提取的，包含集群的用户名和密码。我还需要指定哪些其他属性？

AWS和Kafka网站上的文档都是关于在本地文件系统上生成JAAS和truststore文件的，这些文件将包含秘密：

https://docs.aws.amazon.com/msk/latest/developerguide/msk-password.html

这对我来说是不可能的。基本上，我需要创建一个通用Docker映像，它将在AWS Secrets Manager中作为一个环境变量提供秘密的名称，该名称将包含集群的用户名和密码。

我假定这是人们默认情况下如何对集群进行身份验证的，但没有说明如何做到这一点。我不使用MSK IAM集成的原因是，我们可能会达到每个代理3000个连接的限制。

Answer 1

经过大量的试验和错误，这实际上对亚马逊MSK有效：

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, serverBootstrap);
props.put(ProducerConfig.CLIENT_ID_CONFIG, CLIENT_ID);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

if (useScram()) {
  props.put("security.protocol", "SASL_SSL");
  props.put("sasl.mechanism", "SCRAM-SHA-512");
  props.put("ssl.truststore.location", "/truststore/kafka.client.truststore.jks");
  props.put("sasl.jaas.config", getScramAuthString());
}

producer = new KafkaProducer<>(props);

getscramauthstring()函数定义如下：

private static String getScramAuthString() {
    return String.format("org.apache.kafka.common.security.scram.ScramLoginModule required \nusername=\"%s\"\npassword=\"%s\";", username, password);
}

其中用户名和密码从环境变量中读取，环境变量由ECS从AWS机密管理器传入。