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 2cd6e5958..d6f6205bc 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 @@ -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; + } } 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 d41ef2e98..5da1e0b55 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 @@ -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>> 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> receive; - if (autoCommit) { - receive = this.receivers.get(i) - .receiveAutoAck() - .concatMap(rec -> rec); + Flux> receive = null; + KafkaReceiver 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 message = (Message) ((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 message = (Message) ((RecordMessageConverter) converter) + .toMessage(record, null, null, null); + return addAckHeaderIfNeeded(atMostOnce, record, message); + })); + } } if (concurrency == 1) { subscribeToPublisher(fluxes.get(0)); 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 a1e2a2c69..4d295ab99 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 @@ -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 props = new ExtendedConsumerProperties(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 partitions = new HashSet<>(); + List payloads = Collections.synchronizedList(new ArrayList<>()); + + FluxMessageChannel inbound = new FluxMessageChannel(); + Subscriber> sub = new Subscriber>() { + + @Override + public void onSubscribe(Subscription s) { + s.request(10); + subscriptionLatch.countDown(); + } + + @Override + public void onNext(Message msg) { + ((Message>>) 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 props = + new ExtendedConsumerProperties(ext); + props.setConcurrency(2); + + Binding 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 diff --git a/docs/src/main/asciidoc/kafka/kafka-reactive.adoc b/docs/src/main/asciidoc/kafka/kafka-reactive.adoc index 05e19fbd5..ec2fe951f 100644 --- a/docs/src/main/asciidoc/kafka/kafka-reactive.adoc +++ b/docs/src/main/asciidoc/kafka/kafka-reactive.adoc @@ -64,8 +64,33 @@ public Consumer> 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>>> input() { + return flux -> flux + .doOnNext(inner -> inner + .doOnNext(val -> { + log.info(val.value()); + }) + .subscribe()) + .subscribe(); +} +---- +==== + +Note that `reactor-kafka` returns a `Flux>>` 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