Message Queue for Kafka
Message Queue for Kafka
Document
- Documentation
Message Queue for KafkaSDK ReferenceApache Kafka SDKJava client SDKSASL_PLAINTEXT access point sends and receives messages using SCRAM mechanism
Copy page
Download PDFJava client SDK
SASL_PLAINTEXT access point sends and receives messages using SCRAM mechanism
Copy pageDownload PDFSASL_PLAINTEXT access point sends and receives messages using SCRAM mechanism
This topic uses the Java client as an example to describe how to connect to Message Queue for Kafka in a VPC or public network environment via a SASL_PLAINTEXT access point using the SCRAM mechanism. It also describes how to send and receive messages.
Prerequisites
1. Install Java dependencies
Add the relevant dependencies in the Java project’s pom.xml. Here is an example of Kafka version 2.2.2.
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.2.2</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.7.6</version>
</dependency>
2 Add configuration file
- Create the Message Queue for Kafka configuration file config.properties.
- For detailed descriptions of the configuration file fields, refer to configuration file.
- When accessing via the SASL_PLAINTEXT endpoint SCRAM mechanism, the configuration file example is as follows.
tip
Under SCRAM mechanism, use a SCRAM user with corresponding Topic access rights for SASL authentication. For how to obtain the username and password, refer to 2 Collect Connection Information.
- bootstrap.servers=xxxxxsecurity.protocol=SASL_PLAINTEXTtopic=my-topicconsumer.group.id=testconsumer.auto.offset.reset=earliestconsumer.enable.auto.commit=falseclient.dns.lookup=use_all_dns_ipssasl.mechanism=SCRAM-SHA-256sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="xxxx" password="xxxxx";
- Create the configuration file loader KafkaConfigurer.java.
- package com.volcengine.openservice.kafka;import java.io.FileInputStream;import java.io.InputStream;import java.util.Properties;public class KafkaConfigurer {private static Properties properties;public synchronized static Properties getKafkaProperties(String path) {if (null != properties) {return properties;}//Get the contents of the configuration file config.propertiesProperties kafkaProperties = new Properties();try {FileInputStream conf = new FileInputStream(path);kafkaProperties.load(conf);} catch (Exception e) {//The file was not loaded, the program should consider exitinge.printStackTrace();}properties = kafkaProperties;return kafkaProperties;}}
3 Produce messages
implementation method
- Create message production program ProducerDemo.java.
- Compile and run ProducerDemo.java to produce messages.
- check the running results.
- the running result example is as follows.
- Produce ok:sasl-0@0Produce ok:sasl-0@1Produce ok:sasl-0@2Produce ok:sasl-0@3
tip
Message Queue for Kafka provides an example project Demo for quick integration. After downloading and extracting the Demo, you can directly run the following commands to send messages.
java -cp kafka-demo.jar com.volcengine.openservice.kafka.ProducerDemo ./config.properties
Example code
The following example code demonstrates how to produce messages via the SASL_PLAINTEXT endpoint. You may also refer to the example file in the demo {DemoPath}/src/main/java/com/volcengine/openservice/kafka/ProducerDemo.java to implement related business logic.
package com.volcengine.openservice.kafka;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
class Producer {
// Topics used by producers
private static String topic;
// Configuration used by the producer
private static Properties props = new Properties();
private static KafkaProducer<String, String> producer;
// Constructing a Producer
public Producer(Properties kafkaProperties) {
setProps(kafkaProperties);
setSasl(kafkaProperties);
setTopic(kafkaProperties);
newProducer();
}
// Set the generated topic. Please fill in here after applying in the console.
private void setTopic(Properties kafkaProperties) {
topic = kafkaProperties.getProperty("topic");
}
// Set the security protocol. Kafka supports SASL_PLAINTEXT and PLAINTEXT protocols.
private void setSasl(Properties kafkaProperties) {
String protocol = kafkaProperties.getProperty("security.protocol");
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, protocol);
// If the security protocol is PLAINTEXT, you do not need to fill in the username and password, and return directly
if (protocol.equals("PLAINTEXT")) {
return;
}
// If the security protocol is SASL_PLAINTEXT, you need to obtain the encryption type and sasl config
if (protocol.equals("SASL_PLAINTEXT")) {
props.put(SaslConfigs.SASL_MECHANISM, kafkaProperties.getProperty("sasl.mechanism"));
props.put(SaslConfigs.SASL_JAAS_CONFIG, kafkaProperties.getProperty("sasl.jaas.config"));
return;
}
throw new IllegalArgumentException("security.protocol is not correct");
}
// Set the producer startup parameters
private void setProps(Properties kafkaProperties) {
//Set the access point. Please obtain the access point of the corresponding topic through the console.
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
//Kafka message serialization method
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
//Maximum waiting time for a request
props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 30 * 1000);
//Set the number of internal retries on the client
props.put(ProducerConfig.RETRIES_CONFIG, 5);
//Set the client internal retry interval
props.put(ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG, 3000);
//Set the client's DNS policy
props.put(ProducerConfig.CLIENT_DNS_LOOKUP_CONFIG, kafkaProperties.getProperty("client.dns.lookup"));
}
// Construct a producer object, that is, generate a production instance
private void newProducer() {
// Construct a Producer object. Note that this object is thread-safe. Generally speaking, only one Producer object is required in one process.
// If you want to improve performance, you can construct a few more objects, but not too many, preferably no more than 5
producer = new KafkaProducer<String, String>(props);
}
public void Produce() {
//Constructing a Kafka message
String value = "this is demo message "; //Message content
//Batch fetching futures can speed up the process, but be careful not to make the batch too large.
List<Future<RecordMetadata>> futures = new ArrayList<Future<RecordMetadata>>(128);
for (int i =0; i < 100; i++) {
//Send a message and get a Future object
ProducerRecord<String, String> kafkaMessage = new ProducerRecord<String, String>(topic, value + ": " + i);
Future<RecordMetadata> metadataFuture = producer.send(kafkaMessage);
futures.add(metadataFuture);
}
producer.flush();
for (Future<RecordMetadata> future: futures) {
//Get the result of the Future object synchronously
try {
RecordMetadata recordMetadata = future.get();
System.out.println("Produce ok:" + recordMetadata.toString());
} catch (Throwable t) {
t.printStackTrace();
}
}
}
}
public class ProducerDemo {
public static void main(String args[]) {
if (args.length < 1) {
System.out.println("please input config path");
return;
}
// Get the producer object
Producer producer = new Producer(KafkaConfigurer.getKafkaProperties(args[0]));
try {
// Produce messages
producer.Produce();
} catch (Exception e) {
// After the client has retried internally, the sending still fails. The business needs to deal with this type of error.
System.out.println("error occurred");
e.printStackTrace();
}
}
}
4 Consume messages
implementation method
- Create a consumer subscription program ConsumerDemo.java.
- Compile and run ConsumerDemo.java to consume messages.
- check the running results.
- the running result example is as follows.
- Consume: ConsumerRecord(topic = sasl, patition = 6, leaderEpoch=0, offset=0, CreateTime = 1637207092476, serialized key size = -1, serialized key size = 24, headers = ReacordHeaders(headers = [], isReadOnly = false), key = null, value = this is demo, message : 0)Consume: ConsumerRecord(topic = sasl, patition = 6, leaderEpoch=0, offset=0, CreateTime = 1637207092476, serialized key size = -1, serialized key size = 24, headers = ReacordHeaders(headers = [], isReadOnly = false), key = null, value = this is demo, message : 1)
tip
Message Queue for Kafka provides a example project Demo for quick integration. After downloading and extracting the demo, you can directly execute the following commands to consume messages.
java -cp kafka-demo.jar com.volcengine.openservice.kafka.ConsumerDemo ./config.properties
Example code
The following example code demonstrates how to consume messages through a SASL_PLAINTEXT access point. You can also refer to the sample file in the Demo {DemoPath}/src/main/java/com/volcengine/openservice/kafka/ConsumerDemo.java to implement related business logic.
package com.volcengine.openservice.kafka;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.config.SaslConfigs;
class Consumer {
// Consumer startup configuration
private static Properties props = new Properties();
private KafkaConsumer<String, String> consumer;
// Consumer instance constructor
public Consumer(Properties kafkaProperties) {
setProps(kafkaProperties);
setSasl(kafkaProperties);
newConsumer();
subscribed(kafkaProperties);
}
// Set the security protocol. Kafka supports SASL_PLAINTEXT and PLAINTEXT protocols.
private void setSasl(Properties kafkaProperties) {
String protocol = kafkaProperties.getProperty("security.protocol");
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, protocol);
// If the security protocol is PLAINTEXT, you do not need to fill in the username and password, and return directly
if (protocol.equals("PLAINTEXT")) {
return;
}
// If the security protocol is SASL_PLAINTEXT, you need to obtain the encryption type and sasl config
if (protocol.equals("SASL_PLAINTEXT")) {
props.put(SaslConfigs.SASL_MECHANISM, kafkaProperties.getProperty("sasl.mechanism"));
props.put(SaslConfigs.SASL_JAAS_CONFIG, kafkaProperties.getProperty("sasl.jaas.config"));
return;
}
throw new IllegalArgumentException("security.protocol is not correct");
}
// Set the consumer startup parameters
private void setProps(Properties kafkaProperties) {
//Set the access point. Please obtain the access point of the corresponding topic through the console.
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
//This value can be set according to the actual data pulled and the customer's version, etc. The default value is 30s
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
//The maximum number of polls per time
//Be careful not to set this value too large. If too much data is polled and cannot be consumed before the next poll, load balancing will be triggered, causing lag.
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 30);
//Message deserialization method
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
//The consumer group to which the current consumer instance belongs
//Consumer instances belonging to the same group will load consumption messages
props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getProperty("consumer.group.id"));
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafkaProperties.getProperty("consumer.auto.offset.reset"));
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, kafkaProperties.getProperty("consumer.enable.auto.commit"));
// Set the client's DNS policy
props.put(ConsumerConfig.CLIENT_DNS_LOOKUP_CONFIG, kafkaProperties.getProperty("client.dns.lookup"));
}
// Construct a consumer object, that is, generate a consumer instance
private void newConsumer() {
consumer = new org.apache.kafka.clients.consumer.KafkaConsumer<String, String>(props);
}
// Subscribe to topic
private void subscribed(Properties kafkaProperties) {
//Set the topic that the consumer group subscribes to
List<String> subscribedTopic = new ArrayList<String>();
//If GROUP_ID_CONFIG is the same, it is recommended to set the subscribed topics to the same
//Topic needs to be created in the console first
subscribedTopic.add(kafkaProperties.getProperty("topic"));
consumer.subscribe(subscribedTopic);
}
// Consume messages
public void poll() {
ConsumerRecords<String, String> records = consumer.poll(1000);
//This data must be consumed before the next poll, and the total time must not exceed SESSION_TIMEOUT_MS_CONFIG
//It is recommended to open a separate thread pool to consume messages and then return the results asynchronously.
for (ConsumerRecord<String, String> record : records) {
System.out.printf("Consume: %s%n", record.toString());
}
}
}
public class ConsumerDemo {
public static void main(String args[]) {
if (args.length < 1) {
System.out.println("please input config path");
return;
}
// Get the consumer object
Consumer consumer = new Consumer(KafkaConfigurer.getKafkaProperties(args[0]));
//Circular consumption message
while (true){
try {
// Consume messages
consumer.poll();
} catch (Exception e) {
try {
Thread.sleep(1000);
} catch (Throwable ignore) {
}
e.printStackTrace();
}
}
}
}
Last updated: 2025.09.18 17:51:06
All rights reserved © Beijing Volcano Engine Technology Co., Ltd2024
