diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java index 0a4d2561d..2cd6e5958 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java +++ b/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; + } + + } diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/pom.xml b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/pom.xml index 4695f0a60..b108204f1 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/pom.xml +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/pom.xml @@ -48,7 +48,7 @@ io.projectreactor.kafka reactor-kafka - 1.3.8 + 1.3.16 org.springframework.boot diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java index 736ef26a3..695200787 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java +++ b/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>> 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) ((RecordMessageConverter) converter) - .toMessage(record, null, null, null))); + Flux> 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 message = (Message) ((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 addAckHeaderIfNeeded(boolean autoCommit, ConsumerRecord record, + Message message) { + + if (!autoCommit) { + if (message.getHeaders() instanceof KafkaMessageHeaders headers) { + headers.getRawHeaders().put(KafkaHeaders.ACKNOWLEDGMENT, + ((ReceiverRecord) record).receiverOffset()); + } + else { + message = MessageBuilder.fromMessage(message) + .setHeader(KafkaHeaders.ACKNOWLEDGMENT, + ((ReceiverRecord) record).receiverOffset()) + .build(); + } + } + return message; + } + } return new ReactorMessageProducer(); } diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderTests.java index 4ab3b5afe..3bff3364d 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderTests.java +++ b/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 partitions = new HashSet<>(); + List payloads = Collections.synchronizedList(new ArrayList<>()); FluxMessageChannel inbound = new FluxMessageChannel(); Subscriber> sub = new Subscriber>() { @@ -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 props = new ExtendedConsumerProperties(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 diff --git a/docs/src/main/asciidoc/kafka/kafka-reactive.adoc b/docs/src/main/asciidoc/kafka/kafka-reactive.adoc index 24a4b6849..05e19fbd5 100644 --- a/docs/src/main/asciidoc/kafka/kafka-reactive.adoc +++ b/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> 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` and then produce it as `Flux`. @@ -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.