Hi @blueb  感谢您提供的宝贵的反馈！   基本上 Flink 连接 Kafka 也是遵照常规 Java 项目使用 Kafka 的模式，您可以参考此链接了解常规 Java client 与 Kafka 连接时的主要选项。   您的非 CM 管理的 Kafka 集群若是未启用认证的话，应该属于"Unsecured"。   我在 Cloudera 的官方 Github 上找到了一个 Flink ↔ Kafka 的 demo 项目，您可以参考其中的job.properties。 另外这个项目还有连接 secure Kafka 的 demo 项目，其中有配置连接 Kafka 的部分。   您可以看到 job.properties 文件中定义了:  kafka.security.protocol=SASL_SSL   这个 SASL_SSL 的含义是: 使用 SASL/PLAIN (CDP 中的 Kafka 开启 Kerberos 认证参考此链接) 作为认证方式，并使用 SSL/TLS 作为数据传输方式(也就是除了配置了认证之外，还在CM UI中Enable TLS/SSL for Kafka Broker)。参考: Confluent 官方文档。  如果传输方式没有Enable TLS/SSL，那么 Kafka Broker 的日志 (/var/log/kafka/server.log) 中，您会看到 listeners = SASL_PLAINTEXT；如果开启了Kerberos 认证 (或LDAP、PAM等其他SASL认证) 又Enable TLS/SSL for Kafka Broker，那么您会看到 listeners = SASL_SSL。 另外，值得注意的是，您可以同时配置多个listener，也就是listeners = SASL_PLAINTEXT 和 listeners = SASL_SSL 可以同时存在。   另外此 demo 代码也有一个 YouTube 视频演示。    以上信息供您参考。 ... View more