mergedConfig = configurationProperties
- .mergedConsumerConfiguration();
- if (!ObjectUtils.isEmpty(mergedConfig)) {
- props.putAll(mergedConfig);
- }
- if (!props.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG)) {
- props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
- configurationProperties.getKafkaConnectionString());
- }
- ConsumerFactory consumerFactory = new DefaultKafkaConsumerFactory<>(props);
- ReactorKafkaBinderHealthIndicator indicator = new ReactorKafkaBinderHealthIndicator(
- reactorKafkaBinder, consumerFactory);
- indicator.setTimeout(configurationProperties.getHealthTimeout());
- indicator.setConsiderDownWhenAnyPartitionHasNoLeader(configurationProperties.isConsiderDownWhenAnyPartitionHasNoLeader());
- return indicator;
- }
-
-}
diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReceiverOptionsCustomizer.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReceiverOptionsCustomizer.java
deleted file mode 100644
index 1e29f07aa..000000000
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReceiverOptionsCustomizer.java
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
- * Copyright 2023-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.
- * You may obtain a copy of the License at
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.stream.binder.reactorkafka;
-
-import java.util.function.BiFunction;
-
-import reactor.kafka.receiver.ReceiverOptions;
-
-import org.springframework.core.Ordered;
-
-/**
- * Customizer for {@link ReceiverOptions}; the first parameter contains the binder name;
- * the second is the {@link ReceiverOptions} to customize. Return the customized
- * {@link ReceiverOptions}. Applied in {@link Ordered order} if multiple customizers are
- * found.
- *
- * Generic since 4.0.3 to allow customization of deserializers.
- *
- * @param the key type.
- * @param the value type.
- *
- * @author Gary Russell
- * @since 4.0.2
- *
- * @deprecated since 4.3
- * See the updates in: ...
- * A suggested alternative is to use the regular Kafka binder with reactive types.
- * This approach has some limitations as the application need to handle reactive use cases explicitly.
- */
-@Deprecated(since = "4.3", forRemoval = true)
-public interface ReceiverOptionsCustomizer
- extends BiFunction, ReceiverOptions>, Ordered {
-
- @Override
- default int getOrder() {
- return 0;
- }
-
-}
diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/SenderOptionsCustomizer.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/SenderOptionsCustomizer.java
deleted file mode 100644
index f2a6556a8..000000000
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/SenderOptionsCustomizer.java
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
- * Copyright 2023-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.
- * You may obtain a copy of the License at
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.stream.binder.reactorkafka;
-
-import java.util.function.BiFunction;
-
-import reactor.kafka.sender.SenderOptions;
-
-import org.springframework.core.Ordered;
-
-/**
- * Customizer for {@link SenderOptions}; the first parameter contains the binder name; the
- * second is the {@link SenderOptions} to customize. Return the customized
- * {@link SenderOptions}. Applied in {@link Ordered order} if multiple customizers are
- * found.
- *
- * Generic since 4.0.3 to allow customization of serializers.
- *
- * @param the key type.
- * @param the value type.
- *
- * @author Gary Russell
- * @since 4.0.2
- *
- * @deprecated since 4.3
- * See the updates in: ...
- * A suggested alternative is to use the regular Kafka binder with reactive types.
- * This approach has some limitations as the application need to handle reactive use cases explicitly.
- */
-@Deprecated(since = "4.3", forRemoval = true)
-public interface SenderOptionsCustomizer
- extends BiFunction, SenderOptions>, Ordered {
-
- @Override
- default int getOrder() {
- return 0;
- }
-
-}
diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/package-info.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/package-info.java
deleted file mode 100644
index a12de78b6..000000000
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/package-info.java
+++ /dev/null
@@ -1,20 +0,0 @@
-/*
- * Copyright 2021-2022 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.
- * You may obtain a copy of the License at
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-/**
- * Provides classes reactor-kafka binder support.
- */
-package org.springframework.cloud.stream.binder.reactorkafka;
diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/resources/META-INF/spring.binders b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/resources/META-INF/spring.binders
deleted file mode 100644
index ee8843881..000000000
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/resources/META-INF/spring.binders
+++ /dev/null
@@ -1,2 +0,0 @@
-reactorKafka:\
-org.springframework.cloud.stream.binder.reactorkafka.ReactorKafkaBinderConfiguration
diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/resources/META-INF/spring.factories b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/resources/META-INF/spring.factories
deleted file mode 100644
index dd6e28b8a..000000000
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/resources/META-INF/spring.factories
+++ /dev/null
@@ -1,2 +0,0 @@
-org.springframework.boot.env.EnvironmentPostProcessor:\
-org.springframework.cloud.stream.binder.kafka.common.KafkaBinderEnvironmentPostProcessor
diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderHealthIndicatorTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderHealthIndicatorTests.java
deleted file mode 100644
index 8780aeed2..000000000
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderHealthIndicatorTests.java
+++ /dev/null
@@ -1,100 +0,0 @@
-/*
- * Copyright 2022-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.
- * You may obtain a copy of the License at
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.stream.binder.reactorkafka;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import org.apache.kafka.clients.consumer.KafkaConsumer;
-import org.apache.kafka.common.Node;
-import org.apache.kafka.common.PartitionInfo;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.TestInstance;
-import org.mockito.Mock;
-import org.mockito.MockitoAnnotations;
-
-import org.springframework.boot.actuate.health.Health;
-import org.springframework.boot.actuate.health.Status;
-import org.springframework.cloud.stream.binder.kafka.common.TopicInformation;
-import org.springframework.integration.endpoint.MessageProducerSupport;
-import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
-
-import static java.util.Collections.singleton;
-import static org.assertj.core.api.Assertions.assertThat;
-
-/**
- * @author Soby Chacko
- */
-@TestInstance(TestInstance.Lifecycle.PER_CLASS)
-class ReactorKafkaBinderHealthIndicatorTests {
-
- private static final String TEST_TOPIC = "test";
-
- private ReactorKafkaBinderHealthIndicator indicator;
-
- @Mock
- private DefaultKafkaConsumerFactory consumerFactory;
-
- @Mock
- private KafkaConsumer consumer;
-
- @Mock
- MessageProducerSupport messageProducerSupport1;
-
- @Mock
- private ReactorKafkaBinder binder;
-
- private final Map topicsInUse = new HashMap<>();
-
- @BeforeEach
- public void setup() {
- MockitoAnnotations.openMocks(this);
- org.mockito.BDDMockito.given(consumerFactory.createConsumer())
- .willReturn((consumer));
- org.mockito.BDDMockito.given(binder.getTopicsInUse()).willReturn(topicsInUse);
- this.indicator = new ReactorKafkaBinderHealthIndicator(binder, consumerFactory);
- this.indicator.setTimeout(10);
- }
-
- @Test
- void reactorKafkaBinderIsUp() {
- final List partitions = partitions(new Node(0, null, 0));
- topicsInUse.put(TEST_TOPIC, new TopicInformation(
- "group1-healthIndicator", partitions, false));
- org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC))
- .willReturn(partitions);
- org.mockito.BDDMockito.given(binder.getMessageProducers())
- .willReturn(Map.of("group1-healthIndicator", messageProducerSupport1));
- org.mockito.BDDMockito.given(messageProducerSupport1.isRunning()).willReturn(true);
- org.mockito.BDDMockito.given(messageProducerSupport1.isActive()).willReturn(true);
- Health health = indicator.health();
- assertThat(health.getStatus()).isEqualTo(Status.UP);
- assertThat(health.getDetails()).containsEntry("topicsInUse", singleton(TEST_TOPIC));
- assertThat(health.getDetails()).hasEntrySatisfying("messageProducers", value ->
- assertThat((ArrayList) value).hasSize(1));
- }
-
- private List partitions(Node leader) {
- List partitions = new ArrayList<>();
- partitions.add(new PartitionInfo(TEST_TOPIC, 0, leader, null, null));
- return partitions;
- }
-
-}
diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderIntegrationTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderIntegrationTests.java
deleted file mode 100644
index 61d68e3dc..000000000
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderIntegrationTests.java
+++ /dev/null
@@ -1,260 +0,0 @@
-/*
- * Copyright 2022-2024 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.
- * You may obtain a copy of the License at
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.stream.binder.reactorkafka;
-
-import java.lang.reflect.Type;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.Locale;
-import java.util.Map;
-import java.util.function.Function;
-
-import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.kafka.common.serialization.StringDeserializer;
-import org.junit.jupiter.api.Disabled;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.ValueSource;
-import reactor.core.publisher.Flux;
-import reactor.kafka.receiver.ReceiverOptions;
-import reactor.kafka.receiver.ReceiverRecord;
-import reactor.kafka.sender.SenderResult;
-
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.WebApplicationType;
-import org.springframework.boot.actuate.health.CompositeHealthContributor;
-import org.springframework.boot.actuate.health.Health;
-import org.springframework.boot.actuate.health.Status;
-import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
-import org.springframework.boot.builder.SpringApplicationBuilder;
-import org.springframework.boot.test.context.SpringBootTest;
-import org.springframework.cloud.stream.function.StreamBridge;
-import org.springframework.context.ConfigurableApplicationContext;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.integration.annotation.ServiceActivator;
-import org.springframework.integration.channel.FluxMessageChannel;
-import org.springframework.integration.support.MessageBuilder;
-import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
-import org.springframework.kafka.support.Acknowledgment;
-import org.springframework.kafka.support.converter.MessagingMessageConverter;
-import org.springframework.kafka.support.converter.RecordMessageConverter;
-import org.springframework.kafka.test.EmbeddedKafkaBroker;
-import org.springframework.kafka.test.context.EmbeddedKafka;
-import org.springframework.kafka.test.utils.KafkaTestUtils;
-import org.springframework.messaging.Message;
-import org.springframework.test.annotation.DirtiesContext;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.awaitility.Awaitility.await;
-
-/**
- * Integration tests for {@link ReactorKafkaBinder}.
- *
- * @author Soby Chacko
- * @author Gary Russell
- * @author Chris Bono
- */
-@SpringBootTest
-@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD)
-@EmbeddedKafka(topics = { "uppercased-words", "lowercased-words" })
-class ReactorKafkaBinderIntegrationTests {
-
- private static final List recOptsCustOrder = Collections.synchronizedList(new ArrayList<>());
-
- private static final List patternedDeliveries = Collections.synchronizedList(new ArrayList<>());
-
- @Autowired
- private EmbeddedKafkaBroker embeddedKafka;
-
- @ParameterizedTest
- @ValueSource(booleans = { false, true })
- @Disabled
- void endToEndReactorKafkaBinder(boolean excludeKafkaAutoConfig) throws InterruptedException {
-
- recOptsCustOrder.clear();
- patternedDeliveries.clear();
- Map consumerProps = KafkaTestUtils.consumerProps("group1", "false", embeddedKafka);
- consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
- DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps);
- Consumer consumer1 = cf.createConsumer();
- embeddedKafka.consumeFromEmbeddedTopics(consumer1, "uppercased-words");
- Consumer consumer2 = cf.createConsumer("group2", null);
- embeddedKafka.consumeFromEmbeddedTopics(consumer2, "lowercased-words");
-
- try (ConfigurableApplicationContext context = new SpringApplicationBuilder(ReactiveKafkaApplication.class)
- .web(WebApplicationType.NONE).run(
- "--server.port=0",
- "--spring.jmx.enabled=false",
- "--spring.kafka.consumer.metadata.max.age.ms=1000",
- "--spring.cloud.function.definition=uppercase;lowercase;patternConsumer",
- "--spring.cloud.stream.function.reactive.uppercase=true",
- "--spring.cloud.stream.function.reactive.lowercase=true",
- "--spring.cloud.stream.bindings.uppercase-in-0.group=grp1",
- "--spring.cloud.stream.bindings.uppercase-in-0.destination=words1",
- "--spring.cloud.stream.bindings.uppercase-out-0.destination=uppercased-words",
- "--spring.cloud.stream.bindings.lowercase-in-0.group=grp2",
- "--spring.cloud.stream.bindings.lowercase-in-0.destination=words2",
- "--spring.cloud.stream.bindings.lowercase-out-0.destination=lowercased-words",
- "--spring.cloud.stream.bindings.patternConsumer-in-0.group=grp3",
- "--spring.cloud.stream.bindings.patternConsumer-in-0.destination=.*-words",
- "--spring.cloud.stream.kafka.bindings.words1.producer.record-metadata-channel=sendResults",
- "--spring.cloud.stream.kafka.bindings.words2.producer.record-metadata-channel=sendResults",
- "--spring.cloud.stream.kafka.bindings.patternConsumer-in-0.consumer.destination-is-pattern=true",
- "--spring.cloud.stream.kafka.bindings.lowercase-in-0.consumer.converterBeanName=fullRR",
- "--spring.cloud.stream.kafka.binder.brokers=" + embeddedKafka.getBrokersAsString(),
- excludeKafkaAutoConfigParam(excludeKafkaAutoConfig))) {
-
- StreamBridge streamBridge = context.getBean(StreamBridge.class);
- streamBridge.send("words1", MessageBuilder.withPayload("foobar")
- .setCorrelationId(42)
- .build());
- streamBridge.send("words2", MessageBuilder.withPayload("BAZQUX")
- .setCorrelationId(43)
- .build());
-
- assertThat(KafkaTestUtils.getSingleRecord(consumer1, "uppercased-words"))
- .isNotNull()
- .extracting(ConsumerRecord::value)
- .isEqualTo("FOOBAR");
-
- assertThat(KafkaTestUtils.getSingleRecord(consumer2, "lowercased-words"))
- .isNotNull()
- .extracting(ConsumerRecord::value)
- .isEqualTo("bazqux");
-
- assertThat(recOptsCustOrder).containsExactly("two", "one", "two", "one", "two", "one");
- await().untilAsserted(() -> assertThat(patternedDeliveries).contains("bazqux", "FOOBAR"));
- assertThat(context.getBean(ReactiveKafkaApplication.class).correlation).contains(42, 43);
-
- checkHealth(context, Status.UP);
- }
- }
-
- private static void checkHealth(ConfigurableApplicationContext context,
- Status expected) throws InterruptedException {
- CompositeHealthContributor healthIndicator = context
- .getBean("bindersHealthContributor", CompositeHealthContributor.class);
- ReactorKafkaBinderHealthIndicator rkbhi = (ReactorKafkaBinderHealthIndicator) healthIndicator.getContributor("reactorKafka");
- Health health = rkbhi.health();
- assertThat(health.getStatus()).isEqualTo(expected);
- }
-
- private String excludeKafkaAutoConfigParam(boolean excludeKafkaAutoConfig) {
- return excludeKafkaAutoConfig ?
- "--spring.autoconfigure.exclude=org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration" : "foo=bar";
- }
-
- @EnableAutoConfiguration
- @Configuration
- public static class ReactiveKafkaApplication {
-
- final List correlation = Collections.synchronizedList(new ArrayList<>());
-
- @Bean
- RecordMessageConverter fullRR() {
- return new RecordMessageConverter() {
-
- private final RecordMessageConverter converter = new MessagingMessageConverter();
-
- @Override
- public Message toMessage(ConsumerRecord record, Acknowledgment acknowledgment,
- Consumer consumer, Type payloadType) {
-
- return MessageBuilder.withPayload(record).build();
- }
-
- @Override
- public ProducerRecord fromMessage(Message message, String defaultTopic) {
- return this.converter.fromMessage(message, defaultTopic);
- }
-
- };
- }
-
- @Bean
- Function, Flux> uppercase() {
- return s -> s.map(String::toUpperCase);
- }
-
- @Bean
- Function>, Flux> lowercase() {
- return s -> s.map(rec -> new String(rec.value()).toLowerCase(Locale.ROOT));
- }
-
- @Bean
- java.util.function.Consumer> patternConsumer() {
- return f -> f.doOnNext(s -> patternedDeliveries.add(s))
- .subscribe();
- }
-
- @Bean
- ReceiverOptionsCustomizer cust1() {
- return (t, u) -> {
- recOptsCustOrder.add("one");
- return u;
- };
- }
-
- @Bean
- ReceiverOptionsCustomizer cust2() {
- return new ReceiverOptionsCustomizer<>() {
-
- @Override
- public ReceiverOptions apply(String t, ReceiverOptions u) {
- recOptsCustOrder.add("two");
- u.withKeyDeserializer(new StringDeserializer());
- return u;
- }
-
- @Override
- public int getOrder() {
- return -1;
- }
-
- };
- }
-
- @Bean
- FluxMessageChannel sendResults() {
- return new FluxMessageChannel();
- }
-
- @ServiceActivator(inputChannel = "sendResults")
- void handleResults(SenderResult result) {
- if (result.exception() != null) {
- failureFor(result);
- }
- else {
- successFor(result);
- }
- }
-
- private void failureFor(SenderResult result) {
- this.correlation.clear();
- }
-
- private boolean successFor(SenderResult result) {
- return this.correlation.add(result.correlationMetadata());
- }
-
- }
-
-}
diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderObservationTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderObservationTests.java
deleted file mode 100644
index d83b2386e..000000000
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderObservationTests.java
+++ /dev/null
@@ -1,163 +0,0 @@
-/*
- * Copyright 2022-2024 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.
- * You may obtain a copy of the License at
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.stream.binder.reactorkafka;
-
-import java.lang.reflect.Type;
-import java.time.Duration;
-import java.util.Locale;
-import java.util.function.Function;
-import java.util.stream.Collectors;
-
-import brave.handler.SpanHandler;
-import brave.test.TestSpanHandler;
-import io.micrometer.observation.Observation;
-import io.micrometer.observation.ObservationRegistry;
-import io.micrometer.observation.contextpropagation.ObservationThreadLocalAccessor;
-import io.micrometer.tracing.brave.bridge.BraveFinishedSpan;
-import io.micrometer.tracing.test.simple.SpansAssert;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.clients.producer.ProducerRecord;
-import org.junit.jupiter.api.Disabled;
-import org.junit.jupiter.api.Test;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-import reactor.kafka.receiver.ReceiverRecord;
-import reactor.kafka.receiver.observation.KafkaReceiverObservation;
-import reactor.kafka.receiver.observation.KafkaRecordReceiverContext;
-
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.SpringBootConfiguration;
-import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
-import org.springframework.boot.test.autoconfigure.actuate.observability.AutoConfigureObservability;
-import org.springframework.boot.test.context.SpringBootTest;
-import org.springframework.cloud.stream.function.StreamBridge;
-import org.springframework.context.annotation.Bean;
-import org.springframework.integration.IntegrationMessageHeaderAccessor;
-import org.springframework.integration.support.MessageBuilder;
-import org.springframework.kafka.support.Acknowledgment;
-import org.springframework.kafka.support.converter.MessagingMessageConverter;
-import org.springframework.kafka.support.converter.RecordMessageConverter;
-import org.springframework.kafka.test.EmbeddedKafkaBroker;
-import org.springframework.kafka.test.context.EmbeddedKafka;
-import org.springframework.messaging.Message;
-import org.springframework.test.annotation.DirtiesContext;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.awaitility.Awaitility.await;
-
-/**
- * @author Artem Bilan
- * @author Soby Chacko
- * @since 4.2.0
- */
-@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, properties = {
- "spring.kafka.consumer.metadata.max.age.ms=1000",
- "spring.cloud.function.definition=receive",
- "spring.cloud.stream.function.reactive.uppercase=true",
- "spring.cloud.stream.bindings.receive-in-0.group=rkbot-in-group",
- "spring.cloud.stream.bindings.receive-in-0.destination=rkbot-in-topic",
- "spring.cloud.stream.bindings.receive-out-0.destination=rkbot-out-topic",
- "spring.cloud.stream.kafka.binder.enable-observation=true",
- "spring.cloud.stream.kafka.binder.brokers=${spring.kafka.bootstrap-servers}",
- "management.tracing.sampling.probability=1",
- "spring.cloud.stream.kafka.bindings.receive-in-0.consumer.converterBeanName=fullRR"
- })
-@DirtiesContext
-@AutoConfigureObservability
-@EmbeddedKafka(topics = { "rkbot-out-topic" })
-public class ReactorKafkaBinderObservationTests {
-
- private static final TestSpanHandler SPANS = new TestSpanHandler();
-
- @Autowired
- StreamBridge streamBridge;
-
- @Autowired
- ObservationRegistry observationRegistry;
-
- @Autowired
- TestConfiguration testConfiguration;
-
- @Autowired
- private EmbeddedKafkaBroker embeddedKafka;
-
- @Test
- @Disabled
- void endToEndReactorKafkaBinder1() {
-
- streamBridge.send("rkbot-in-topic", MessageBuilder.withPayload("data")
- .build());
-
- await().timeout(Duration.ofSeconds(10)).untilAsserted(() -> assertThat(SPANS.spans()).hasSize(4));
- SpansAssert.assertThat(SPANS.spans().stream().map(BraveFinishedSpan::fromBrave).collect(Collectors.toList()))
- .haveSameTraceId();
- }
-
- @SpringBootConfiguration
- @EnableAutoConfiguration(exclude = org.springframework.cloud.function.observability.ObservationAutoConfiguration.class)
- public static class TestConfiguration {
-
- @Bean
- SpanHandler testSpanHandler() {
- return SPANS;
- }
-
- @Bean
- RecordMessageConverter fullRR() {
- return new RecordMessageConverter() {
-
- private final RecordMessageConverter converter = new MessagingMessageConverter();
-
- @Override
- public Message toMessage(ConsumerRecord record, Acknowledgment acknowledgment,
- org.apache.kafka.clients.consumer.Consumer consumer, Type payloadType) {
-
- return MessageBuilder.withPayload(record).build();
- }
-
- @Override
- public ProducerRecord fromMessage(Message message, String defaultTopic) {
- return this.converter.fromMessage(message, defaultTopic);
- }
-
- };
- }
-
- @Bean
- Function>, Flux>> receive(ObservationRegistry observationRegistry) {
- return s -> s
- .flatMap(record -> {
- Observation receiverObservation =
- KafkaReceiverObservation.RECEIVER_OBSERVATION.start(null,
- KafkaReceiverObservation.DefaultKafkaReceiverObservationConvention.INSTANCE,
- () ->
- new KafkaRecordReceiverContext(
- record, "user.receiver", "localhost:9092"),
- observationRegistry);
-
- return Mono.deferContextual(contextView -> Mono.just(record)
- .map(rec -> new String(rec.value()).toLowerCase(Locale.ROOT))
- .map(rec -> MessageBuilder.withPayload(rec).setHeader(IntegrationMessageHeaderAccessor.REACTOR_CONTEXT, contextView).build()))
- .doOnTerminate(receiverObservation::stop)
- .doOnError(receiverObservation::error)
- .contextWrite(context -> context.put(ObservationThreadLocalAccessor.KEY, receiverObservation));
- });
- }
- }
-
-}
-
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
deleted file mode 100644
index a20862f07..000000000
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderTests.java
+++ /dev/null
@@ -1,353 +0,0 @@
-/*
- * Copyright 2021-2024 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.
- * You may obtain a copy of the License at
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-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;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicReference;
-
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.junit.jupiter.api.Disabled;
-import org.junit.jupiter.api.Test;
-import org.reactivestreams.Subscriber;
-import org.reactivestreams.Subscription;
-import reactor.core.publisher.Flux;
-import reactor.kafka.receiver.ReceiverOffset;
-import reactor.kafka.sender.SenderResult;
-
-import org.springframework.beans.factory.ObjectProvider;
-import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
-import org.springframework.cloud.stream.binder.Binding;
-import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
-import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
-import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
-import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
-import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties;
-import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner;
-import org.springframework.context.support.GenericApplicationContext;
-import org.springframework.integration.IntegrationMessageHeaderAccessor;
-import org.springframework.integration.channel.FluxMessageChannel;
-import org.springframework.integration.config.ConsumerEndpointFactoryBean;
-import org.springframework.integration.support.MessageBuilder;
-import org.springframework.kafka.core.DefaultKafkaProducerFactory;
-import org.springframework.kafka.core.KafkaTemplate;
-import org.springframework.kafka.support.KafkaHeaders;
-import org.springframework.kafka.test.condition.EmbeddedKafkaCondition;
-import org.springframework.kafka.test.context.EmbeddedKafka;
-import org.springframework.kafka.test.utils.KafkaTestUtils;
-import org.springframework.messaging.Message;
-import org.springframework.messaging.MessageChannel;
-import org.springframework.messaging.MessageHandler;
-import org.springframework.messaging.MessagingException;
-import org.springframework.retry.support.RetryTemplate;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.BDDMockito.given;
-import static org.mockito.Mockito.mock;
-
-/**
- * @author Gary Russell
- * @author Soby Chacko
- * @since 4.0
- *
- */
-@EmbeddedKafka(topics = { "testCa", "testCb", "testC1", "testC-Manual", "testC-AMO", "testP" })
-class ReactorKafkaBinderTests {
-
- @SuppressWarnings({ "rawtypes", "unchecked" })
- @Test
- @Disabled
- void consumerBinding() throws Exception {
- KafkaProperties kafkaProperties = new KafkaProperties();
- kafkaProperties.setBootstrapServers(
- Collections.singletonList(EmbeddedKafkaCondition.getBroker().getBrokersAsString()));
- KafkaBinderConfigurationProperties binderProps = new KafkaBinderConfigurationProperties(kafkaProperties, mock(ObjectProvider.class));
- KafkaTopicProvisioner provisioner = new KafkaTopicProvisioner(binderProps, kafkaProperties, prop -> {
- });
- provisioner.setMetadataRetryOperations(new RetryTemplate());
- ReactorKafkaBinder binder = new ReactorKafkaBinder(binderProps, provisioner, null);
- binder.setApplicationContext(mock(GenericApplicationContext.class));
-
- CountDownLatch latch = new CountDownLatch(2);
-
- FluxMessageChannel inbound = new FluxMessageChannel();
- Subscriber> sub = new Subscriber<>() {
-
- @Override
- public void onSubscribe(Subscription s) {
- s.request(2);
- }
-
- @Override
- public void onNext(Message t) {
- latch.countDown();
- }
-
- @Override
- public void onError(Throwable t) {
- }
-
- @Override
- public void onComplete() {
- }
-
- };
- inbound.subscribe(sub);
-
- KafkaConsumerProperties ext = new KafkaConsumerProperties();
- ExtendedConsumerProperties props =
- new ExtendedConsumerProperties(ext);
- props.setMultiplex(true);
-
- Binding consumer = binder.bindConsumer("testCa, testCb", "foo", inbound, props);
-
- DefaultKafkaProducerFactory pf =
- new DefaultKafkaProducerFactory<>(KafkaTestUtils.producerProps(EmbeddedKafkaCondition.getBroker()));
- KafkaTemplate kt = new KafkaTemplate<>(pf);
- kt.send("testCa", "foo").get(10, TimeUnit.SECONDS);
- kt.send("testCb", "bar").get(10, TimeUnit.SECONDS);
- assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
- consumer.unbind();
- pf.destroy();
- }
-
- @Test
- @Disabled
- void concurrencyManual() throws Exception {
- concurrency("testC-Manual", "concurrencyManual-group", false);
- }
-
- @Test
- void concurrencyAtMostOnce() throws Exception {
- concurrency("testC-AMO", "concurrencyAtMostOnce-group", true);
- }
-
- @SuppressWarnings({ "rawtypes", "unchecked" })
- void concurrency(String topic, String group, boolean atMostOnce) throws Exception {
- KafkaProperties kafkaProperties = new KafkaProperties();
- kafkaProperties.setBootstrapServers(
- Collections.singletonList(EmbeddedKafkaCondition.getBroker().getBrokersAsString()));
- KafkaBinderConfigurationProperties binderProps = new KafkaBinderConfigurationProperties(kafkaProperties, mock(ObjectProvider.class));
- KafkaTopicProvisioner provisioner = new KafkaTopicProvisioner(binderProps, kafkaProperties, prop -> {
- });
- provisioner.setMetadataRetryOperations(new RetryTemplate());
- ReactorKafkaBinder binder = new ReactorKafkaBinder(binderProps, provisioner, null);
- binder.setApplicationContext(mock(GenericApplicationContext.class));
-
- CountDownLatch subscriptionLatch = new CountDownLatch(1);
- CountDownLatch messageLatch1 = new CountDownLatch(4);
- CountDownLatch messageLatch2 = new CountDownLatch(6);
- 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) {
- payloads.add((String) msg.getPayload());
- partitions.add(msg.getHeaders().get(KafkaHeaders.RECEIVED_PARTITION, Integer.class));
- if (!atMostOnce) {
- msg.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, ReceiverOffset.class).acknowledge();
- }
- messageLatch1.countDown();
- messageLatch2.countDown();
- }
-
- @Override
- public void onError(Throwable t) {
- }
-
- @Override
- public void onComplete() {
- }
-
- };
- inbound.subscribe(sub);
-
- KafkaConsumerProperties ext = new KafkaConsumerProperties();
- ext.setReactiveAtMostOnce(atMostOnce);
- ExtendedConsumerProperties props =
- new ExtendedConsumerProperties<>(ext);
- props.setConcurrency(2);
-
- Binding consumer = binder.bindConsumer(topic, group, inbound, props);
-
- assertThat(subscriptionLatch.await(10, TimeUnit.SECONDS)).isTrue();
- DefaultKafkaProducerFactory pf =
- new DefaultKafkaProducerFactory<>(KafkaTestUtils.producerProps(EmbeddedKafkaCondition.getBroker()));
- KafkaTemplate kt = new KafkaTemplate<>(pf);
- kt.send(topic, 0, null, "foo").get(10, TimeUnit.SECONDS);
- kt.send(topic, 1, null, "bar").get(10, TimeUnit.SECONDS);
- kt.send(topic, 0, null, "baz").get(10, TimeUnit.SECONDS);
- kt.send(topic, 1, null, "qux").get(10, TimeUnit.SECONDS);
- assertThat(messageLatch1.await(10, TimeUnit.SECONDS)).isTrue();
- consumer.stop();
- consumer.start();
- kt.send(topic, 0, null, "fiz").get(10, TimeUnit.SECONDS);
- kt.send(topic, 1, null, "buz").get(10, TimeUnit.SECONDS);
- assertThat(messageLatch2.await(10, TimeUnit.SECONDS)).isTrue();
- assertThat(partitions).hasSize(2);
- consumer.unbind();
- pf.destroy();
- Collections.sort(payloads);
- 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, mock(ObjectProvider.class));
- KafkaTopicProvisioner provisioner = new KafkaTopicProvisioner(binderProps, kafkaProperties, prop -> {
- });
- provisioner.setMetadataRetryOperations(new RetryTemplate());
- ReactorKafkaBinder binder = new ReactorKafkaBinder(binderProps, provisioner, null);
- binder.setApplicationContext(mock(GenericApplicationContext.class));
-
- CountDownLatch subscriptionLatch = new CountDownLatch(1);
- CountDownLatch messageLatch1 = new CountDownLatch(4);
- 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
- @SuppressWarnings("unchecked")
- void producerBinding() throws InterruptedException {
- KafkaProperties kafkaProperties = new KafkaProperties();
- kafkaProperties.setBootstrapServers(
- Collections.singletonList(EmbeddedKafkaCondition.getBroker().getBrokersAsString()));
- KafkaBinderConfigurationProperties binderProps = new KafkaBinderConfigurationProperties(kafkaProperties, mock(ObjectProvider.class));
- KafkaTopicProvisioner provisioner = new KafkaTopicProvisioner(binderProps, kafkaProperties, prop -> {
- });
- provisioner.setMetadataRetryOperations(new RetryTemplate());
- ReactorKafkaBinder binder = new ReactorKafkaBinder(binderProps, provisioner, null);
- CountDownLatch latch = new CountDownLatch(1);
- GenericApplicationContext context = new GenericApplicationContext();
- context.registerBean("sendResults", FluxMessageChannel.class);
- context.refresh();
- FluxMessageChannel results = context.getBean("sendResults", FluxMessageChannel.class);
- ConsumerEndpointFactoryBean fb = new ConsumerEndpointFactoryBean();
- AtomicReference> senderResult = new AtomicReference<>();
- fb.setHandler(new MessageHandler() {
-
- @SuppressWarnings("unchecked")
- @Override
- public void handleMessage(Message message) throws MessagingException {
- senderResult.set((SenderResult) message.getPayload());
- latch.countDown();
- }
-
- });
- fb.setInputChannel(results);
- fb.setBeanFactory(context.getBeanFactory());
- fb.afterPropertiesSet();
- fb.start();
- binder.setApplicationContext(context);
- @SuppressWarnings("rawtypes")
- ObjectProvider cust = mock(ObjectProvider.class);
- AtomicBoolean custCalled = new AtomicBoolean();
- given(cust.getIfUnique()).willReturn((name, opts) -> {
- custCalled.set(true);
- return opts;
- });
- binder.senderOptionsCustomizers(cust);
-
- MessageChannel outbound = new FluxMessageChannel();
- KafkaProducerProperties ext = new KafkaProducerProperties();
- ExtendedProducerProperties props =
- new ExtendedProducerProperties(ext);
- props.getExtension().setRecordMetadataChannel("sendResults");
-
- Binding bindProducer = binder.bindProducer("testP", outbound, props);
- outbound.send(MessageBuilder.withPayload("foo")
- .setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, 1)
- .build());
- assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
- assertThat(senderResult.get().correlationMetadata()).isEqualTo(1);
- bindProducer.unbind();
- assertThat(custCalled).isTrue();
- }
-
-}
diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/resources/logback.xml b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/resources/logback.xml
deleted file mode 100644
index 5fdac6f28..000000000
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/resources/logback.xml
+++ /dev/null
@@ -1,20 +0,0 @@
-
-
-
- %d{ISO8601} %5p %t %c{2}:%L - %m%n
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
diff --git a/docs/modules/ROOT/nav.adoc b/docs/modules/ROOT/nav.adoc
index f5ea867ff..d683176c1 100644
--- a/docs/modules/ROOT/nav.adoc
+++ b/docs/modules/ROOT/nav.adoc
@@ -163,7 +163,6 @@
#* xref:README.adoc[]
#* xref:configprops.adoc[]
#* xref:intro.adoc[]
-#** xref:kafka/kafka_reactive_binder.adoc[]
#** xref:kafka/kafka-streams.adoc[]
#*** xref:kafka/kafka-streams-binder/usage.adoc[]
#*** xref:kafka/kafka-streams-binder/overview.adoc[]
diff --git a/docs/modules/ROOT/pages/kafka/kafka_reactive_binder.adoc b/docs/modules/ROOT/pages/kafka/kafka_reactive_binder.adoc
deleted file mode 100644
index c3b0a879e..000000000
--- a/docs/modules/ROOT/pages/kafka/kafka_reactive_binder.adoc
+++ /dev/null
@@ -1,223 +0,0 @@
-[[reactive-kafka-binder]]
-= Reactive Kafka Binder
-
-Kafka binder ecosystem in Spring Cloud Stream provides a dedicated reactive binder based on the https://projectreactor.io/docs/kafka/release/reference/[Reactor Kafka] project.
-This reactive Kafka binder enables full end-to-end reactive capabilities such as backpressure, reactive streams etc. in applications based on Apache Kafka.
-When your Spring Cloud Stream Kafka application is written using reactive types (`Flux`, `Mono` etc.), it is recommended to use this reactive Kafka binder instead of the regular message channel based Kafka binder.
-
-[[maven-coordinates]]
-== Maven Coordinates
-
-Following are the maven coordinates for the reactive Kafka binder.
-
-```
-
- org.springframework.cloud
- spring-cloud-stream-binder-kafka-reactive
-
-```
-
-[[basic-example-using-the-reactive-kafka-binder]]
-== Basic Example using the Reactive Kafka Binder
-
-In this section, we show some basic code snippets for writing a reactive Kafka application using the reactive binder and details around them.
-
-[source, java]
-----
-@Bean
-public Function, Flux> uppercase() {
- return s -> s.map(String::toUpperCase);
-}
-----
-
-You can use the above `upppercase` function with both message channel based Kafka binder (`spring-cloud-stream-binder-kafka`) as well as the reactive Kafka binder (`spring-cloud-stream-binder-kafka-reactive`), the topic of discussion in this section.
-When using this function with the regular Kafka binder, although you are using reactive types in the application (i.e., in the `uppercase` function), you only get the reactive streams within the execution of your function.
-Outside the function's execution context, there is no reactive benefits since the underlying binder is not based on the reactive stack.
-Therefore, although this might look like it is bringing a full end-to-end reactive stack, this application is only partially reactive.
-
-Now assume that you are using the proper reactive binder for Kafka - `spring-cloud-stream-binder-kafka-reactive` with the above function's application.
-This binder implementation will give the full reactive benefits all the way from consumption on the top end to publishing at the bottom end of the chain.
-This is because the underlying binder is built on top of https://projectreactor.io/docs/kafka/release/reference/[Reactor Kafka]'s core API's.
-On the consumer side, it makes use of the https://projectreactor.io/docs/kafka/release/reference/#api-guide-receiver[KafkaReceiver] which is a reactive implementation of a Kafka consumer.
-Similarly, on the producer side, it uses https://projectreactor.io/docs/kafka/release/reference/#api-guide-sender[KafkaSender] API which is the reactive implementation of a Kafka producer.
-Since the foundations of the reactive Kafka binder is built upon a proper reactive Kafka API, applications get the full benefits of using reactive technologies.
-Things like automatic back pressure, among other reactive capabilities, are built-in for the application when using this reactive Kafka binder.
-
-Starting with version 4.0.2, you can customize the `ReceiverOptions` and `SenderOptions` by providing one or more `ReceiverOptionsCustomizer` or `SenderOptionsCustomizer` beans respectively.
-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, 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]]
-== Consuming Records in the Raw Format
-
-In the above `upppercase` function, we are consuming the record as `Flux` and then produce it as `Flux`.
-There might be occasions in which you need to receive the record in the original received format - the `ReceiverRecord`.
-Here is such a function.
-
-[source, java]
-----
-@Bean
-public Function>, Flux> lowercase() {
- return s -> s.map(rec -> new String(rec.value()).toLowerCase());
-}
-----
-
-In this function, note that, we are consuming the record as `Flux>` and then producing it as `Flux`.
-`ReceiverRecord` is the basic received record which is a specialized Kafka `ConsumerRecord` in Reactor Kafka.
-When using the reactive Kafka binder, the above function will give you access to the `ReceiverRecord` type for each incoming record.
-However, in this case, you need to provide a custom implementation for a https://docs.spring.io/spring-kafka/docs/current/api/org/springframework/kafka/support/converter/RecordMessageConverter.html[RecordMessageConverter].
-By default, the reactive Kafka binder uses a https://docs.spring.io/spring-kafka/docs/current/api/org/springframework/kafka/support/converter/MessagingMessageConverter.html[MessagingMessageConverter] that converts the payload and headers from the `ConsumerRecord`.
-Therefore, by the time your handler method receives it, the payload is already extracted from the received record and passed onto the method as in the case of the first function we looked above.
-By providing a custom `RecordMessageConverter` implementation in the application, you can override the default behavior.
-For example, if you want to consume the record as raw `Flux>`, then you can provide the following bean definition in the application.
-
-[source, java]
-----
-@Bean
-RecordMessageConverter fullRawReceivedRecord() {
- return new RecordMessageConverter() {
-
- private final RecordMessageConverter converter = new MessagingMessageConverter();
-
- @Override
- public Message toMessage(ConsumerRecord record, Acknowledgment acknowledgment,
- Consumer consumer, Type payloadType) {
- return MessageBuilder.withPayload(record).build();
- }
-
- @Override
- public ProducerRecord fromMessage(Message message, String defaultTopic) {
- return this.converter.fromMessage(message, defaultTopic);
- }
-
- };
-}
-----
-
-Then, you need to instruct the framework to use this converter for the required binding.
-Here is an example based on our `lowercase` function.
-
-```
-spring.cloud.stream.kafka.bindings.lowercase-in-0.consumer.converterBeanName=fullRawReceivedRecord"
-```
-
-`lowercase-in-0` is the input binding name for our `lowercase` function.
-For the outbound (`lowecase-out-0`), we still use the regular `MessagingMessageConverter`.
-
-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]]
-== 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.
-In other words, this creates multiple reactive streams with separate `Flux` implementations.
-This could be useful when you are consuming records from a partitioned topic.
-
-For example, assume that the incoming topic has at least three partitions.
-Then you can set the following property.
-
-```
-spring.cloud.stream.bindings.lowercase-in-0.consumer.concurrency=3
-```
-
-That will create three dedicated `KafkaReceiver` objects that generate three separate `Flux` implementations and then stream them to the handler method.
-
-[[multiplex]]
-== Multiplex
-
-Starting with version 4.0.3, the common consumer property `multiplex` is now supported by the reactive binder, where a single binding can consume from multiple topics.
-When `false` (default), a separate binding is created for each topic specified in a comma-delimited list in the common `destination` property.
-
-[[destination-is-pattern]]
-== Destination is Pattern
-
-Starting with version 4.0.3, the `destination-is-pattern` Kafka binding consumer property is now supported.
-The receiver options are conigured with a regex `Pattern`, allowing the binding to consume from any topic that matches the pattern.
-
-[[sender-result-channel]]
-== Sender Result Channel
-
-Starting with version 4.0.3, you can configure the `resultMetadataChannel` to receive `SenderResult` s to determine success/failure of sends.
-
-The `SenderResult` contains `correlationMetadata` to allow you to correlate results with sends; it also contains `RecordMetadata`, which indicates the `TopicPartition` and offset of the sent record.
-
-The `resultMetadataChannel` **must** be a `FluxMessageChannel` instance.
-
-Here is an example of how to use this feature, with correlation metadata of type `Integer`:
-
-[source, java]
-----
-@Bean
-FluxMessageChannel sendResults() {
- return new FluxMessageChannel();
-}
-
-@ServiceActivator(inputChannel = "sendResults")
-void handleResults(SenderResult result) {
- if (result.exception() != null) {
- failureFor(result);
- }
- else {
- successFor(result);
- }
-}
-----
-
-To set the correlation metadata on an output record, set the `CORRELATION_ID` header:
-
-[source, java]
-----
-streamBridge.send("words1", MessageBuilder.withPayload("foobar")
- .setCorrelationId(42)
- .build());
-----
-
-When using the feature with a `Function`, the function output type must be a `Message` with the correlation id header set to the desired value.
-
-Metadata should be unique, at least for the duration of the send.