All Products
Search

This Product

    ApsaraMQ for Kafka:Send and receive messages by using an instance endpoint

    Document Center

    ApsaraMQ for Kafka:Send and receive messages by using an instance endpoint

    Last Updated:Jul 10, 2026

    Connect your application to an ApsaraMQ for Kafka instance through a default, SSL, or SASL endpoint to send and receive messages.

    Prerequisites

    • Install JDK 1.8 or later. For more information, see Install JDK.

    • Install Maven 2.5 or later. For more information, see Install Maven.

    • Install a build tool.

      This topic uses IntelliJ IDEA Ultimate as an example.

    • Purchase and deploy an ApsaraMQ for Kafka instance.

      • VPC instance: Provides only a default endpoint accessible from within the same VPC.

      • Internet/VPC instance: Provides a default endpoint and an SSL endpoint. Accessible over the Internet or from a VPC.

      Note

    Install Java dependencies

    The following dependencies are required. They are already included in the pom.xml file in the kafka-java-demo folder, so you do not need to add them manually.

    <dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.6.0</version>
</dependency>
<dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-log4j12</artifactId>
    <version>1.7.6</version>
</dependency>
    Note

    We recommend that you keep the major version of the client library consistent with the major version of your ApsaraMQ for Kafka instance. You can find the major version of your ApsaraMQ for Kafka instance on the Instance Details page in the ApsaraMQ for Kafka console.

    Set up the configuration

    1. Optional: Download the SSL root certificate. If you use an SSL endpoint, you must download this certificate.

    2. Go to Aliware-kafka-demos, click download, and download and decompress the demo project.

    3. Import the kafka-java-demo folder from the decompressed project into IntelliJ IDEA.

    4. Optional:If you use an SSL endpoint or a SASL endpoint to connect to the instance, you must modify the kafka_client_jaas.conf file. For more information about the instance endpoints, see Endpoint comparison.

      KafkaClient {
  org.apache.kafka.common.security.plain.PlainLoginModule required
  username="xxxx"
  password="xxxx";
};

      For a VPC instance, only resources in the same VPC can access the Kafka instance, which ensures secure and private data transmission. For higher security, you can enable the ACL feature so that messages are transmitted only after SASL authentication. You can use the PLAIN or SCRAM mechanism based on your security requirements. For more information, see Enable the ACL feature.

      For an Internet/VPC instance, messages transmitted over the Internet must be authenticated and encrypted. SASL with the PLAIN mechanism must be used over an SSL transport layer. The SASL_SSL protocol prevents plaintext transmission over the Internet.

      The username and password in the example are the SASL username and password for the instance.

      • If the ACL feature is disabled for an instance that is accessible over the Internet, you can obtain the username and password of the default user from the Configuration Information section on the Instance Details page of the ApsaraMQ for Kafka console.

      • If the ACL feature is enabled for the instance, make sure that the SASL user is of the PLAIN type and is granted the permissions to send and receive messages. For more information, see Grant permissions to a SASL user.

    5. Modify the kafka.properties configuration file.

      ##==============================Common parameters==============================
