GH-2653: Reactor Kafka Binder Fix Offset Commits

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2653

Add support for auto and manual offset commits.
Resolves #2656

binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java

@@ -216,6 +216,13 @@ public class KafkaConsumerProperties {
 	 */
 	private String commonErrorHandlerBeanName;
 
+	/**
+	 * When using the reactive binder, automatically commit offsets when records received
+	 * by the poll have been processed.
+	 * @since 4.0.2
+	 */
+	private boolean reactiveAutoCommit;
+
 	/**
 	 * @return if each record needs to be acknowledged.
 	 *
@@ -542,4 +549,14 @@ public class KafkaConsumerProperties {
 	public void setCommonErrorHandlerBeanName(String commonErrorHandlerBeanName) {
 		this.commonErrorHandlerBeanName = commonErrorHandlerBeanName;
 	}
+
+	public boolean isReactiveAutoCommit() {
+		return this.reactiveAutoCommit;
+	}
+
+	public void setReactiveAutoCommit(boolean reactiveAutoCommit) {
+		this.reactiveAutoCommit = reactiveAutoCommit;
+	}
+
+
 }

binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/pom.xml

@@ -48,7 +48,7 @@
         <dependency>
 			<groupId>io.projectreactor.kafka</groupId>
 			<artifactId>reactor-kafka</artifactId>
-			<version>1.3.8</version>
+			<version>1.3.16</version>
 		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>

binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java

@@ -25,12 +25,14 @@ import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Sinks;
 import reactor.kafka.receiver.KafkaReceiver;
 import reactor.kafka.receiver.ReceiverOptions;
+import reactor.kafka.receiver.ReceiverRecord;
 import reactor.kafka.sender.KafkaSender;
 import reactor.kafka.sender.SenderOptions;
 import reactor.kafka.sender.SenderRecord;
@@ -56,6 +58,9 @@ import org.springframework.context.Lifecycle;
 import org.springframework.integration.core.MessageProducer;
 import org.springframework.integration.endpoint.MessageProducerSupport;
 import org.springframework.integration.handler.AbstractMessageHandler;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.kafka.support.converter.KafkaMessageHeaders;
 import org.springframework.kafka.support.converter.MessageConverter;
 import org.springframework.kafka.support.converter.MessagingMessageConverter;
 import org.springframework.kafka.support.converter.RecordMessageConverter;
@@ -201,11 +206,24 @@ public class ReactorKafkaBinder
 			protected void doStart() {
 				List<Flux<Message<Object>>> fluxes = new ArrayList<>();
 				int concurrency = properties.getConcurrency();
+				boolean autoCommit = properties.getExtension().isReactiveAutoCommit();
 				for (int i = 0; i < concurrency; i++) {
-					fluxes.add(this.receivers.get(i)
-							.receive()
-							.map(record -> (Message<Object>) ((RecordMessageConverter) converter)
-									.toMessage(record, null, null, null)));
+					Flux<? extends ConsumerRecord<Object, Object>> receive;
+					if (autoCommit) {
+						receive = this.receivers.get(i)
+								.receiveAutoAck()
+								.concatMap(rec -> rec);
+					}
+					else {
+						receive = this.receivers.get(i)
+								.receive();
+					}
+					fluxes.add(receive
+							.map(record -> {
+								Message<Object> message = (Message<Object>) ((RecordMessageConverter) converter)
+									.toMessage(record, null, null, null);
+								return addAckHeaderIfNeeded(autoCommit, record, message);
+							}));
 				}
 				if (concurrency == 1) {
 					subscribeToPublisher(fluxes.get(0));
@@ -215,6 +233,24 @@ public class ReactorKafkaBinder
 				}
 			}
 
+			private Message<Object> addAckHeaderIfNeeded(boolean autoCommit, ConsumerRecord<Object, Object> record,
+					Message<Object> message) {
+
+				if (!autoCommit) {
+					if (message.getHeaders() instanceof KafkaMessageHeaders headers) {
+						headers.getRawHeaders().put(KafkaHeaders.ACKNOWLEDGMENT,
+								((ReceiverRecord<Object, Object>) record).receiverOffset());
+					}
+					else {
+						message = MessageBuilder.fromMessage(message)
+								.setHeader(KafkaHeaders.ACKNOWLEDGMENT,
+										((ReceiverRecord<Object, Object>) record).receiverOffset())
+								.build();
+					}
+				}
+				return message;
+			}
+
 		}
 		return new ReactorMessageProducer();
 	}

binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderTests.java

@@ -16,8 +16,10 @@
 
 package org.springframework.cloud.stream.binder.reactorkafka;
 
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -29,6 +31,7 @@ import org.junit.jupiter.api.Test;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 import reactor.core.publisher.Mono;
+import reactor.kafka.receiver.ReceiverOffset;
 
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
@@ -118,9 +121,18 @@ public class ReactorKafkaBinderTests {
 		pf.destroy();
 	}
 
-	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Test
-	void concurrency() throws Exception {
+	void concurrencyAuto() throws Exception {
+		concurrency(false);
+	}
+
+	@Test
+	void concurrencyManual() throws Exception {
+		concurrency(true);
+	}
+
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	void concurrency(boolean manualCommit) throws Exception {
 		KafkaProperties kafkaProperties = new KafkaProperties();
 		kafkaProperties.setBootstrapServers(
 				Collections.singletonList(EmbeddedKafkaCondition.getBroker().getBrokersAsString()));
@@ -133,8 +145,9 @@ public class ReactorKafkaBinderTests {
 
 		CountDownLatch subscriptionLatch = new CountDownLatch(1);
 		CountDownLatch messageLatch1 = new CountDownLatch(4);
-		CountDownLatch messageLatch2 = new CountDownLatch(10);
+		CountDownLatch messageLatch2 = new CountDownLatch(6);
 		Set<Integer> partitions = new HashSet<>();
+		List<String> payloads = Collections.synchronizedList(new ArrayList<>());
 
 		FluxMessageChannel inbound = new FluxMessageChannel();
 		Subscriber<Message<?>> sub = new Subscriber<Message<?>>() {
@@ -147,7 +160,11 @@ public class ReactorKafkaBinderTests {
 
 			@Override
 			public void onNext(Message<?> msg) {
+				payloads.add((String) msg.getPayload());
 				partitions.add(msg.getHeaders().get(KafkaHeaders.RECEIVED_PARTITION, Integer.class));
+				if (manualCommit) {
+					msg.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, ReceiverOffset.class).acknowledge();
+				}
 				messageLatch1.countDown();
 				messageLatch2.countDown();
 			}
@@ -164,6 +181,7 @@ public class ReactorKafkaBinderTests {
 		inbound.subscribe(sub);
 
 		KafkaConsumerProperties ext = new KafkaConsumerProperties();
+		ext.setReactiveAutoCommit(!manualCommit);
 		ExtendedConsumerProperties<KafkaConsumerProperties> props =
 				new ExtendedConsumerProperties<KafkaConsumerProperties>(ext);
 		props.setConcurrency(2);
@@ -187,6 +205,8 @@ public class ReactorKafkaBinderTests {
 		assertThat(partitions).hasSize(2);
 		consumer.unbind();
 		pf.destroy();
+		Collections.sort(payloads);
+		assertThat(payloads).containsExactly("bar", "baz", "buz", "fiz", "foo", "qux");
 	}
 
 	@Test

docs/src/main/asciidoc/kafka/kafka-reactive.adoc

@@ -46,6 +46,27 @@ Starting with version 4.0.2, you can customize the `ReceiverOptions` and `Sender
 They are `BiFunction` s which receive the binding name and initial options, returning the customized options.
 The interfaces extend `Ordered` so the customizers will be applied in the order required, when more than one are present.
 
+IMPORTANT: The binder does not commit offsets by default.
+Starting with version 4.0.2, the `KafkaHeaders.ACKNOWLEDGMENT` header contains a `ReceiverOffset` object which allows you to cause the offset to be committed by calling its `acknowledge()` or `commit()` methods.
+
+====
+[source, java]
+----
+@Bean
+public Consumer<Flux<Message<String>> consume() {
+    return msg -> {
+        process(msg.getPayload());
+        msg.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, ReceiverOffset.class).acknowledge();
+    }
+}
+----
+====
+
+Refer to the `reactor-kafka` documentation and javadocs for more information.
+
+In addition, the Kafka consumer property `reactiveAutoCommit` can be set to `true` and the binder will automatically commit the offsets after all records returned by each poll are processed.
+In this case, the acknowledgment header is not present.
+
 === Consuming Records in the Raw Format
 
 In the above `upppercase` function, we are consuming the record as `Flux<String>` and then produce it as `Flux<String>`.
@@ -109,6 +130,9 @@ For the outbound (`lowecase-out-0`), we still use the regular `MessagingMessageC
 In the `toMessage` implementation above, we receive the raw `ConsumerRecord` (`ReceiverRecord` since we are in a reactive binder context) and then wrap it inside a `Message`.
 Then that message payload which is the `ReceiverRecord` is provided to the user method.
 
+If `reactiveAutoCommit` is `false` (default), call `rec.receiverOffset().acknowledge()` (or `commit()`) to cause the offset to be committed; if `reactiveAutoCommit` is `true`, the flux supplies `ConsumerRecord` s instead.
+Refer to the `reactor-kafka` documentation and javadocs for more information.
+
 === Concurrency
 
 When using reactive functions with the reactive Kafka binder, if you set concurrency on the consumer binding, then the binder creates as many dedicated `KafkaReceiver` objects as provided by the concurrency value.