Out of the box support for consuming requests within specified timestamps

[rishavjayswal]

I think it would be pretty cool if we could have out of the box support to consume messages within a specified timestamp with reactor-kafka.
I have the following workaround for now:

ReceiverOptions<ByteBuffer, ByteBuffer> options =
        receiverOptions
            .subscription(topicConfig.getTopics())
            .pollTimeout(Duration.ofMillis(topicConfig.getPollWaitTimeoutMs()))
            .addAssignListener(
                partitions -> {
                  partitions.forEach(partition -> partition.seekToTimestamp(startTimeInMillis));
                });

KafkaReceiver.create(options)
        .receive()
        .takeWhile(record -> record.timestamp() < endTimeInMillis)
        .map(this::handleConsumerRecord);

But the solution poses the problem that when the above condition(.record -> record.timestamp() < endTimeInMillis) is met for a message, that message might be the last valid message in its partition, but it might be possible that other partitions in the topic still have messages below the end timestamp and would be left out.

Desired solution

An out of the box method that would let us consume messages till a certain predicate is true.
Alternatively, an option to access offset information for all the partitions before consuming records could also help as they can be stored in memory and be checked for each consumed record.

[garyrussell]

You should be able to achieve your desired behavior by implementing a ConsumerInterceptor and adding it to ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG.

/**
 * A plugin interface that allows you to intercept (and possibly mutate) records received by the consumer. A primary use-case
 * is for third-party components to hook into the consumer applications for custom monitoring, logging, etc.
 *
 * <p>
 * This class will get consumer config properties via <code>configure()</code> method, including clientId assigned
 * by KafkaConsumer if not specified in the consumer config. The interceptor implementation needs to be aware that it will be
 * sharing consumer config namespace with other interceptors and serializers, and ensure that there are no conflicts.
 * <p>
 * Exceptions thrown by ConsumerInterceptor methods will be caught, logged, but not propagated further. As a result, if
 * the user configures the interceptor with the wrong key and value type parameters, the consumer will not throw an exception,
 * just log the errors.
 * <p>
 * ConsumerInterceptor callbacks are called from the same thread that invokes
 * {@link org.apache.kafka.clients.consumer.KafkaConsumer#poll(java.time.Duration)}.
 * <p>
 * Implement {@link org.apache.kafka.common.ClusterResourceListener} to receive cluster metadata once it's available. Please see the class documentation for ClusterResourceListener for more information.
 */
public interface ConsumerInterceptor<K, V> extends Configurable, AutoCloseable {

You could iterate over the partitions and discard the unwanted records, and ending with a final one that won't pass your predicate.