bootstrap.servers=xxxxxxxxxxxxxxxxxxxxx
topic=xxx
group.id=xxx
##=======================Configure the following parameters based on your requirements========================
##SSL endpoint configuration
ssl.truststore.location=/xxxx/only.4096.client.truststore.jks
##The value of ssl.truststore.password must be KafkaOnsClient and cannot be changed. 
ssl.truststore.password=KafkaOnsClient
##Hostname verification algorithm. Keep this parameter empty. Do not change the setting.
ssl.endpoint.identification.algorithm=
java.security.auth.login.config=/xxxx/kafka_client_jaas.conf
##SASL endpoint with the PLAIN mechanism
java.security.auth.login.config.plain=/xxxx/kafka_client_jaas_plain.conf
##SASL endpoint with the SCRAM mechanism
java.security.auth.login.config.scram=/xxxx/kafka_client_jaas_scram.conf

      Parameter

      Description

      bootstrap.servers

      The endpoint. You can obtain the endpoint from the Endpoint Information section on the Instance Details page of the ApsaraMQ for Kafka console.

      topic

      The name of the topic. You can obtain the topic name on the Topics page of the ApsaraMQ for Kafka console.

      group.id

      The Group of the instance. You can obtain it on the Groups page of the ApsaraMQ for Kafka console.

      Note

      This parameter is optional for producers but required for consumers.

      ssl.truststore.location

      The local path of the downloaded SSL root certificate. Replace xxxx with your actual path. Example: /home/ssl/only.4096.client.truststore.jks.

      Important

      This parameter is not required if you use a default endpoint or a SASL endpoint. This parameter is required if you use an SSL endpoint.

      ssl.truststore.password

      The truststore password. The value is fixed at KafkaOnsClient. Do not change it.

      ssl.endpoint.identification.algorithm

      The hostname verification algorithm. Leave this empty to disable hostname verification.

      java.security.auth.login.config

      The path of the JAAS configuration file. Save the kafka_client_jaas.conf file from the demo project to a local directory and replace xxxx with the actual path. Example: /home/ssl/kafka_client_jaas.conf.

      Important

      This parameter is not required if you use a default endpoint. This parameter is required if you use an SSL endpoint or a SASL endpoint.

    Send messages

    Compile and run KafkaProducerDemo.java to send messages.

    import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.Future;
// If you use an SSL or SASL endpoint, comment out the following line.
import java.util.concurrent.TimeUnit;
import org.apache.kafka.clients.CommonClientConfigs;
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;
/*
* If you use an SSL or SASL endpoint, uncomment the following two lines.
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
*/

public class KafkaProducerDemo {

    public static void main(String args[]) {
          
       /*
        * If you use an SSL endpoint, uncomment the following line.
        * Set the path of the JAAS configuration file.
        JavaKafkaConfigurer.configureSasl();
        */
         
       /*
        * If you use a SASL endpoint with the PLAIN mechanism, uncomment the following line.
        * Set the path of the JAAS configuration file.
        JavaKafkaConfigurer.configureSaslPlain();
        */
       
       /*
        * If you use a SASL endpoint with the SCRAM mechanism, uncomment the following line.
        * Set the path of the JAAS configuration file.
        JavaKafkaConfigurer.configureSaslScram();
        */

        // Load kafka.properties.
        Properties kafkaProperties =  JavaKafkaConfigurer.getKafkaProperties();

        Properties props = new Properties();
        // Set the endpoint. Obtain the instance endpoint from the console.
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
         
       /*
        * If you use an SSL endpoint, uncomment the following four lines.
        * Similar to the SASL path, this file cannot be packaged into a JAR file.
        props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaProperties.getProperty("ssl.truststore.location"));
        * The password for the truststore. Do not change this value.
        props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");
        * The security protocol. For SSL connections, this must be SASL_SSL.
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
        * The SASL authentication method. Do not change this value.
        props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
        */

       /*
        * If you use a SASL endpoint with the PLAIN mechanism, uncomment the following two lines.
        * The security protocol.
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
        * The PLAIN mechanism.
        props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
        */

       /*
        * If you use a SASL endpoint with the SCRAM mechanism, uncomment the following two lines.
        * The security protocol.
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
        * The SCRAM mechanism.
        props.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-256");
        */

        // The serializers for message keys and values.
        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");
        // The maximum wait time for a request.
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 30 * 1000);
        // The number of client-side retries.
        props.put(ProducerConfig.RETRIES_CONFIG, 5);
        // The client-side retry interval.
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 3000);
         
       /*
        * If you use an SSL endpoint, uncomment the following line.
        * An empty value disables hostname verification.
        props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");
        */

        // Create a producer instance. The producer is thread-safe. A single producer instance is generally sufficient for one process.
        // To improve performance, you can create more producer instances, but no more than five is recommended.
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);

        // Create a Kafka message.
        String topic = kafkaProperties.getProperty("topic"); // The topic to send messages to. You must create this topic in the console first.
        String value = "this is the message's value"; // The content of the message.

        try {
            // Sending messages in a batch and collecting the futures can improve performance, but do not make the batch size too large.
            List<Future<RecordMetadata>> futures = new ArrayList<Future<RecordMetadata>>(128);
            for (int i =0; i < 100; i++) {
                // Send the message and obtain 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) {
                // Synchronously obtain the result of the Future object.
                try {
                    RecordMetadata recordMetadata = future.get();
                    System.out.println("Produce ok:" + recordMetadata.toString());
                } catch (Throwable t) {
                    t.printStackTrace();
                }
            }
        } catch (Exception e) {
            // If the message fails to be sent after client-side retries, your application must handle this error.
            System.out.println("error occurred");
            e.printStackTrace();
        }
    }
}

    Subscribe to messages

    Choose one of the following methods to subscribe to messages.

    Single-consumer subscription

    Compile and run KafkaConsumerDemo.java to receive messages.

    import java.util.ArrayList;
