Kafka-CDI - A extension for Apache Kafka

Getting started

The config is done by implementing the interface KafkaConfigProperties with qualifier @KafkaConfig. The interface is composed by three methods commonClientConfigs(), producerConfigs() and consumerConfigs(), each methods should return only the specific Kafka configuration org.apache.kafka.clients.CommonClientConfigs, org.apache.kafka.clients.producer.ProducerConfig and org.apache.kafka.clients.consumer.ConsumerConfig respectively.

@KafkaConfig
public class KafkaConfigPropertiesImpl implements KafkaConfigProperties {
    
    @Override
    public Properties commonClientConfigs() {
        Properties properties = new Properties();
        properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        properties.put(CommonClientConfigs.CLIENT_ID_CONFIG, "client.id");

        return properties;
    }
    
    
    @Override
    public Properties producerConfig() {
        Properties properties = new Properties();
        properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transactional.id");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
        properties.put(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, Long.MAX_VALUE);
        
        return properties;
    }

    @Override
    public Properties consumerProperties() {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "group.id");
        properties.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
    
        return properties;
    }

}

Creating a Kafka Consumer

The API provides the @Consumer annotation to register Kafka Consumers which internally will be used to create a vanilla KafkaConsumer.

public class MyConsumer {

    private final Logger LOG = LoggerFactory.getLogger(MyConsumer.class);

    @Consumer(topics = {"topic.one", "topic.two"}, groupId = "my-consumer-group")
    public void consume(@Value String payload) {
        LOG.info("Received message: {}", payload);
    }
}

Receive additional informations is also possible using @Key and @Header annotations to inject the record key and record header.

import org.apache.kafka.common.header.Headers;

public class MyConsumer {

    private final Logger LOG = LoggerFactory.getLogger(MyConsumer.class);

    @Consumer(topics = {"topic.one", "topic.two"}, groupId = "my-consumer-group")
    public void consume(@Value String payload, @Key String key, @Headers Headers headers) {
        LOG.info("Received message: {}, key: {}, header: {}", payload, key, headers);
    }
}

The groupId is optional if the KafkaConfigProperties.consumerConfig() method returns the ConsumerConfig.GROUP_ID_CONFIG property. If both are present, @Consumer(groupId = "any") will be used.

Deserializer

The Consumer default deserializer is org.apache.kafka.common.serialization.StringDeserializer for the key and value, both can be changed in the @Consumer annotation by any class that implements the org.apache.kafka.common.serialization.Deserializer interface.

public class MyConsumer {

    private final Logger LOG = LoggerFactory.getLogger(MyConsumer.class);

    @Consumer(topics = {"topic.one", "topic.two"}, groupId = "my-consumer-group", keyDeserializer = IntegerDeserializer.class)
    public void consume(@Value String payload, @Key Integer key) {
        LOG.info("Received message: {}", payload);
    }
}

Injecting a KafkaProducer

For better flexibility and compatibility, the API provides a properly configured vanilla KafkaProducer, without wrappers or decorators. However, if the ProducerConfig.TRANSACTIONAL_ID_CONFIG property is returned by the KafkaConfigProperties.producerConfig() method, the API will be responsible for calling the KafkaProducer.initTransactions() method before any other method.

public class MyProducer {

    @Inject
    @Producer
    private KafkaProducer<Integer, String> producer;

    public void myProducerMethod(String message) {
        producer.send(new ProducerRecord<>("kafka.topic", message));
    }
}

KafkaProducer is thread safe, so the producer is declared as @ApplicationScoped.