java client 中添加 SASL 设置信息：

Java client consumer properties配置.png

注意 sasl.jaas.config 配置中的分号必不可少。

package kafka;

import java.time.Duration;

import java.util.Arrays;

import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;

import org.apache.kafka.clients.consumer.ConsumerRecords;

import org.apache.kafka.clients.consumer.KafkaConsumer;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

public class TestKafkaSasl {

private static final Logger logger = LoggerFactory.getLogger(TestKafkaSasl.class);

public static void main(String[] args) {

Properties props = new Properties();

props.put("bootstrap.servers", "localhost:9092");

props.put("group.id", "test_group");

props.put("enable.auto.commit", "true");

props.put("auto.commit.interval.ms", "1000");

props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

// sasl.jaas.config的配置, 结尾分号必不可少.

props.setProperty("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"admin\";");

props.setProperty("security.protocol", "SASL_PLAINTEXT");

props.setProperty("sasl.mechanism", "PLAIN");

@SuppressWarnings("resource")

KafkaConsumer consumer = new KafkaConsumer(props);

consumer.subscribe(Arrays.asList("test"));

while (true) {

try {

ConsumerRecords records = consumer.poll(Duration.ofMillis(1000));

for (ConsumerRecord record : records) {

System.out.printf("offset = %d, partition = %d, key = %s, value = %s%n",

record.offset(), record.partition(), record.key(), record.value());

logger.info("offset = {}, partition = {}, key = {}, value = {}",

record.offset(), record.partition(), record.key(), record.value());

}

} catch (Exception e) {

e.printStackTrace();

logger.error(e.getMessage());

}

}

}

}

运行测试：

继续生产数据.png

java client sasl消费数据测试.png