How to continually consume from Kafka topic using spring webflux?

user_2096986 0
Viewed 628

I have a spring-webflux handler named WordListenerHandler which I can request a HTTP GET from the browser. This request is a stream of Long using Flux: .body(Flux.interval(Duration.ofSeconds(1)).log(), Long.class) so it continually update the HTML without refreshing the page. I want to show the messages of a kafka topi instead of the stream of Long values. So I have the consumer method onMessage(ConsumerRecord<String, String> consumerRecord) which is receiving messages and I am populating a Queue<String> stringStream. My idea is to use this queue on the HTTP GET method with Flux.fromStream. But is is static and I have to refresh the browser to see the updates. How do make the HTTP GET stream continually get data?

@Component
public class WordListenerHandler {

    private final String WORDS_STREAMING_OUTPUT_TOPIC_NAME = "streaming-words-output-topic";

    Queue<String> stringStream = new LinkedList<String>();

    public Mono<ServerResponse> fluxWordCountStream(ServerRequest serverRequest) {

        return ServerResponse.ok()
                .contentType(MediaType.APPLICATION_STREAM_JSON)

                 // DOES NOT WORK. i HAVE TO REFRESH THE BROWSER
                .body(Flux.fromStream(stringStream.stream()), String.class);

               // THIS WORKS AND I CAN CONTINUALLY SEE MY BROWSER UPDATE WITHOUT REFRESHING IT
               // .body(Flux.interval(Duration.ofSeconds(1)).log(), Long.class);
    }

    @KafkaListener(topics = {WORDS_STREAMING_OUTPUT_TOPIC_NAME})
    public void onMessage(ConsumerRecord<String, String> consumerRecord) throws JsonProcessingException {
        log.info("ConsumerRecord received: {}", consumerRecord);

        String message = consumerRecord.key() + " : " + consumerRecord.value() + "\n";
        stringStream.add(message);
    }
}
project-reactor spring-kafka spring-webflux java
2 Answers

I would recommend to look into a Sink Reactor's API: https://projectreactor.io/docs/core/release/reference/#sinks

So, your @KafkaListener is going to dump received data into a shared Sinks.Many in a back-pressure manner.

And your MediaType.APPLICATION_STREAM_JSON would just use an asFlux() from that sink.

I believe your problem that stringStream.stream() from that collection doesn't give a live object, but rather a snapshot of the current state of the collection. So, that's indeed looks natural that you need to refresh request to get an update state from the collection.

user_27565470

Try using a LinkedBlockingQueue instead of a LinkedList (which is not thread-safe).

 * <p>The iterators returned by this class's {@code iterator} and
 * {@code listIterator} methods are <i>fail-fast</i>: if the list is
 * structurally modified at any time after the iterator is created, in
 * any way except through the Iterator's own {@code remove} or
 * {@code add} methods, the iterator will throw a {@link
 * ConcurrentModificationException}.  Thus, in the face of concurrent
 * modification, the iterator fails quickly and cleanly, rather than
 * risking arbitrary, non-deterministic behavior at an undetermined
 * time in the future.
user_12407630
Related