1KafkaClient {
2  org.apache.kafka.common.security.plain.PlainLoginModule required
3  # 用户管理中创建用户的用户名
4  username="XXX"
5  # 用户管理中创建用户的密码
6  password="XXX";
7};

1input {
2    stdin{}
3}
4
5output {
6   kafka {
7        bootstrap_servers => "接入点"
8        topic_id => "topic名称"
9         # 固定为 SSL
10        security_protocol => "SSL"
11        # jaas.conf 文件地址
12        jaas_path => "/home/kafka/logstash-7.10.2/jaas.conf"
13        # 配置为 client.truststore.jks 文件路径
14        ssl.truststore.location =>  "/home/kafka/logstash-7.10.2/client.truststore.jks"
15        #  固定为 bms@kafka
16        ssl.truststore.password => "bms@kafka"
17        # 配置为 kafka_client_jaas.conf 文件路径
18        ssl.keystore.location => "/home/kafka/logstash-7.10.2/client.keystore.jks"
19        # 设置为client_ssl.properties文件内ssl.keystore.password
20        ssl.keystore.password => "qKiHmaov8GNR"
21        # SSL接入点辨识算法。6.x及以上版本Logstash需要加上该参数。空值
22        ssl_endpoint_identification_algorithm => ""
23    }
24}

1./logstash -f output.conf

1KafkaClient {
2  org.apache.kafka.common.security.plain.PlainLoginModule required
3  username="XXX"
4  password="XXX";
5};

1input {
2    kafka {
3         bootstrap_servers => "接入点"
4        # topic名称
5        topics => ["logstash_test"]
6        # 固定为 SSL
7        security_protocol => "SSL"
8        # jaas.conf 文件地址
9        jaas_path => "/home/kafka/logstash-7.10.2/jaas.conf"
10        # 配置为 client.truststore.jks 文件路径
11        ssl.truststore.location =>  "/home/kafka/logstash-7.10.2/client.truststore.jks"
12        #  固定为 bms@kafka
13        ssl.truststore.password => "bms@kafka"
14        # 配置为 kafka_client_jaas.conf 文件路径
15        ssl.keystore.location => "/home/kafka/logstash-7.10.2/client.keystore.jks"
16        # 设置为client_ssl.properties文件内ssl.keystore.password
17        ssl.keystore.password => "qKiHmaov8GNR"
18
19        group_id => "logstash_group"
20        consumer_threads => 3
21        auto_offset_reset => "earliest"
22    }
23}
24
25output {
26    stdout {
27        codec => rubydebug
28    }
29}

1./logstash -f input.conf