Confluent Cloud

I am trying to build a Java Spring Boot application that would post & get the messages from Confluent Cloud Kafka.

I followed the article for publishing a Kafka message into Confluent Cloud and it works.

Below is the implementation

KafkaController.java

package com.seroter.confluentboot.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import com.seroter.confluentboot.dto.Product;
import com.seroter.confluentboot.engine.Producer;

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

    private final Producer producer;
    
    private final com.seroter.confluentboot.engine.Consumer consumer;

    @Autowired
    KafkaController(Producer producer,com.seroter.confluentboot.engine.Consumer consumer) {
        this.producer = producer;
        this.consumer=consumer;
    }

    @PostMapping(value = "/publish")
    public void sendMessageToKafkaTopic(@RequestParam("message") String message) {
        this.producer.sendMessage(message);
    }
   
    
    @PostMapping(value="/publishJson")
    public ResponseEntity<Product> publishJsonMessage(@RequestBody Product product) {
        producer.sendJsonMessage(product);
        ResponseEntity<Product> responseEntity=new ResponseEntity<>(product,HttpStatus.CREATED);
        return responseEntity;
    }
    
    
}

Product.java

package com.seroter.confluentboot.dto;

import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonPropertyOrder;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

@Data
@NoArgsConstructor
@AllArgsConstructor
@JsonPropertyOrder(value = {"product_id","product_name","quantity","price"})
public class Product {

    @JsonProperty(value = "product_id")
    private int productId;
    @JsonProperty(value="product_name")
    private String productName;
    
    private int quantity;
    
    private double price;
    
}

Producer.java

package com.seroter.confluentboot.engine;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.stereotype.Service;

import com.seroter.confluentboot.dto.Product;

@Service
@EnableBinding(Source.class)
public class Producer {

    private static final Logger logger = LoggerFactory.getLogger(Producer.class);
    private static final String TOPIC = "users";

    @Autowired
    private Source source;

    public void sendMessage(String message) {
        logger.info(String.format("#### -> Producing message -> %s", message));
        this.source.output().send(new GenericMessage<>(message));
    }
    
    
    public void sendJsonMessage(Product product)
    {
        logger.info(String.format("#### -> Producing message -> %s",product.toString()));
        this.source.output().send(new GenericMessage<>(product));
    }
    
}

ConfluentBootApplication.java

package com.seroter.confluentboot;

import org.apache.tomcat.util.net.WriteBuffer.Sink;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.http.ResponseEntity;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import com.seroter.confluentboot.dto.Product;


@SpringBootApplication
@EnableBinding(Source.class)
@RestController
@RequestMapping(value = "/confluent")

public class ConfluentBootApplication {
    @Autowired
    private  com.seroter.confluentboot.engine.Consumer consumer;
    
    public static void main(String[] args) {
        SpringApplication.run(ConfluentBootApplication.class, args);
    }
    
     
}

application.properties

spring.cloud.stream.kafka.binder.brokers=pkc-epwny.eastus.azure.confluent.cloud:9092
spring.cloud.stream.bindings.output.destination=test
  
spring.cloud.stream.kafka.binder.configuration.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="user" password="password";
spring.cloud.stream.kafka.binder.configuration.sasl.mechanism=PLAIN
spring.cloud.stream.kafka.binder.configuration.security.protocol=SASL_SSL

server.port=9000

It works

and I could verify

I want to build a Spring Boot Consumer REST Endpoint? How do I do it?

Update:

ConfluentConsumer.java

package com.seroter.confluentboot.controller;

import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;

import com.seroter.confluentboot.dto.Product;

//@RestController
@EnableBinding(Sink.class)
public class ConfluentConsumer {
    
    @StreamListener(Sink.INPUT)
    public void consumeMessage(Product product)
    {
        System.out.println("******************************");
        System.out.println("============= "+product.getProductId()+" ================");
        System.out.println("******************************");
    }

}

Consumer.java

package com.seroter.confluentboot.engine;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.PropertySource;
import org.springframework.stereotype.Service;

@Service
@PropertySource("classpath:application.properties")
public class Consumer {

    private final Logger logger = LoggerFactory.getLogger(Producer.class);
}