GH-2686: Reactive Kafka Fix Auto Commit

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

Due to `.concatMap()`, offsets were committed before passing the records to
the pipeline.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2021 the original author or authors.
+ * Copyright 2016-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -223,6 +223,13 @@ public class KafkaConsumerProperties {
 	 */
 	private boolean reactiveAutoCommit;
 
+	/**
+	 * When using the reactive binder, automatically commit offsets of records before they
+	 * are passed to the user code.
+	 * @since 4.0.3
+	 */
+	private boolean reactiveAtMostOnce;
+
 	/**
 	 * @return if each record needs to be acknowledged.
 	 *
@@ -558,5 +565,12 @@ public class KafkaConsumerProperties {
 		this.reactiveAutoCommit = reactiveAutoCommit;
 	}
 
+	public boolean isReactiveAtMostOnce() {
+		return this.reactiveAtMostOnce;
+	}
+
+	public void setReactiveAtMostOnce(boolean reactiveAtMostOnce) {
+		this.reactiveAtMostOnce = reactiveAtMostOnce;
+	}
 
 }

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

@@ -67,6 +67,7 @@ import org.springframework.kafka.support.converter.RecordMessageConverter;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.support.GenericMessage;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
@@ -209,23 +210,33 @@ public class ReactorKafkaBinder
 				List<Flux<Message<Object>>> fluxes = new ArrayList<>();
 				int concurrency = properties.getConcurrency();
 				boolean autoCommit = properties.getExtension().isReactiveAutoCommit();
+				boolean atMostOnce = properties.getExtension().isReactiveAtMostOnce();
+				Assert.state(!(autoCommit && atMostOnce),
+						"Cannot set both reactiveAutoCommit and reactiveAtMostOnce");
 				for (int i = 0; i < concurrency; i++) {
-					Flux<? extends ConsumerRecord<Object, Object>> receive;
-					if (autoCommit) {
-						receive = this.receivers.get(i)
-								.receiveAutoAck()
-								.concatMap(rec -> rec);
+					Flux<? extends ConsumerRecord<Object, Object>> receive = null;
+					KafkaReceiver<Object, Object> kafkaReceiver = this.receivers.get(i);
+					if (atMostOnce) {
+						receive = kafkaReceiver
+								.receiveAtmostOnce();
 					}
-					else {
-						receive = this.receivers.get(i)
+					else if (!autoCommit) {
+						receive = kafkaReceiver
 								.receive();
 					}
-					fluxes.add(receive
-							.map(record -> {
-								Message<Object> message = (Message<Object>) ((RecordMessageConverter) converter)
-									.toMessage(record, null, null, null);
-								return addAckHeaderIfNeeded(autoCommit, record, message);
-							}));
+					if (autoCommit) {
+						fluxes.add(kafkaReceiver
+								.receiveAutoAck()
+								.map(inner -> new GenericMessage<>(inner)));
+					}
+					else {
+						fluxes.add(receive
+								.map(record -> {
+									Message<Object> message = (Message<Object>) ((RecordMessageConverter) converter)
+										.toMessage(record, null, null, null);
+									return addAckHeaderIfNeeded(atMostOnce, record, message);
+								}));
+					}
 				}
 				if (concurrency == 1) {
 					subscribeToPublisher(fluxes.get(0));

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

@@ -26,10 +26,12 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
+import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.junit.jupiter.api.Test;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.kafka.receiver.ReceiverOffset;
 
@@ -122,17 +124,17 @@ public class ReactorKafkaBinderTests {
 	}
 
 	@Test
-	void concurrencyAuto() throws Exception {
+	void concurrencyManual() throws Exception {
 		concurrency(false);
 	}
 
 	@Test
-	void concurrencyManual() throws Exception {
+	void concurrencyAtMostOnce() throws Exception {
 		concurrency(true);
 	}
 
 	@SuppressWarnings({ "rawtypes", "unchecked" })
-	void concurrency(boolean manualCommit) throws Exception {
+	void concurrency(boolean atMostOnce) throws Exception {
 		KafkaProperties kafkaProperties = new KafkaProperties();
 		kafkaProperties.setBootstrapServers(
 				Collections.singletonList(EmbeddedKafkaCondition.getBroker().getBrokersAsString()));
@@ -162,7 +164,7 @@ public class ReactorKafkaBinderTests {
 			public void onNext(Message<?> msg) {
 				payloads.add((String) msg.getPayload());
 				partitions.add(msg.getHeaders().get(KafkaHeaders.RECEIVED_PARTITION, Integer.class));
-				if (manualCommit) {
+				if (!atMostOnce) {
 					msg.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, ReceiverOffset.class).acknowledge();
 				}
 				messageLatch1.countDown();
@@ -181,7 +183,7 @@ public class ReactorKafkaBinderTests {
 		inbound.subscribe(sub);
 
 		KafkaConsumerProperties ext = new KafkaConsumerProperties();
-		ext.setReactiveAutoCommit(!manualCommit);
+		ext.setReactiveAtMostOnce(atMostOnce);
 		ExtendedConsumerProperties<KafkaConsumerProperties> props =
 				new ExtendedConsumerProperties<KafkaConsumerProperties>(ext);
 		props.setConcurrency(2);
@@ -206,7 +208,80 @@ public class ReactorKafkaBinderTests {
 		consumer.unbind();
 		pf.destroy();
 		Collections.sort(payloads);
-		assertThat(payloads).containsExactly("bar", "baz", "buz", "fiz", "foo", "qux");
+		if (!atMostOnce) {
+			assertThat(payloads).containsExactly("bar", "baz", "buz", "fiz", "foo", "qux");
+		}
+	}
+
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	@Test
+	void autoCommit() throws Exception {
+		KafkaProperties kafkaProperties = new KafkaProperties();
+		kafkaProperties.setBootstrapServers(
+				Collections.singletonList(EmbeddedKafkaCondition.getBroker().getBrokersAsString()));
+		KafkaBinderConfigurationProperties binderProps = new KafkaBinderConfigurationProperties(kafkaProperties);
+		KafkaTopicProvisioner provisioner = new KafkaTopicProvisioner(binderProps, kafkaProperties, prop -> {
+		});
+		provisioner.setMetadataRetryOperations(new RetryTemplate());
+		ReactorKafkaBinder binder = new ReactorKafkaBinder(binderProps, provisioner);
+		binder.setApplicationContext(mock(GenericApplicationContext.class));
+
+		CountDownLatch subscriptionLatch = new CountDownLatch(1);
+		CountDownLatch messageLatch1 = new CountDownLatch(4);
+		Set<Integer> partitions = new HashSet<>();
+		List<String> payloads = Collections.synchronizedList(new ArrayList<>());
+
+		FluxMessageChannel inbound = new FluxMessageChannel();
+		Subscriber<Message<?>> sub = new Subscriber<Message<?>>() {
+
+			@Override
+			public void onSubscribe(Subscription s) {
+				s.request(10);
+				subscriptionLatch.countDown();
+			}
+
+			@Override
+			public void onNext(Message<?> msg) {
+				((Message<Flux<ConsumerRecord<?, String>>>) msg).getPayload()
+						.doOnNext(rec -> {
+							payloads.add(rec.value());
+							messageLatch1.countDown();
+						})
+						.subscribe();
+			}
+
+			@Override
+			public void onError(Throwable t) {
+			}
+
+			@Override
+			public void onComplete() {
+			}
+
+		};
+		inbound.subscribe(sub);
+
+		KafkaConsumerProperties ext = new KafkaConsumerProperties();
+		ext.setReactiveAutoCommit(true);
+		ExtendedConsumerProperties<KafkaConsumerProperties> props =
+				new ExtendedConsumerProperties<KafkaConsumerProperties>(ext);
+		props.setConcurrency(2);
+
+		Binding<MessageChannel> consumer = binder.bindConsumer("testC1", "foo", inbound, props);
+
+		assertThat(subscriptionLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		DefaultKafkaProducerFactory pf =
+				new DefaultKafkaProducerFactory<>(KafkaTestUtils.producerProps(EmbeddedKafkaCondition.getBroker()));
+		KafkaTemplate kt = new KafkaTemplate<>(pf);
+		kt.send("testC1", 0, null, "foo").get(10, TimeUnit.SECONDS);
+		kt.send("testC1", 1, null, "bar").get(10, TimeUnit.SECONDS);
+		kt.send("testC1", 0, null, "baz").get(10, TimeUnit.SECONDS);
+		kt.send("testC1", 1, null, "qux").get(10, TimeUnit.SECONDS);
+		assertThat(messageLatch1.await(10, TimeUnit.SECONDS)).isTrue();
+		consumer.unbind();
+		pf.destroy();
+		Collections.sort(payloads);
+		assertThat(payloads).containsExactly("bar", "baz", "foo", "qux");
 	}
 
 	@Test

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

@@ -64,8 +64,33 @@ public Consumer<Flux<Message<String>> consume() {
 
 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.
+In addition, starting with version 4.0.3, the Kafka consumer property `reactiveAtmostOnce` can be set to `true` and the binder will automatically commit the offsets before records returned by each poll are processed.
+Also, starting with version 4.0.3, you can set the consumer property `reactiveAutoCommit` to `true` and the the binder will automatically commit the offsets after the records returned by each poll are processed.
+In these cases, the acknowledgment header is not present.
+
+IMPORTANT: 4.0.2 also provided `reactiveAutoCommit`, but the implementation was incorrect, it behaved similarly to `reactiveAtMostOnce`.
+
+The following is an example of how to use `reaciveAutoCommit`.
+
+====
+[source, java]
+----
+@Bean
+Consumer<Flux<Flux<ConsumerRecord<?, String>>>> input() {
+	return flux -> flux
+			.doOnNext(inner -> inner
+				.doOnNext(val -> {
+					log.info(val.value());
+				})
+				.subscribe())
+			.subscribe();
+}
+----
+====
+
+Note that `reactor-kafka` returns a `Flux<Flux<ConsumerRecord<?, ?>>>` when using auto commit.
+Given that Spring has no access to the contents of the inner flux, the application must deal with the native `ConsumerRecord`; there is no message conversion or conversion service applied to the contents.
+This requires the use of native decoding (by specifying a `Deserializer` of the appropriate type in the configuration) to return record keys/values of the desired types.
 
 === Consuming Records in the Raw Format