Implementing Custom Interceptors

1. Understanding Interceptor Concept

Interceptors in Apache Kafka are components that allow you to hook into the producer and consumer processing pipelines to add custom logic without modifying the core Kafka client code. They provide a way to observe, modify, or even block messages as they are sent to or received from Kafka.

Interceptor Type	Description	Usage Example
Producer Interceptor	Executes before a message is sent to Kafka. Can modify or inspect messages.	`public void onSend(ProducerRecord<K, V> record)`
Consumer Interceptor	Executes after a message is received from Kafka. Can modify or inspect messages.	`public ConsumerRecord<K, V> onConsume(ConsumerRecord<K, V> record)`

Example: Basic Producer Interceptor

public class CustomProducerInterceptor<K, V> implements ProducerInterceptor<K, V> {
    @Override
    public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record) {
        // Add custom header or modify record
        return new ProducerRecord<>(record.topic(), record.key(), record.value());
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        // Handle acknowledgment or errors
    }

    @Override
    public void close() {
        // Cleanup logic
    }
}

Example: Basic Consumer Interceptor

public class CustomConsumerInterceptor<K, V> implements ConsumerInterceptor<K, V> {
    @Override
    public ConsumerRecord<K, V> onConsume(ConsumerRecord<K, V> record) {
        // Modify or inspect record
        return record;
    }

    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
        // Handle commit logic
    }

    @Override
    public void close() {
        // Cleanup logic
    }
}

Note: Interceptors are executed in the order they are specified in the configuration. Ensure proper ordering to avoid unexpected behavior.

2. Creating Producer Interceptor

Property	Description	Default Value	Notes
onSend	Method called before a record is sent to the broker	None	Must be implemented by interceptor
onAcknowledgement	Method called when a record is acknowledged by the broker	None	Optional, can be left unimplemented
close	Method called when the interceptor is closed	None	Optional, for cleanup operations

Example: Implementing a Kafka Producer Interceptor

import org.apache.kafka.clients.producer.Interceptor;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

public class CustomProducerInterceptor implements Interceptor<String, String> {
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        // Modify or log the record before sending
        System.out.println("Sending record: " + record.key() + " - " + record.value());
        return record;
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        if (exception != null) {
            System.err.println("Error sending record: " + exception.getMessage());
        } else {
            System.out.println("Record sent successfully to topic: " + metadata.topic());
        }
    }

    @Override
    public void close() {
        System.out.println("Producer interceptor closed.");
    }

    @Override
    public void configure(Map<String, ?> configs) {
        // Configuration logic if needed
    }
}

Note: Interceptors are invoked in the order they are specified in the configuration.

Warning: Avoid blocking operations in interceptor methods as they can affect producer performance.

3. Creating Consumer Interceptor

Example: Basic Consumer Interceptor Implementation

import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.util.Map;

public class LoggingConsumerInterceptor<K, V> implements ConsumerInterceptor<K, V> {
    @Override
    public void configure(Map<String, ?> configs) {
        // Configuration logic if needed
    }

    @Override
    public ConsumerRecords<K, V> onConsume(ConsumerRecords<K, V> records) {
        System.out.println("Received " + records.count() + " records");
        return records;
    }

    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
        System.out.println("Committing offsets: " + offsets);
    }

    @Override
    public void close() {
        // Cleanup logic if needed
    }
}

Method	Description	Notes
onConsume(ConsumerRecords<K, V> records)	Called before records are returned to the consumer application	Allows modification or logging of records
onCommit(Map<TopicPartition, OffsetAndMetadata> offsets)	Called when consumer commits offsets	Useful for logging or monitoring offset commits
configure(Map<String, ?> configs)	Initializes the interceptor with configuration properties	Called once during consumer initialization
close()	Called when the consumer is closed	Allows for cleanup of resources

Note: Interceptor methods are called on the consumer thread, so they should be lightweight to avoid blocking message processing.

4. Implementing onSend() Method

Method	Description	Return Type	Notes
onSend()	Invoked before a record is sent to the Kafka broker	void	Allows modification of the ProducerRecord before sending

Example: Modifying Record Before Sending

public class CustomProducerInterceptor implements ProducerInterceptor<String, String> {
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        // Add a custom header to every record
        return new ProducerRecord<>(
            record.topic(),
            record.partition(),
            record.key(),
            record.value(),
            Collections.singletonList(new Header() {
                @Override
                public String key() {
                    return "custom-header";
                }

                @Override
                public byte[] value() {
                    return "modified".getBytes();
                }
            })
        );
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        // Handle acknowledgment
    }

    @Override
    public void close() {
        // Cleanup logic
    }

    @Override
    public void configure(Map<String, ?> configs) {
        // Configuration logic
    }
}

Note: The onSend() method is called on the producer thread and must be thread-safe.

5. Implementing onConsume() Method

The onConsume() method is part of Kafka's interceptor interface and is invoked whenever a consumer receives a record from the broker. It allows you to intercept and potentially modify or log messages before they are processed by your application.

Method	Description	Parameters	Return Type
onConsume()	Invoked when a record is consumed	`ConsumerRecord<K, V> record`	`void`

Example: Implementing onConsume()

public class LoggingInterceptor<K, V> implements ConsumerInterceptor<K, V> {
    @Override
    public void onConsume(ConsumerRecord<K, V> record) {
        System.out.println("Consumed record: " + record.key() + " - " + record.value());
    }

    @Override
    public void close() {}

    @Override
    public void configure(Map<String, ?> configs) {}
}

Note: The onConsume() method runs synchronously and can impact performance if not implemented efficiently.

6. Configuring Interceptor Chain

