Kafka Consumer with JSON?

is it possible to use Kafka as getting a JSON objects from a post HTTP request putting them into topic and then sending them to Consumer(Database)?

BTW here is my KafkaConfig class:

@EnableKafka
@Configuration
public class KafkaConfig {

    @Bean
    public KafkaTemplate<String, User> kafkaTemplate(){
        return new KafkaTemplate<>(producerFactory());
    }
    @Bean
    static public ProducerFactory<String,User> producerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);

        return new DefaultKafkaProducerFactory<>(config);
    }

    @Bean
    public ConsumerFactory<String,User> consumerFactory(){
        Map<String, Object> config =  new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        config.put(ConsumerConfig.GROUP_ID_CONFIG,"group_id");
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);

        return new DefaultKafkaConsumerFactory<>(config);
    }
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String,User> kafkaListenerContainerFactory(){
        ConcurrentKafkaListenerContainerFactory<String,User> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }

}

Answer

I assume you know how to create a post REST point with a spring project. Basically after you get the json input from your endpoint, you can just use the kafkaTemplate reference to send the json object to kafka. Something like this as as pseudo-code

@RestController
class ExampleController

@Autowired
private final KafkaTemplate kafkaTemplate;

@PostMapping("/anyPath")
public void post(final ObjectAsJson yourObject) {

   kafkaTemplate.doSend​(// here map your object to a Producer Record)
   // depending on your use you can return a custom success response
}

Then you can hook up a method with KafkaListener annotation to consume and write it to database.

@KafkaListener(topics = "topicName", groupId = "foo", containerFactory = "kafkaListenerContainerFactory")
public void listen(YourCustomObject message) {
    // with kafkaListenerContainerFactory it should deserialise it to your desired object and here you can just write your database insertion here
}

Also i would have look at Kafka Connect, it helps out with integrations like this you want to achieve http as source and database as sink and kafka topic in between.

Hope it is helpful.