import java.util.List;
import java.util.Properties;


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;
/*
* If you use an SSL endpoint, uncomment the following three lines. If you use a SASL endpoint, uncomment the first two lines.
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
*/

public class KafkaConsumerDemo {

    public static void main(String args[]) {

        // Set the path of the JAAS configuration file.
        /*
         * If you use an SSL endpoint, uncomment the following line.
        JavaKafkaConfigurer.configureSasl();
         */
                        
        /*
         * If you use a SASL endpoint with the PLAIN mechanism, uncomment the following line.
        JavaKafkaConfigurer.configureSaslPlain();
         */
                        
        /*
        * If you use a SASL endpoint with the SCRAM mechanism, uncomment the following line.
        JavaKafkaConfigurer.configureSaslScram();
        */

        // Load kafka.properties.
        Properties kafkaProperties =  JavaKafkaConfigurer.getKafkaProperties();

        Properties props = new Properties();
        // Set the endpoint. Obtain the instance endpoint from the console.
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));

        // If you use an SSL endpoint, comment out the following line.
        // Adjust this value based on the amount of data to pull and the client version. Default value: 30s.
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
        /*
         * If you use an SSL endpoint, uncomment the following six lines.
         * The path to the SSL truststore file. Ensure this path is correctly configured in your kafka.properties file.
         * Similar to the SASL path, this file cannot be packaged into a JAR file.
         props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaProperties.getProperty("ssl.truststore.location"));
         * The password for the truststore. Do not change this value.
         props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");
         * The security protocol. For SSL connections, this must be SASL_SSL.
         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
         * The SASL authentication method. Do not change this value.
         props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
         * The maximum interval allowed between two polls.
         * If a consumer fails to send a heartbeat within this interval, the broker considers the consumer inactive, removes it from the consumer group, and triggers a rebalance. Default value: 30s.
         props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
         * The amount of data to fetch per request. This parameter can significantly affect performance when accessing the instance over the Internet.
         props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 32000);
         props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 32000);
         */

        // If you use a SASL endpoint with the PLAIN mechanism, comment out the following line.
        // Adjust this value based on the amount of data to pull and the client version. Default value: 30s.
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
        /*
         * If you use a SASL endpoint with the PLAIN mechanism, uncomment the following three lines.
         * The security protocol.
         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
         * The PLAIN mechanism.
         props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
         * The maximum interval allowed between two polls.
         * If a consumer fails to send a heartbeat within this interval, the broker considers the consumer inactive, removes it from the consumer group, and triggers a rebalance. Default value: 30s.
         props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
         */

        // If you use a SASL endpoint with the SCRAM mechanism, comment out the following line.
        // Adjust this value based on the amount of data to pull and the client version. Default value: 30s
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
        /*
         * If you use a SASL endpoint with the SCRAM mechanism, uncomment the following four lines.
         * The security protocol.
         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
         * The SCRAM mechanism.
         props.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-256");
         * The maximum interval allowed between two polls.
         * If a consumer fails to send a heartbeat within this interval, the broker considers the consumer inactive, removes it from the consumer group, and triggers a rebalance. Default value: 30s.
         props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
         */

        // The maximum number of records to return in a single call to poll().
        // Do not set this value to a large number. If you poll a large amount of data but cannot consume the data before the next poll, a rebalance is triggered, which may cause lags.
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 30);
        // The deserializers for message keys and values.
        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. You must create the consumer group in the console first.
        // Consumer instances in the same group consume messages in a load-balanced manner.
        props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getProperty("group.id"));
        
        // If you use an SSL endpoint, uncomment the following line.
        // An empty value disables hostname verification.
        //props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");

        // Create a consumer instance.
        KafkaConsumer<String, String> consumer = new org.apache.kafka.clients.consumer.KafkaConsumer<String, String>(props);
        // Subscribe to topics. You can subscribe to multiple topics.
        // If the consumer instances have the same GROUP_ID_CONFIG value, we recommend that you configure them to subscribe to the same topics.
        List<String> subscribedTopics =  new ArrayList<String>();
        
        // If you use an SSL endpoint, comment out the first five lines and uncomment the sixth line.
        // If you want to subscribe to multiple topics, add them here.
        // You must create each topic in the console first.
        String topicStr = kafkaProperties.getProperty("topic");
        String[] topics = topicStr.split(",");
        for (String topic: topics) {
            subscribedTopics.add(topic.trim());
        }
        //subscribedTopics.add(kafkaProperties.getProperty("topic"));
        consumer.subscribe(subscribedTopics);

        // Consume messages in a loop.
        while (true){
            try {
                ConsumerRecords<String, String> records = consumer.poll(1000);
                // The polled data must be consumed before the next poll. The total time cannot exceed the value of SESSION_TIMEOUT_MS_CONFIG.
                // We recommend that you create a separate thread pool to consume messages and asynchronously return the results.
                for (ConsumerRecord<String, String> record : records) {
                    System.out.println(String.format("Consume partition:%d offset:%d", record.partition(), record.offset()));
                }
            } catch (Exception e) {
                try {
                    Thread.sleep(1000);
                } catch (Throwable ignore) {

                }
          
                e.printStackTrace();
            }
        }
    }
}

    Multi-consumer subscription

    Compile and run KafkaMultiConsumerDemo.java to consume messages.

    import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;
