Application Real-Time Monitoring Service:Implement tracing in Kafka asynchronous messaging

Spring Kafka

No additional configuration is required. If the ARMS agent and Kafka client versions meet the prerequisites, the ARMS agent automatically extracts trace context from incoming message headers.

Native Kafka client

The native Kafka Consumer API uses continuous poll requests to pull messages, which prevents the ARMS agent from directly extracting trace data from headers. To work around this, register a custom consumption method in the ARMS console.

The consumption method must accept a parameter of type org.apache.kafka.clients.consumer.ConsumerRecord or org.apache.kafka.clients.consumer.ConsumerRecords.

Example

The following example defines a handler method that the ARMS agent intercepts to extract trace context from each record:

package arms.test.kafka;

public class KafkaConsumeTest {
    public void testConsumer() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "<bootstrap-server-address>");
        props.put("group.id", UUID.randomUUID().toString());
        props.put("enable.auto.commit", "true");
        props.put("auto.offset.reset", "earliest");
        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");

        KafkaConsumer<String, String> kafkaConsumer =
            new KafkaConsumer<>(props);
        kafkaConsumer.subscribe(Arrays.asList("test"));

        while (true) {
            ConsumerRecords<String, String> records =
                kafkaConsumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                // Delegate each record to the handler method
                handler(record);
            }
        }
    }

    // ARMS agent intercepts this method to extract trace context
    public void handler(ConsumerRecord<String, String> record) {
        LOGGER.info(Utils.string(record));
    }
}

Replace the following placeholder with your actual value:

Register the consumption method

1. Log on to the ARMS console.

2. In the left-side navigation pane of the application details page, click Application Settings.

3. On the Custom Configuration tab, set Kafka custom consumption method to the fully qualified method name. Example: arms.test.kafka.KafkaConsumeTest.handler.

4. Restart the consumer application to apply the change.

For details, see Customize application settings.