Kafka Cluster 2019(4)Spring Boot Kafka Reactive

Kafka Cluster 2019(4)Spring Boot Kafka Reactive

Found some articles and one samples here https://github.com/vanseverk/paymentprocessor-kafka-intro/tree/master/paymentprocessor-gateway

to avoid exception, disable SerializationFeature.FAIL_ON_EMPTY_BEANS

Add dependency supports to POM, pom.xml
<dependency>
<groupId>io.projectreactor.kafka</groupId>
<artifactId>reactor-kafka</artifactId>
</dependency>

Kafka Configuration class to load the configurations from YAML
package com.sillycat.webfluxlatency.config;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.IntegerDeserializer;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import reactor.kafka.receiver.KafkaReceiver;
import reactor.kafka.receiver.ReceiverOptions;
import reactor.kafka.sender.KafkaSender;
import reactor.kafka.sender.SenderOptions;

@Slf4j
@Setter
@Getter
@Configuration
@EnableConfigurationProperties
public class KafkaConfig {
   
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;
   
    @Bean
    public KafkaSender<Integer, String> kafkaProducer() {
        final Map<String, Object> producerProps = new HashMap<>();
        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        final SenderOptions<Integer, String> producerOptions = SenderOptions.create(producerProps);
        return KafkaSender.create(producerOptions);
    }
   
    @Bean
    public KafkaReceiver<Object, Object> kafkaReceiver() {
        final Map<String, Object> consumerProps = new HashMap<>();
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        consumerProps.put(ConsumerConfig.CLIENT_ID_CONFIG, "webflux-1");
        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "webflux");
        consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);

        ReceiverOptions<Object, Object> consumerOptions = ReceiverOptions.create(consumerProps)
                .subscription(Collections.singleton("hello-webflux"))
                .addAssignListener(partitions -> log.debug("onPartitionsAssigned {}", partitions))
                .addRevokeListener(partitions -> log.debug("onPartitionsRevoked {}", partitions));

        return KafkaReceiver.create(consumerOptions);
    }

}

The implementation class which send out messages and consume messages
package com.sillycat.webfluxlatency.service;

import javax.annotation.PostConstruct;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.kafka.receiver.KafkaReceiver;
import reactor.kafka.receiver.ReceiverRecord;
import reactor.kafka.sender.KafkaSender;
import reactor.kafka.sender.SenderRecord;
import reactor.kafka.sender.SenderResult;

@Slf4j
@Service
public class KafkaServiceImpl implements KafkaService {

    @Autowired
    KafkaSender<Integer, String> kafkaSender;

    @Autowired
    KafkaReceiver<Object, Object> kafkaReceiver;

    @PostConstruct
    public void init(){
        ((Flux<ReceiverRecord<Object, Object>>) kafkaReceiver.receive()).doOnNext(r -> {
            processEvent(r.value().toString());
            r.receiverOffset().acknowledge();
        }).subscribe();
    }

    private void processEvent(String message) {
        log.info("received message:" + message);
    }

    public Mono<SenderResult<Integer>> send(final String payload) {
        log.info("send out message:" + payload);
        SenderRecord<Integer, String, Integer> message = SenderRecord
                .create(new ProducerRecord<>("hello-webflux", payload), 1);
        return kafkaSender.send(Mono.just(message)).next();
    }

}

Webflux Controller which send back MONO
package com.sillycat.webfluxlatency.web;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;

import com.sillycat.webfluxlatency.service.KafkaService;

import io.swagger.annotations.Api;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Mono;
import reactor.kafka.sender.SenderResult;

@Slf4j
@Api(value = "/kafka/")
@RestController
public class KafkaController {

    @Autowired
    KafkaService kafkaService;

    @PostMapping("/kafka/{topicName}")
    public Mono<SenderResult<Integer>> sendToTopic(@PathVariable String topicName, @RequestBody String message) {
        log.info("topicName " + topicName + " message " + message);
        return kafkaService.send(message);
    }

}

Application YAML configuration File
spring:
jackson:
  serialization:
   FAIL_ON_EMPTY_BEANS: false
kafka:
  consumer:
   group-id: mvccluster
  bootstrap-servers: ubuntu-master:9092, ubuntu-dev2:9092, ubuntu-dev4:9092

Then I can easily send out messages and consume messages.


References:
https://github.com/reactor/reactor-kafka
https://www.reactiveprogramming.be/an-introduction-to-reactor-kafka/
https://codar.club/blogs/reactor-kafka-via-spring-boot-webflux.html