Property	Values	Description
interceptor.classes	`org.apache.kafka.clients.producer.ProducerInterceptor`	List of interceptor class names to be added to the producer's interceptor chain
interceptor.class	`org.apache.kafka.clients.producer.ProducerInterceptor`	Single interceptor class name to be added to the producer's interceptor chain

Example: Producer Interceptor Configuration

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put("interceptor.classes", "com.example.MyProducerInterceptor");

Example: Consumer Interceptor Configuration

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put("interceptor.classes", "com.example.MyConsumerInterceptor");

Note: Interceptor classes must be available in the classpath of both producer and consumer applications.

Warning: Interceptors should not block or delay processing significantly as they run synchronously in the producer/consumer threads.

7. Adding Monitoring Metrics

Example: Custom Interceptor with Metrics

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

public class MetricsInterceptor implements ProducerInterceptor<String, String> {
    private final MeterRegistry meterRegistry;
    private Counter sentCounter;
    private Timer sendTimer;

    public MetricsInterceptor(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
        this.sentCounter = Counter.builder("kafka.producer.sent")
            .description("Number of messages sent")
            .register(meterRegistry);
        this.sendTimer = Timer.builder("kafka.producer.send.duration")
            .description("Time taken to send messages")
            .register(meterRegistry);
    }

    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        return record;
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        if (exception == null) {
            sentCounter.increment();
        }
    }

    @Override
    public void close() {
        // Cleanup logic if needed
    }
}

Property	Description	Default Value	Notes
interceptor.classes	List of interceptor classes to use	Empty list	Comma-separated class names
kafka.producer.sent	Counter for messages sent	None	Registered via MeterRegistry
kafka.producer.send.duration	Timer for send duration	None	Measures latency in milliseconds

Note: Ensure that the MeterRegistry is properly configured and accessible to interceptors for accurate metric collection.

8. Implementing Tracing

Example: Basic Tracing Interceptor

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

public class TracingInterceptor<K, V> implements ProducerInterceptor<K, V> {
    private Tracer tracer;

    @Override
    public void configure(Map<String, ?> configs) {
        tracer = TracerFactory.getTracer();
    }

    @Override
    public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record) {
        Span span = tracer.spanBuilder("send.record")
            .setAttribute("topic", record.topic())
            .startSpan();
        span.makeCurrent();
        return record;
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        Span.current().end();
    }

    @Override
    public void close() {
        // Cleanup logic if needed
    }
}

Method	Description	Notes
onSend	Called before sending a record to Kafka	Allows modification of the record or starting of tracing spans
onAcknowledgement	Called when Kafka acknowledges the record	Used to end tracing spans or handle errors
configure	Initializes the interceptor with configuration properties	Called once during interceptor setup
close	Cleans up resources when interceptor is closed	Called once before the interceptor is destroyed

Note: The tracing interceptor must be registered in the Kafka producer configuration using interceptor.classes.

Tracing Interceptor Configuration

Add interceptor.classes to producer config
Implement ProducerInterceptor interface
Use OpenTelemetry or similar tracing library

9. Adding Security Checks

Check Type Description Implementation Status

Authentication

Verifies the identity of clients before allowing access to Kafka resources.

Check Type	Description	Implementation	Status
Authentication	Verifies the identity of clients before allowing access to Kafka resources.	`public class SecurityInterceptor implements ProducerInterceptor<String, String> { @Override public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) { if (!isAuthenticated()) { throw new SecurityException("Authentication failed"); } return record; } }`	NEW
Authorization	Ensures that authenticated clients have appropriate permissions to perform actions.	`public class AuthzInterceptor implements ConsumerInterceptor<String, String> { @Override public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> records) { if (!hasPermission()) { throw new SecurityException("Authorization denied"); } return records; } }`	NEW

public class SecurityInterceptor implements ProducerInterceptor<String, String> {
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        if (!isAuthenticated()) {
            throw new SecurityException("Authentication failed");
        }
        return record;
    }
}

NEW

Authorization

Ensures that authenticated clients have appropriate permissions to perform actions.

public class AuthzInterceptor implements ConsumerInterceptor<String, String> {
    @Override
    public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> records) {
        if (!hasPermission()) {
            throw new SecurityException("Authorization denied");
        }
        return records;
    }
}

NEW

Example: Authentication Check

public boolean isAuthenticated() {
    String token = getCurrentToken();
    return token != null && validateToken(token);
}

Example: Authorization Check

public boolean hasPermission() {
    String user = getCurrentUser();
    String topic = getCurrentTopic();
    return isAuthorized(user, topic);
}

Note: Security interceptors should be used in conjunction with Kafka's built-in security mechanisms like SASL/SSL for robust protection.

10. Transforming Messages

Property	Description	Example
message.key	Key used for partitioning	`"user-123"`
message.value	Actual message content	`"Hello Kafka"`
message.headers	Metadata associated with the message	`{"content-type": "application/json"}`

Example: Modifying Message Value

public class MessageTransformer implements ConsumerInterceptor<String, String> {
    @Override
    public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> records) {
        for (ConsumerRecord<String, String> record : records) {
            String transformedValue = record.value().toUpperCase();
            // In real implementation, you'd create a new record with the transformed value
        }
        return records;
    }
}

Example: Adding Headers to Messages

public class HeaderAdder implements ProducerInterceptor<String, String> {
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        Headers headers = new RecordHeaders();
        headers.add("processed-by", "transformer-v1".getBytes());
        return new ProducerRecord<>(record.topic(), record.partition(), record.key(), record.value(), headers);
    }
}

Note: Interceptor transformations occur before the message is processed by the consumer or producer, allowing for real-time message modification.