Using Kafka through Observable(RxJava)

user_8003665 0
Viewed 7743

I have a producer (using Kafka), and more than one consumer. So I publish a message in a topic and then my consumers receive and process the message.

I need to receive a response in the producer from at least one consumer (better if it be the first). I'm trying to use RxJava to do it (observables).

Is it possible to do in that way? Anyone have an example?

rx-java spring-cloud-stream apache-kafka
3 Answers

Here is how I am using rxjava '2.2.6' without any additional dependencies to process Kafka events:

import io.reactivex.Observable;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

...

// Load consumer props 
Properties props = new Properties();  
props.load(KafkaUtils.class.getClassLoader().getResourceAsStream("kafka-client.properties")); 

// Create a consumer
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

// Subscribe to topics
consumer.subscribe(Arrays.asList(props.getProperty("kafkaTopics").split("\\s*,\\s*")));

// Create an Observable for topic events
Observable<ConsumerRecords<String, String>> observable = Observable.fromCallable(() -> {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofSecond(10);
    return records;
});

// Process Observable events
observable.subscribe(records -> {
    if ((records != null) && (!records.isEmpty())) {
        for (ConsumerRecord<String, String> record : records) {
            System.out.println(record.offset() + ": " + record.value());
        }
    }
});
user_2567700
Related