Any example of reactive kafka Non-blocking Back-pressure

Question

Need java code example of reactive kafka Non-blocking Back-pressure. As per[documentation] The Reactor Kafka API benefits from non-blocking back-pressure provided by Reactor. Is there any implementation in java or java spring boot for the same for reference.

Answer 1

Here's an example:

// Import the necessary classes
import org.apache.kafka.common.serialization.StringDeserializer;
import reactor.kafka.receiver.ReceiverOptions;
import reactor.kafka.receiver.KafkaReceiver;

public class ReactiveKafkaExample {
    public static void main(String[] args) {
        // Create a receiver options object to configure the Kafka receiver
        ReceiverOptions<String, String> receiverOptions = ReceiverOptions.create(consumerProps)
                .subscription(Collections.singleton("my-topic"))
                .addAssignListener(partitions -> System.out.println("onPartitionsAssigned {}" + partitions))
                .addRevokeListener(partitions -> System.out.println("onPartitionsRevoked {}" + partitions));

        // Create a Kafka receiver using the receiver options
        KafkaReceiver<String, String> kafkaReceiver = KafkaReceiver.create(receiverOptions);

        // Use the `flatMap` operator to apply back-pressure to the receiver
        kafkaReceiver.flatMap(message -> {
            // Handle the incoming message
            // ...
            return Mono.empty();
        });
    }
}