// If you use an SSL or SASL endpoint, uncomment the following line.

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;
/*
* If you use an SSL endpoint, uncomment the following three lines. If you use a SASL endpoint, uncomment the first two lines.
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
*/
import org.apache.kafka.common.errors.WakeupException;

/**
 * This demo shows how to start multiple consumers in a single process to consume messages from a topic at the same time.
 * Make sure that the total number of consumers does not exceed the total number of partitions for the subscribed topic.
 */
public class KafkaMultiConsumerDemo {

    public static void main(String args[]) throws InterruptedException {
        
        // Set the path of the JAAS configuration file.
        /* 
         * If you use an SSL endpoint, uncomment the following line.
         JavaKafkaConfigurer.configureSasl();
         */
                            
        /* 
         * If you use a SASL endpoint with the PLAIN mechanism, uncomment the following line.
         JavaKafkaConfigurer.configureSaslPlain(); 
         */
                            
        /* 
         * If you use a SASL endpoint with the SCRAM mechanism, uncomment the following line.
         JavaKafkaConfigurer.configureSaslScram();
         */


        // Load kafka.properties.
        Properties kafkaProperties = JavaKafkaConfigurer.getKafkaProperties();

        Properties props = new Properties();
        // Set the endpoint. Obtain the instance endpoint from the console.
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
        
        /*
         * If you use an SSL endpoint, uncomment the following four lines.
         * Similar to the SASL path, this file cannot be packaged into a JAR file.
         props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaProperties.getProperty("ssl.truststore.location"));
         * The password for the truststore. Do not change this value.
         props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");
         * The security protocol. For SSL connections, this must be SASL_SSL.
         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
         * The SASL authentication method. Do not change this value.
         props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
         */
        
        /*
         * If you use a SASL endpoint with the PLAIN mechanism, uncomment the following two lines.
         * The security protocol.
         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
         * The PLAIN mechanism.
         props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
         */

        /* 
         * If you use a SASL endpoint with the SCRAM mechanism, uncomment the following two lines.
         * The security protocol.
         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
         * The SCRAM mechanism.
         props.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-256");
         */

        // The maximum interval allowed between two polls.
        // If a consumer fails to send a heartbeat within this interval, the broker considers the consumer inactive, removes it from the consumer group, and triggers a rebalance. Default value: 30s.
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
        // The maximum number of records to return in a single call to poll().
        // Do not set this value to a large number. If you poll a large amount of data but cannot consume the data before the next poll, a rebalance is triggered, which may cause lags.
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 30);
        // The deserializers for message keys and values.
        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. You must create the consumer group in the console first.
        // Consumer instances in the same group consume messages in a load-balanced manner.
        props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getProperty("group.id"));

        /* 
         * If you use an SSL endpoint, uncomment the following line.
         * An empty value disables hostname verification.
         props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");
         */

        int consumerNum = 2;
        Thread[] consumerThreads = new Thread[consumerNum];
        for (int i = 0; i < consumerNum; i++) {
            KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

            List<String> subscribedTopics = new ArrayList<String>();
            subscribedTopics.add(kafkaProperties.getProperty("topic"));
            consumer.subscribe(subscribedTopics);

            KafkaConsumerRunner kafkaConsumerRunner = new KafkaConsumerRunner(consumer);
            consumerThreads[i] = new Thread(kafkaConsumerRunner);
        }

        for (int i = 0; i < consumerNum; i++) {
            consumerThreads[i].start();
        }

        for (int i = 0; i < consumerNum; i++) {
            consumerThreads[i].join();
        }
    }

    static class KafkaConsumerRunner implements Runnable {
        private final AtomicBoolean closed = new AtomicBoolean(false);
        private final KafkaConsumer consumer;

        KafkaConsumerRunner(KafkaConsumer consumer) {
            this.consumer = consumer;
        }

        @Override
        public void run() {
            try {
                while (!closed.get()) {
                    try {
                        ConsumerRecords<String, String> records = consumer.poll(1000);
                        // The polled data must be consumed before the next poll. The total time cannot exceed the value of SESSION_TIMEOUT_MS_CONFIG.
                        for (ConsumerRecord<String, String> record : records) {
                            System.out.println(String.format("Thread:%s Consume partition:%d offset:%d", Thread.currentThread().getName(), record.partition(), record.offset()));
                        }
                    } catch (Exception e) {
                        try {
                            Thread.sleep(1000);
                        } catch (Throwable ignore) {

                        }
                        e.printStackTrace();
                    }
                }
            } catch (WakeupException e) {
                // Ignore the exception if the consumer is closed.
                if (!closed.get()) {
                    throw e;
                }
            } finally {
                consumer.close();
            }
        }
        // A shutdown hook that can be called by another thread.
        public void shutdown() {
            closed.set(true);
            consumer.wakeup();
        }
    }
}

    FAQ

    Configure the SASL_SSL certificate

    Download the SSL certificate from the URL in Step 1 of the "Set up the configuration" section. Then, in your kafka.properties file, set ssl.truststore.location to the local path of the certificate.

    Use a custom SSL certificate

    No. You must use the SSL certificate provided by ApsaraMQ for Kafka.

    References