reactor-kafka

KafkaProducer.java

@Configuration
@AllArgsConstructor
public class KafkaProducer {

    private static final String BOOTSTRAP_SERVERS = "localhost:9092";

    /**
    * Building kafka producer configurations
    * */
    @Bean
    public Map<String, Object> bootstrapConfig() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ProducerConfig.CLIENT_ID_CONFIG, "sample-producer");
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        return props;
    }

    /**
     * Creating reactive kafka Producer template with producer configurations
     * */
    @Bean
    public ReactiveKafkaProducerTemplate<Integer, String> kafkaSender() {
        return new ReactiveKafkaProducerTemplate<>(SenderOptions.create(bootstrapConfig()));
    }
}