KafkaSASL_SSL认证错误，如何认证？

Question

我正在Java一个使用来自Kafka集群的数据的管道，它可在

https://github.com/confluentinc/confluent-google-examples/blob/master/ccloud-dataflow-demo/entry-df-pipeline/src/main/java/com/ecuevas/DataflowPipeline.java

根据存储库，定义了道具和管道：

        Map<String, Object> props = new HashMap<>();
        props.put("auto.offset.reset", "earliest");
        props.put("ssl.endpoint.identification.algorithm", "https");
        props.put("sasl.mechanism", "PLAIN");
        props.put("request.timeout.ms", 20000);
        props.put("retry.backoff.ms", 500);
        props.put("security.protocol", "SASL_SSL");
        props.put("sasl.jaas.config",String.format("org.apache.kafka.common.security.
        plain.PlainLoginModule required username=\"%s\" password=\"%s\";",username, password));

        LogKafkaMsg logKafkaMsg = new LogKafkaMsg();

        Pipeline pipeline = Pipeline.create(options);

        PCollection<KV<String, String>> entries =
                pipeline
                        .apply(
                                "Read Entries from Confluent Cloud Topic",
                                KafkaIO.<String, String>read()
                                        .withBootstrapServers("<your-bootstrap-server>")
                                        .withTopic("entries")
                                        .withConsumerConfigUpdates(props)
                                        .withKeyDeserializer(StringDeserializer.class)
                                        .withValueDeserializer(StringDeserializer.class)
                                        .withoutMetadata()
                        );

我的问题是，执行代码时，返回以下错误：

at org.apache.beam.runners.dataflow.DataflowRunner.run(DataflowRunner.java:969)
        at org.apache.beam.runners.dataflow.DataflowRunner.run(DataflowRunner.java:198)
        at org.apache.beam.sdk.Pipeline.run(Pipeline.java:322)
        at org.apache.beam.sdk.Pipeline.run(Pipeline.java:308)
        at com.ecuevas.DataflowPipeline.main(DataflowPipeline.java:214)
Caused by: org.apache.kafka.common.errors.SaslAuthenticationException: Invalid SASL/PLAIN response: expected 3 tokens, got 4

这个问题可能与SASL/普通认证有关，但是我不知道一个可能的解决方案。我认为这与:

< li>SSL认证:包括< code>ca-cert 、< code>ca-key 、< code>ca-password或其他特殊认证。 < li >依赖项更新:库版本不更新。

Answer 1

数据流虚拟机似乎没有经过身份验证，无法连接到Kafka集群。您是否尝试按照此处的说明使用密钥库证书文件进行身份验证？