messageProducerDetails = new HashMap<>();
- for (String groupId : messageProducerSupportInfo.keySet()) {
- MessageProducerSupport messageProducerSupport = messageProducerSupportInfo.get(groupId);
- boolean isRunning = messageProducerSupport.isRunning();
- boolean isOk = messageProducerSupport.isActive();
- if (!isOk) {
- status = Status.DOWN;
- }
- messageProducerDetails.put("isRunning", isRunning);
- messageProducerDetails.put("isStoppedAbnormally", !isRunning && !isOk);
- //messageProducerDetails.put("isPaused", messageProducerSupport.isPaused());
- messageProducerDetails.put("messageProducerId", messageProducerSupport.getApplicationContextId());
- messageProducerDetails.put("groupId", groupId);
- }
- messageProducers.add(messageProducerDetails);
- return Health.status(status)
- .withDetail("messageProducers", messageProducers)
- .build();
- }
-}
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 2f2fbc230..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,48 +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
- *
- */
-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 37070e865..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,48 +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
- *
- */
-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 c8b39414c..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)
-public 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 79c7bb4b8..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,242 +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.lang.reflect.Type;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-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.extension.ExtendWith;
-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.autoconfigure.EnableAutoConfiguration;
-import org.springframework.boot.builder.SpringApplicationBuilder;
-import org.springframework.cloud.stream.function.StreamBridge;
-import org.springframework.context.ConfigurableApplicationContext;
-import org.springframework.context.annotation.Bean;
-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 org.springframework.test.context.junit.jupiter.SpringExtension;
-
-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
- */
-@ExtendWith(SpringExtension.class)
-@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 })
- void endToEndReactorKafkaBinder(boolean excludeKafkaAutoConfig) {
-
- 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);
- }
- }
-
- private String excludeKafkaAutoConfigParam(boolean excludeKafkaAutoConfig) {
- return excludeKafkaAutoConfig ?
- "--spring.autoconfigure.exclude=org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration" : "foo=bar";
- }
-
- @EnableAutoConfiguration
- 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());
- }
-
- @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/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 919691745..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,352 +0,0 @@
-/*
- * Copyright 2021-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.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.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;
-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
- * @since 4.0
- *
- */
-@EmbeddedKafka(topics = { "testCa", "testCb", "testC1", "testP" })
-public class ReactorKafkaBinderTests {
-
- @SuppressWarnings({ "rawtypes", "unchecked" })
- @Test
- void consumerBinding() 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 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
- void concurrencyManual() throws Exception {
- concurrency(false);
- }
-
- @Test
- void concurrencyAtMostOnce() throws Exception {
- concurrency(true);
- }
-
- @SuppressWarnings({ "rawtypes", "unchecked" })
- void concurrency(boolean atMostOnce) 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);
- 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("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.stop();
- consumer.start();
- kt.send("testC1", 0, null, "fiz").get(10, TimeUnit.SECONDS);
- kt.send("testC1", 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);
- 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
- void producerBinding() throws InterruptedException {
- 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);
- 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);
- AtomicReference> sendResult = new AtomicReference<>();
- 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/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/.jdk8 b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/.jdk8
deleted file mode 100644
index e69de29bb..000000000
diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/pom.xml b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/pom.xml
deleted file mode 100644
index c6b27579a..000000000
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/pom.xml
+++ /dev/null
@@ -1,86 +0,0 @@
-
-
- 4.0.0
-
- spring-cloud-stream-binder-kafka-streams
- jar
- spring-cloud-stream-binder-kafka-streams
- Kafka Streams Binder Implementation
-
-
- org.springframework.cloud
- spring-cloud-stream-binder-kafka-parent
- 4.1.0-SNAPSHOT
-
-
-
- 1.8.2
-
-
-
-
- org.springframework.cloud
- spring-cloud-stream-binder-kafka-core
-
-
- org.springframework.boot
- spring-boot-starter-actuator
- true
-
-
- org.springframework.boot
- spring-boot-configuration-processor
- true
-
-
- org.springframework.boot
- spring-boot-autoconfigure
- true
-
-
- org.apache.kafka
- kafka-streams
-
-
- org.springframework.kafka
- spring-kafka
-
-
- org.springframework.kafka
- spring-kafka-test
- test
-
-
- org.springframework.boot
- spring-boot-starter-test
- test
-
-
- org.springframework.boot
- spring-boot-autoconfigure-processor
- true
-
-
- org.awaitility
- awaitility
- test
-
-
- org.springframework.integration
- spring-integration-test
-
-
- org.apache.avro
- avro-compiler
-
-
- test
-
-
- org.springframework.cloud
- spring-cloud-stream-binder-kafka
- test
-
-
-
-
diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java
deleted file mode 100644
index ba97e4057..000000000
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java
+++ /dev/null
@@ -1,663 +0,0 @@
-/*
- * Copyright 2019-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.kafka.streams;
-
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Optional;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicReference;
-import java.util.regex.Pattern;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.common.header.Header;
-import org.apache.kafka.common.header.Headers;
-import org.apache.kafka.common.serialization.Serde;
-import org.apache.kafka.common.serialization.Serdes;
-import org.apache.kafka.common.utils.Bytes;
-import org.apache.kafka.streams.StreamsBuilder;
-import org.apache.kafka.streams.StreamsConfig;
-import org.apache.kafka.streams.Topology;
-import org.apache.kafka.streams.errors.LogAndContinueExceptionHandler;
-import org.apache.kafka.streams.errors.LogAndFailExceptionHandler;
-import org.apache.kafka.streams.kstream.Consumed;
-import org.apache.kafka.streams.kstream.GlobalKTable;
-import org.apache.kafka.streams.kstream.KStream;
-import org.apache.kafka.streams.kstream.KTable;
-import org.apache.kafka.streams.kstream.Materialized;
-import org.apache.kafka.streams.processor.TimestampExtractor;
-import org.apache.kafka.streams.processor.api.Processor;
-import org.apache.kafka.streams.processor.api.ProcessorContext;
-import org.apache.kafka.streams.processor.api.ProcessorSupplier;
-import org.apache.kafka.streams.processor.api.Record;
-import org.apache.kafka.streams.processor.api.RecordMetadata;
-import org.apache.kafka.streams.state.KeyValueStore;
-import org.apache.kafka.streams.state.StoreBuilder;
-
-import org.springframework.beans.BeansException;
-import org.springframework.beans.factory.config.BeanDefinition;
-import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
-import org.springframework.beans.factory.support.BeanDefinitionBuilder;
-import org.springframework.beans.factory.support.BeanDefinitionRegistry;
-import org.springframework.boot.context.properties.bind.BindContext;
-import org.springframework.boot.context.properties.bind.BindHandler;
-import org.springframework.boot.context.properties.bind.Bindable;
-import org.springframework.boot.context.properties.bind.PropertySourcesPlaceholdersResolver;
-import org.springframework.boot.context.properties.source.ConfigurationPropertyName;
-import org.springframework.boot.context.properties.source.ConfigurationPropertySources;
-import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
-import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties;
-import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsConsumerProperties;
-import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsExtendedBindingProperties;
-import org.springframework.cloud.stream.config.BindingProperties;
-import org.springframework.cloud.stream.config.BindingServiceProperties;
-import org.springframework.context.ApplicationContext;
-import org.springframework.context.ApplicationContextAware;
-import org.springframework.context.ConfigurableApplicationContext;
-import org.springframework.core.ResolvableType;
-import org.springframework.core.env.ConfigurableEnvironment;
-import org.springframework.core.env.MutablePropertySources;
-import org.springframework.integration.support.utils.IntegrationUtils;
-import org.springframework.kafka.config.KafkaStreamsConfiguration;
-import org.springframework.kafka.config.StreamsBuilderFactoryBean;
-import org.springframework.kafka.config.StreamsBuilderFactoryBeanConfigurer;
-import org.springframework.kafka.core.CleanupConfig;
-import org.springframework.kafka.streams.RecoveringDeserializationExceptionHandler;
-import org.springframework.messaging.MessageHeaders;
-import org.springframework.messaging.support.MessageBuilder;
-import org.springframework.util.Assert;
-import org.springframework.util.CollectionUtils;
-import org.springframework.util.ObjectUtils;
-import org.springframework.util.StringUtils;
-
-/**
- * @author Soby Chacko
- * @since 3.0.0
- */
-public abstract class AbstractKafkaStreamsBinderProcessor implements ApplicationContextAware {
-
- private static final Log LOG = LogFactory.getLog(AbstractKafkaStreamsBinderProcessor.class);
-
- private final KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue;
-
- private final BindingServiceProperties bindingServiceProperties;
-
- private final KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties;
-
- private final CleanupConfig cleanupConfig;
-
- private final KeyValueSerdeResolver keyValueSerdeResolver;
-
- protected ConfigurableApplicationContext applicationContext;
-
- public AbstractKafkaStreamsBinderProcessor(BindingServiceProperties bindingServiceProperties,
- KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue,
- KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties,
- KeyValueSerdeResolver keyValueSerdeResolver, CleanupConfig cleanupConfig) {
- this.bindingServiceProperties = bindingServiceProperties;
- this.kafkaStreamsBindingInformationCatalogue = kafkaStreamsBindingInformationCatalogue;
- this.kafkaStreamsExtendedBindingProperties = kafkaStreamsExtendedBindingProperties;
- this.keyValueSerdeResolver = keyValueSerdeResolver;
- this.cleanupConfig = cleanupConfig;
- }
-
- @Override
- public final void setApplicationContext(ApplicationContext applicationContext)
- throws BeansException {
- this.applicationContext = (ConfigurableApplicationContext) applicationContext;
- }
-
- protected Topology.AutoOffsetReset getAutoOffsetReset(String inboundName, KafkaStreamsConsumerProperties extendedConsumerProperties) {
- final KafkaConsumerProperties.StartOffset startOffset = extendedConsumerProperties
- .getStartOffset();
- Topology.AutoOffsetReset autoOffsetReset = null;
- if (startOffset != null) {
- autoOffsetReset = switch (startOffset) {
- case earliest -> Topology.AutoOffsetReset.EARLIEST;
- case latest -> Topology.AutoOffsetReset.LATEST;
- };
- }
- if (extendedConsumerProperties.isResetOffsets()) {
- AbstractKafkaStreamsBinderProcessor.LOG.warn("Detected resetOffsets configured on binding "
- + inboundName + ". "
- + "Setting resetOffsets in Kafka Streams binder does not have any effect.");
- }
- return autoOffsetReset;
- }
-
- @SuppressWarnings("unchecked")
- protected void handleKTableGlobalKTableInputs(Object[] arguments, int index, String input, Class parameterType, Object targetBean,
- StreamsBuilderFactoryBean streamsBuilderFactoryBean, StreamsBuilder streamsBuilder,
- KafkaStreamsConsumerProperties extendedConsumerProperties,
- Serde keySerde, Serde valueSerde, Topology.AutoOffsetReset autoOffsetReset, boolean firstBuild) {
- if (firstBuild) {
- addStateStoreBeans(streamsBuilder);
- }
- if (parameterType.isAssignableFrom(KTable.class)) {
- String materializedAs = extendedConsumerProperties.getMaterializedAs();
- String bindingDestination = this.bindingServiceProperties.getBindingDestination(input);
- KTable table = getKTable(extendedConsumerProperties, streamsBuilder, keySerde, valueSerde, materializedAs,
- bindingDestination, autoOffsetReset);
- KTableBoundElementFactory.KTableWrapper kTableWrapper =
- (KTableBoundElementFactory.KTableWrapper) targetBean;
- //wrap the proxy created during the initial target type binding with real object (KTable)
- kTableWrapper.wrap((KTable) table);
- this.kafkaStreamsBindingInformationCatalogue.addStreamBuilderFactoryPerBinding(input, streamsBuilderFactoryBean);
- this.kafkaStreamsBindingInformationCatalogue.addConsumerPropertiesPerSbfb(streamsBuilderFactoryBean,
- bindingServiceProperties.getConsumerProperties(input));
- arguments[index] = table;
- }
- else if (parameterType.isAssignableFrom(GlobalKTable.class)) {
- String materializedAs = extendedConsumerProperties.getMaterializedAs();
- String bindingDestination = this.bindingServiceProperties.getBindingDestination(input);
- GlobalKTable table = getGlobalKTable(extendedConsumerProperties, streamsBuilder, keySerde, valueSerde, materializedAs,
- bindingDestination, autoOffsetReset);
- GlobalKTableBoundElementFactory.GlobalKTableWrapper globalKTableWrapper =
- (GlobalKTableBoundElementFactory.GlobalKTableWrapper) targetBean;
- //wrap the proxy created during the initial target type binding with real object (KTable)
- globalKTableWrapper.wrap((GlobalKTable) table);
- this.kafkaStreamsBindingInformationCatalogue.addStreamBuilderFactoryPerBinding(input, streamsBuilderFactoryBean);
- this.kafkaStreamsBindingInformationCatalogue.addConsumerPropertiesPerSbfb(streamsBuilderFactoryBean,
- bindingServiceProperties.getConsumerProperties(input));
- arguments[index] = table;
- }
- }
-
- @SuppressWarnings({ "unchecked" })
- protected StreamsBuilderFactoryBean buildStreamsBuilderAndRetrieveConfig(String beanNamePostPrefix,
- ApplicationContext applicationContext, String inboundName,
- KafkaStreamsBinderConfigurationProperties kafkaStreamsBinderConfigurationProperties,
- StreamsBuilderFactoryBeanConfigurer customizer,
- ConfigurableEnvironment environment, BindingProperties bindingProperties) {
- ConfigurableListableBeanFactory beanFactory = this.applicationContext
- .getBeanFactory();
-
- Map streamConfigGlobalProperties = applicationContext
- .getBean("streamConfigGlobalProperties", Map.class);
-
- // Use a copy because the global configuration will be shared by multiple processors.
- Map streamConfiguration = new HashMap<>(streamConfigGlobalProperties);
-
- if (kafkaStreamsBinderConfigurationProperties != null) {
- final Map functionConfigMap = kafkaStreamsBinderConfigurationProperties.getFunctions();
- if (!CollectionUtils.isEmpty(functionConfigMap)) {
- final KafkaStreamsBinderConfigurationProperties.Functions functionConfig = functionConfigMap.get(beanNamePostPrefix);
- if (functionConfig != null) {
- final Map functionSpecificConfig = functionConfig.getConfiguration();
- if (!CollectionUtils.isEmpty(functionSpecificConfig)) {
- streamConfiguration.putAll(functionSpecificConfig);
- }
-
- String applicationId = functionConfig.getApplicationId();
- if (StringUtils.hasText(applicationId)) {
- streamConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);
- }
- }
- }
- }
-
- final MutablePropertySources propertySources = environment.getPropertySources();
-
- if (StringUtils.hasText(bindingProperties.getBinder())) {
- final KafkaStreamsBinderConfigurationProperties multiBinderKafkaStreamsBinderConfigurationProperties =
- applicationContext.getBean(bindingProperties.getBinder() + "-KafkaStreamsBinderConfigurationProperties", KafkaStreamsBinderConfigurationProperties.class);
- String connectionString = multiBinderKafkaStreamsBinderConfigurationProperties.getKafkaConnectionString();
- if (!StringUtils.hasText(connectionString)) {
- connectionString = (String) propertySources.get(bindingProperties.getBinder() + "-kafkaStreamsBinderEnv").getProperty("spring.cloud.stream.kafka.binder.brokers");
- }
-
- streamConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, connectionString);
-
- String binderProvidedApplicationId = multiBinderKafkaStreamsBinderConfigurationProperties.getApplicationId();
- if (StringUtils.hasText(binderProvidedApplicationId)) {
- streamConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG,
- binderProvidedApplicationId);
- }
-
- if (multiBinderKafkaStreamsBinderConfigurationProperties
- .getDeserializationExceptionHandler() == DeserializationExceptionHandler.logAndContinue) {
- streamConfiguration.put(
- StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
- LogAndContinueExceptionHandler.class);
- }
- else if (multiBinderKafkaStreamsBinderConfigurationProperties
- .getDeserializationExceptionHandler() == DeserializationExceptionHandler.logAndFail) {
- streamConfiguration.put(
- StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
- LogAndFailExceptionHandler.class);
- }
- else if (multiBinderKafkaStreamsBinderConfigurationProperties
- .getDeserializationExceptionHandler() == DeserializationExceptionHandler.sendToDlq) {
- streamConfiguration.put(
- StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
- RecoveringDeserializationExceptionHandler.class);
- SendToDlqAndContinue sendToDlqAndContinue = applicationContext.getBean(SendToDlqAndContinue.class);
- streamConfiguration.put(RecoveringDeserializationExceptionHandler.KSTREAM_DESERIALIZATION_RECOVERER, sendToDlqAndContinue);
- }
-
- if (!ObjectUtils.isEmpty(multiBinderKafkaStreamsBinderConfigurationProperties.getConfiguration())) {
- streamConfiguration.putAll(multiBinderKafkaStreamsBinderConfigurationProperties.getConfiguration());
- }
- if (!streamConfiguration.containsKey(StreamsConfig.REPLICATION_FACTOR_CONFIG)) {
- streamConfiguration.put(StreamsConfig.REPLICATION_FACTOR_CONFIG,
- (int) multiBinderKafkaStreamsBinderConfigurationProperties.getReplicationFactor());
- }
- }
-
- //this is only used primarily for StreamListener based processors. Although in theory, functions can use it,
- //it is ideal for functions to use the approach used in the above if statement by using a property like
- //spring.cloud.stream.kafka.streams.binder.functions.process.configuration.num.threads (assuming that process is the function name).
- KafkaStreamsConsumerProperties extendedConsumerProperties = this.kafkaStreamsExtendedBindingProperties
- .getExtendedConsumerProperties(inboundName);
- Map bindingConfig = extendedConsumerProperties.getConfiguration();
- Assert.state(!bindingConfig.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG),
- ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG + " cannot be overridden at the binding level; "
- + "use multiple binders instead");
- // We will only add the per binding configuration to the current streamConfiguration and not the global one.
- streamConfiguration
- .putAll(bindingConfig);
-
- String bindingLevelApplicationId = extendedConsumerProperties.getApplicationId();
- // override application.id if set at the individual binding level.
- // We provide this for backward compatibility with StreamListener based processors.
- // For function based processors see the approach used above
- // (i.e. use a property like spring.cloud.stream.kafka.streams.binder.functions.process.applicationId).
- if (StringUtils.hasText(bindingLevelApplicationId)) {
- streamConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG,
- bindingLevelApplicationId);
- }
-
- //If the application id is not set by any mechanism, then generate it.
- streamConfiguration.computeIfAbsent(StreamsConfig.APPLICATION_ID_CONFIG,
- k -> {
- String generatedApplicationID = beanNamePostPrefix + "-applicationId";
- LOG.info("Binder Generated Kafka Streams Application ID: " + generatedApplicationID);
- LOG.info("Use the binder generated application ID only for development and testing. ");
- LOG.info("For production deployments, please consider explicitly setting an application ID using a configuration property.");
- LOG.info("The generated applicationID is static and will be preserved over application restarts.");
- return generatedApplicationID;
- });
-
- handleConcurrency(applicationContext, inboundName, streamConfiguration);
-
- // Override deserialization exception handlers per binding
- final DeserializationExceptionHandler deserializationExceptionHandler =
- extendedConsumerProperties.getDeserializationExceptionHandler();
- if (deserializationExceptionHandler == DeserializationExceptionHandler.logAndFail) {
- streamConfiguration.put(
- StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
- LogAndFailExceptionHandler.class);
- }
- else if (deserializationExceptionHandler == DeserializationExceptionHandler.logAndContinue) {
- streamConfiguration.put(
- StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
- LogAndContinueExceptionHandler.class);
- }
- else if (deserializationExceptionHandler == DeserializationExceptionHandler.sendToDlq) {
- streamConfiguration.put(
- StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
- RecoveringDeserializationExceptionHandler.class);
- streamConfiguration.put(RecoveringDeserializationExceptionHandler.KSTREAM_DESERIALIZATION_RECOVERER,
- applicationContext.getBean(SendToDlqAndContinue.class));
- }
- else if (deserializationExceptionHandler == DeserializationExceptionHandler.skipAndContinue) {
- streamConfiguration.put(
- StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
- SkipAndContinueExceptionHandler.class);
- }
-
- KafkaStreamsConfiguration kafkaStreamsConfiguration = new KafkaStreamsConfiguration(streamConfiguration);
-
- StreamsBuilderFactoryBean streamsBuilderFactoryBean = this.cleanupConfig == null
- ? new StreamsBuilderFactoryBean(kafkaStreamsConfiguration)
- : new StreamsBuilderFactoryBean(kafkaStreamsConfiguration,
- this.cleanupConfig);
-
- streamsBuilderFactoryBean.setAutoStartup(false);
- BeanDefinition streamsBuilderBeanDefinition = BeanDefinitionBuilder
- .genericBeanDefinition(
- (Class) streamsBuilderFactoryBean.getClass(),
- () -> streamsBuilderFactoryBean)
- .getRawBeanDefinition();
- ((BeanDefinitionRegistry) beanFactory).registerBeanDefinition(
- "stream-builder-" + beanNamePostPrefix, streamsBuilderBeanDefinition);
-
- extendedConsumerProperties.setApplicationId((String) streamConfiguration.get(StreamsConfig.APPLICATION_ID_CONFIG));
-
- final StreamsBuilderFactoryBean streamsBuilderFactoryBeanFromContext = applicationContext.getBean(
- "&stream-builder-" + beanNamePostPrefix, StreamsBuilderFactoryBean.class);
- //At this point, the StreamsBuilderFactoryBean is created. If the users call, getObject()
- //in the customizer, that should grant access to the StreamsBuilder.
- if (customizer != null) {
- customizer.configure(streamsBuilderFactoryBean);
- }
- return streamsBuilderFactoryBeanFromContext;
- }
-
- private void handleConcurrency(ApplicationContext applicationContext, String inboundName,
- Map streamConfiguration) {
- // This rebinding is necessary to capture the concurrency explicitly set by the application.
- // This is added to fix this issue: https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/899
- org.springframework.boot.context.properties.bind.Binder explicitConcurrencyResolver =
- new org.springframework.boot.context.properties.bind.Binder(ConfigurationPropertySources.get(applicationContext.getEnvironment()),
- new PropertySourcesPlaceholdersResolver(applicationContext.getEnvironment()),
- IntegrationUtils.getConversionService(this.applicationContext.getBeanFactory()), null);
-
- boolean[] concurrencyExplicitlyProvided = new boolean[] {false};
- BindHandler handler = new BindHandler() {
-
- @Override
- public Object onSuccess(ConfigurationPropertyName name, Bindable target,
- BindContext context, Object result) {
- if (!concurrencyExplicitlyProvided[0]) {
- concurrencyExplicitlyProvided[0] = name.getLastElement(ConfigurationPropertyName.Form.UNIFORM).equals("concurrency") &&
- // name is normalized to contain only uniform elements and thus safe to call toLowerCase here.
- ConfigurationPropertyName.of("spring.cloud.stream.bindings." + inboundName.toLowerCase() + ".consumer")
- .isAncestorOf(name);
- }
- return result;
- }
- };
- //Re-bind spring.cloud.stream properties to check if the application explicitly provided concurrency.
- try {
- explicitConcurrencyResolver.bind("spring.cloud.stream",
- Bindable.ofInstance(new BindingServiceProperties()), handler);
- }
- catch (Exception e) {
- // Ignore this exception
- }
-
- int concurrency = this.bindingServiceProperties.getConsumerProperties(inboundName)
- .getConcurrency();
- // override concurrency if set at the individual binding level.
- // Concurrency will be mapped to num.stream.threads.
- // This conditional also takes into account explicit concurrency settings left at the default value of 1
- // by the application to address concurrency behavior in applications with multiple processors.
- // See this GH issue: https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/844
- if (concurrency >= 1 && concurrencyExplicitlyProvided[0]) {
- streamConfiguration.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG,
- concurrency);
- }
- }
-
- protected Serde getValueSerde(String inboundName, KafkaStreamsConsumerProperties kafkaStreamsConsumerProperties, ResolvableType resolvableType) {
- if (bindingServiceProperties.getConsumerProperties(inboundName).isUseNativeDecoding()) {
- BindingProperties bindingProperties = this.bindingServiceProperties
- .getBindingProperties(inboundName);
- return this.keyValueSerdeResolver.getInboundValueSerde(
- bindingProperties.getConsumer(), kafkaStreamsConsumerProperties, resolvableType);
- }
- else {
- return Serdes.ByteArray();
- }
- }
-
- @SuppressWarnings({"unchecked"})
- protected KStream getKStream(String inboundName, BindingProperties bindingProperties, KafkaStreamsConsumerProperties kafkaStreamsConsumerProperties,
- StreamsBuilder streamsBuilder, Serde keySerde, Serde valueSerde, Topology.AutoOffsetReset autoOffsetReset, boolean firstBuild) {
- if (firstBuild) {
- addStateStoreBeans(streamsBuilder);
- }
-
- final boolean nativeDecoding = this.bindingServiceProperties
- .getConsumerProperties(inboundName).isUseNativeDecoding();
- if (nativeDecoding) {
- LOG.info("Native decoding is enabled for " + inboundName
- + ". Inbound deserialization done at the broker.");
- }
- else {
- LOG.info("Native decoding is disabled for " + inboundName
- + ". Inbound message conversion done by Spring Cloud Stream.");
- }
-
- KStream stream;
- final Serde valueSerdeToUse = getValueSerdeToUse(kafkaStreamsConsumerProperties, valueSerde);
- final Consumed consumed = getConsumed(kafkaStreamsConsumerProperties, keySerde, valueSerdeToUse, autoOffsetReset);
-
- if (this.kafkaStreamsExtendedBindingProperties
- .getExtendedConsumerProperties(inboundName).isDestinationIsPattern()) {
- final Pattern pattern = Pattern.compile(this.bindingServiceProperties.getBindingDestination(inboundName));
-
- stream = streamsBuilder.stream(pattern, consumed);
- }
- else {
- String[] bindingTargets = StringUtils.commaDelimitedListToStringArray(
- this.bindingServiceProperties.getBindingDestination(inboundName));
- stream = streamsBuilder.stream(Arrays.asList(bindingTargets),
- consumed);
- }
- //Check to see if event type based routing is enabled.
- //See this issue for more context: https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1003
- if (StringUtils.hasText(kafkaStreamsConsumerProperties.getEventTypes())) {
- AtomicBoolean matched = new AtomicBoolean();
- AtomicReference topicObject = new AtomicReference<>();
- AtomicReference headersObject = new AtomicReference<>();
- // Processor to retrieve the header value.
- stream.process(() -> eventTypeProcessor(kafkaStreamsConsumerProperties, matched, topicObject, headersObject));
- // Branching based on event type match.
- final KStream[] branch = stream.branch((key, value) -> matched.getAndSet(false));
- // Deserialize if we have a branch from above.
- final KStream deserializedKStream = !kafkaStreamsConsumerProperties.isUseConfiguredSerdeWhenRoutingEvents() ?
- branch[0].mapValues(value -> valueSerde.deserializer().deserialize(
- topicObject.get(), headersObject.get(), ((Bytes) value).get())) : (KStream) branch[0];
- return getkStream(bindingProperties, deserializedKStream, nativeDecoding);
- }
- return getkStream(bindingProperties, stream, nativeDecoding);
- }
-
- private Serde getValueSerdeToUse(KafkaStreamsConsumerProperties kafkaStreamsConsumerProperties, Serde valueSerde) {
- if (StringUtils.hasText(kafkaStreamsConsumerProperties.getEventTypes())) {
- return kafkaStreamsConsumerProperties.isUseConfiguredSerdeWhenRoutingEvents() ? valueSerde : new Serdes.BytesSerde();
- }
- return valueSerde;
- }
-
- private KStream getkStream(BindingProperties bindingProperties, KStream stream, boolean nativeDecoding) {
- if (!nativeDecoding) {
- AtomicReference headersAtomicReference = new AtomicReference<>();
- stream.process((ProcessorSupplier) () -> new Processor() {
-
- @Override
- public void init(ProcessorContext context) {
- Processor.super.init(context);
- }
-
- @Override
- public void process(Record record) {
- final Headers headers = record.headers();
- headersAtomicReference.set(headers);
- }
-
- @Override
- public void close() {
- Processor.super.close();
- }
- });
- stream = stream.mapValues((value) -> {
- Object returnValue;
- String contentType = bindingProperties.getContentType();
- if (value != null && StringUtils.hasText(contentType)) {
- final Headers headers = headersAtomicReference.get();
- final Map headersMap = new HashMap<>();
- headers.forEach(header -> headersMap.put(header.key(), header.value()));
- returnValue = MessageBuilder.withPayload(value).copyHeaders(headersMap)
- .setHeader(MessageHeaders.CONTENT_TYPE, contentType).build();
- }
- else {
- returnValue = value;
- }
- return returnValue;
- });
- }
- return stream;
- }
-
- @SuppressWarnings("rawtypes")
- private void addStateStoreBeans(StreamsBuilder streamsBuilder) {
- try {
- final Map storeBuilders = applicationContext.getBeansOfType(StoreBuilder.class);
- if (!CollectionUtils.isEmpty(storeBuilders)) {
- storeBuilders.values().forEach(storeBuilder -> {
- streamsBuilder.addStateStore(storeBuilder);
- if (LOG.isInfoEnabled()) {
- LOG.info("state store " + storeBuilder.name() + " added to topology");
- }
- });
- }
- }
- catch (Exception e) {
- // Pass through.
- }
- }
-
- private KTable materializedAs(StreamsBuilder streamsBuilder, String destination, String storeName,
- Serde k, Serde v, Topology.AutoOffsetReset autoOffsetReset, KafkaStreamsConsumerProperties kafkaStreamsConsumerProperties) {
-
- final Consumed consumed = getConsumed(kafkaStreamsConsumerProperties, k, v, autoOffsetReset);
- return streamsBuilder.table(this.bindingServiceProperties.getBindingDestination(destination),
- consumed, getMaterialized(storeName, k, v));
- }
-
- private Materialized> getMaterialized(
- String storeName, Serde k, Serde v) {
- return Materialized.>as(storeName)
- .withKeySerde(k).withValueSerde(v);
- }
-
- private GlobalKTable materializedAsGlobalKTable(
- StreamsBuilder streamsBuilder, String destination, String storeName,
- Serde k, Serde