Исключение в потоке «основной» org.apache.kafka.common.errors.SaslAuthenticationException: аутентификация не удалась ⇐ JAVA

[Anonymous]

Я пытаюсь использовать в своем проекте Confluent Cloud, размещенное в Azure, и у нас есть ключ и секрет API, но я не могу подключиться и не могу прочитать сообщение из темы kafka. Я перешел по ссылке: https://docs.confluent.io/cloud/current ... -keys.html, но неясно, как установить ее в Java. потребительский код, а также ошибка ниже
Exception in thread "main" org.apache.kafka.common.errors.SaslAuthenticationException: Authentication failed

Я использовал код ниже
public class HelloConsumer {
public static void main(String[] args) {
Properties props = new Properties();
props.put(ConsumerConfig.CLIENT_ID_CONFIG, AppConfigs.applicationID);
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "XXXXXXXXXX.azure.confluent.cloud:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "someid");
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
props.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"" + "apiKeyValue" + "\" password=\"" + "secreteValue" + "\";");

KafkaConsumer kafkaConsumer = new KafkaConsumer(props);
kafkaConsumer.subscribe(Arrays.asList("test-topic"));

while(true){
ConsumerRecords records = kafkaConsumer.poll(Duration.ofMillis(100));

for (ConsumerRecord record : records){
System.out.println("Key: " + record.key() + ", Value: " + record.value());
System.out.println("Partition: " + record.partition() + ", Offset:" + record.offset());
}
}
}


Подробнее здесь: https://stackoverflow.com/questions/784 ... cationexce