如何从Zeppelin连接到安全的Kafka集群(“构建kafka消费者失败”)？

import java.util.Properties
import org.apache.kafka.clients.consumer.KafkaConsumer

val props = new Properties()
props.put("bootstrap.servers", "kafka集群的地址") // 替换为实际的Kafka集群地址
props.put("group.id", "消费者组ID") // 替换为实际的消费者组ID
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")

val consumer = new KafkaConsumer[String, String](props)
consumer.subscribe(java.util.Collections.singletonList("要消费的主题")) // 替换为实际的要消费的主题

while (true) {
  val records = consumer.poll(100)
  for (record <- records.asScala) {
    println(record.value())
  }
}

props.put("security.protocol", "SSL")
props.put("ssl.truststore.location", "SSL证书路径")
props.put("ssl.truststore.password", "证书密码")
props.put("ssl.keystore.location", "SSL密钥路径")
props.put("ssl.keystore.password", "密钥密码")
props.put("ssl.key.password", "密钥密码")
props.put("sasl.mechanism", "PLAIN")
props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"用户名\" password=\"密码\";")