Test a Reactive-Kafka Consumer and Producer Template using embedded kafka + custom serialised

Question

We need an example on how to test ReactiveKafkaConsumerTemplate and ReactiveKafkaProducerTemplate with an embedded-kafka-broker. Thanks.

CORRECT CODE IS HERE AFTR DISCUSSION

You can have your custom de-serializer accordingly to use custom ReactiveKafkaConsumerTemplate

Custom Serialiser:

import org.apache.kafka.common.serialization.Serializer;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;

public class EmployeeSerializer implements Serializer<Employee> {

    @Override
    public byte[] serialize(String topic, Employee data) {
        
        byte[] rb = null;
        ObjectMapper mapper = new ObjectMapper();
        try {
            rb = mapper.writeValueAsString(data).getBytes();
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        return rb;
    }

}

Use it part of embedded-kfka-reactive test:

import java.util.Map;

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.connect.json.JsonSerializer;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.kafka.core.reactive.ReactiveKafkaProducerTemplate;
import org.springframework.kafka.support.converter.MessagingMessageConverter;
import org.springframework.kafka.test.condition.EmbeddedKafkaCondition;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.kafka.test.utils.KafkaTestUtils;

import reactor.kafka.sender.SenderOptions;
import reactor.kafka.sender.SenderRecord;
import reactor.test.StepVerifier;

@EmbeddedKafka(topics = EmbeddedKafkareactiveTest.REACTIVE_INT_KEY_TOPIC,
brokerProperties = { "transaction.state.log.replication.factor=1", "transaction.state.log.min.isr=1" })
public class EmbeddedKafkareactiveTest {

    public static final String REACTIVE_INT_KEY_TOPIC = "reactive_int_key_topic";

    private static final Integer DEFAULT_KEY = 1;

    private static final String DEFAULT_VERIFY_TIMEOUT = null;

    private ReactiveKafkaProducerTemplate<Integer, Employee> reactiveKafkaProducerTemplate;

    @BeforeEach
    public void setUp() {
        reactiveKafkaProducerTemplate = new ReactiveKafkaProducerTemplate<>(setupSenderOptionsWithDefaultTopic(),
                new MessagingMessageConverter());
    }

    private SenderOptions<Integer, Employee> setupSenderOptionsWithDefaultTopic() {
        Map<String, Object> senderProps = KafkaTestUtils
                .producerProps(EmbeddedKafkaCondition.getBroker().getBrokersAsString());
        SenderOptions<Integer, Employee> senderOptions = SenderOptions.create(senderProps);
        senderOptions = senderOptions.producerProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "reactive.transaction")
                .producerProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true)
                .producerProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class.getName())
                ;
        return senderOptions;
    }

    @Test
    public void test_When_Publish() {
        
        
        Employee employee = new Employee();
        
        ProducerRecord<Integer, Employee> producerRecord = new ProducerRecord<Integer, Employee>(REACTIVE_INT_KEY_TOPIC, DEFAULT_KEY, employee);
                
        StepVerifier.create(reactiveKafkaProducerTemplate.send(producerRecord)
                .then())
                .expectComplete()
                .verify();
    }   

    @AfterEach
    public void tearDown() {
        reactiveKafkaProducerTemplate.close();
    }
}

Answer 1

The tests in the framework use an embedded kafka broker.

https://github.com/spring-projects/spring-kafka/tree/main/spring-kafka/src/test/java/org/springframework/kafka/core/reactive

@EmbeddedKafka(topics = ReactiveKafkaProducerTemplateIntegrationTests.REACTIVE_INT_KEY_TOPIC, partitions = 2)
public class ReactiveKafkaProducerTemplateIntegrationTests {
...

Answer 2

added correct serialised with a non-transactional producer. please see the code on top of this page for the answer.