Connection types

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 compilation tool.

Install dependency libraries

Add the following dependency to the pom.xml file.

<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>2.4.0</version>
</dependency>

Prepare configurations

1. Download the SSL root certificate.

2. Create a kafka.properties file in the src/resource directory of your project and add the following content.

   Note

   Log on to the LinkedMall Open Platform to obtain the values for parameters such as bootstrap.servers, group.id, sasl.username, and sasl.password.

## SSL endpoint
bootstrap.servers=xxxx
## Group
group.id=xxxx
## SASL username
sasl.username=12345
## SASL password
sasl.password=12345
## The absolute path of the SSL root certificate file that you downloaded in Step 1.
ssl.truststore.location=/xxxx/only.4096.client.truststore.jks

3. Create a configuration file loader.

import java.util.Properties;

public class KafkaConfigurer {
    private static volatile Properties properties;

    public static Properties getKafkaProperties() {
        if (properties == null) {
            synchronized (KafkaConfigurer.class) {
                if (properties == null) {
                    // Obtain the content of the kafka.properties configuration file.
                    Properties kafkaProperties = new Properties();
                    try {
                        kafkaProperties.load(KafkaConfigurer.class.getClassLoader().getResourceAsStream("kafka.properties"));
                    } catch (Exception e) {
                        // If the file fails to load, the program must exit.
                        e.printStackTrace();
                    }
                    properties = kafkaProperties;
                }
            }
        }
        return properties;
    }
}

Consume messages

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;
import org.apache.kafka.common.config.SslConfigs;

import java.io.IOException;
import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.*;
import java.util.stream.Collectors;


public class LinkedmallConsumerExample {

    private static final String JAAS_CONFIG_TEMPLATE = "org.apache.kafka.common.security.plain.PlainLoginModule required" +
            " username=\"%s\"" +
            " password=\"%s\";";

    public static void main(String[] args) throws IOException {

        // Load kafka.properties.
        Properties kafkaProperties = KafkaConfigurer.getKafkaProperties();
        Properties props = new Properties();

        // Set the endpoint and authentication configurations.
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
        props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaProperties.getProperty("ssl.truststore.location"));
        props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
        props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
        String jaasConfig = String.format(JAAS_CONFIG_TEMPLATE, kafkaProperties.get("sasl.username"), kafkaProperties.get("sasl.password"));
        props.put(SaslConfigs.SASL_JAAS_CONFIG, jaasConfig);
        props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");

        // Set the group.
        props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getProperty("group.id"));

        // The maximum allowed interval between two polls.
        // If a consumer does not send a heartbeat within this interval, the server-side considers the consumer inactive. The server-side then removes the consumer from the Group and triggers a rebalancing. The default value is 30000.
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);

        // The maximum number of messages to poll each time.
        // Do not set this value too high. Consider the consumption rate of the consumer. If too much data is polled and cannot be consumed before the next poll, load balancing is triggered, which may cause stuttering.
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 30);

        // The number of bytes to pull at a time. We recommend that you set this value to the maximum number of messages per poll multiplied by 1024. A LinkedMall message is about 1 KB. A value that is too large may trigger throttling.
        props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 32000);
        props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 32000);

        // The deserialization method for messages.
        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");

        // We recommend that you disable autocommit.
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);


        // Construct a message object, which generates a consumer instance.
        KafkaConsumer<String, String> consumer = new org.apache.kafka.clients.consumer.KafkaConsumer<>(props);

        // Set the subscribed topics.
        List<String> topics = Arrays.stream(kafkaProperties.getProperty("topics").split(",")).collect(Collectors.toList());
        consumer.subscribe(topics);

        // A thread pool for asynchronous message processing.
        ExecutorService executorService = new ThreadPoolExecutor(0,
                Math.max(1, Runtime.getRuntime().availableProcessors() - 1),
                60L, TimeUnit.SECONDS,
                new SynchronousQueue<>(), Executors.defaultThreadFactory());
        // Consume messages in a loop.
        while (true) {
            try {
                ConsumerRecords<String, String> records = consumer.poll(Duration.of(1, ChronoUnit.SECONDS));
                System.out.printf("Record pulled: %d\n", records.count());

                // Process messages serially by partition to ensure order.
                Collection<Callable<Void>> tasks = records.partitions().stream()
                        // Get messages within a single partition.
                        .map(records::records)
                        // Create logic for serial consumption of messages within each partition.
                        .map(partitionRecords -> (Callable<Void>) () -> {
                            for (ConsumerRecord<String, String> record : partitionRecords) {
                                System.out.printf("Consume topic:%s partition:%d offset:%d%n", record.topic(), record.partition(), record.offset());

                                consume(record);
                            }
                            return null;
                        })
                        .collect(Collectors.toList());

                // Wait for all partition messages to be consumed. The timeout is 25s. The wait time must be less than SESSION_TIMEOUT_MS_CONFIG. Otherwise, rebalancing is triggered.
                List<Future<Void>> result = executorService.invokeAll(tasks, 25, TimeUnit.SECONDS);
                for (Future<Void> future : result) {
                    future.get();
                }

                // After all messages are successfully consumed, manually commit the offset.
                consumer.commitSync();
            } catch (Exception e) {
                // If consumption fails, record the reason.
                try {
                    Thread.sleep(1000);
                } catch (Throwable ignore) {
                }
                e.printStackTrace();
            }
        }
    }

    private static void consume(ConsumerRecord<String, String> record) {
        int retryTime = 10;
        Exception finalException = null;
        // Manually catch exceptions and retry.
        for (int i = 0; i < retryTime; i++) {
            try {
                doConsume(record);
                return;
            } catch (Exception e) {
                finalException = e;
            }
        }
        // Record the cause of the exception.
        finalException.printStackTrace();
    }

    private static void doConsume(ConsumerRecord<String, String> record) {
        // Message business processing logic.
    }
}

Notes

1. Consumption retry

The Kafka client does not provide a built-in retry mechanism. You can catch exceptions and implement your own retry logic, or use a component such as spring-kafka to implement a retry mechanism.

• Note: You must differentiate between retryable and non-retryable failures. For non-retryable messages, such as those that fail due to unexpected exceptions, do not block consumption. Instead, record these messages using logs or other methods for later analysis and correction. For retryable messages, you must set a limit on the maximum number of retries. Unlimited retries can block message processing and cause issues such as message accumulation.

• Note: If an offset is not committed, the client pulls messages from the last committed offset after a restart. This can cause duplicate message consumption. Idempotent processing is required.

2. Ordered consumption of messages

LinkedMall messages are ordered by partition. To consume messages in the correct order, you must process them serially within each partition. For an example, see the consumption section in the sample code.

// Pull messages.
ConsumerRecords<String, String> records = consumer.poll(Duration.of(1, ChronoUnit.SECONDS));

// Get partition information.
Set<TopicPartition> partitions = records.partitions();

for (TopicPartition partition: partitions) {
  // Get messages within a single partition.
  List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
  // You can use an asynchronous thread to process messages within a partition serially.
  consumePartition(partitionRecords);
}

3. Message types

   A topic can contain multiple event types, such as ProductCreated and SkuEdited in a product topic. You can process the events that are relevant to your business and skip the others during message processing.