From e5d0d1483697639e38ac0548477e56e4cb5236f7 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Thu, 12 Jun 2025 11:31:36 -0400 Subject: [PATCH] Remove reactive-kafka binder Since the reactor-kafka project has been discontined, we are removing reactive Kafka binder from Spring Cloud Stream which was already deprecated on the 4.3.x line. See: https://spring.io/blog/2025/05/20/reactor-kafka-discontinued Signed-off-by: Soby Chacko --- binders/kafka-binder/pom.xml | 1 - .../pom.xml | 122 ----- .../reactorkafka/ReactorKafkaBinder.java | 458 ------------------ .../ReactorKafkaBinderConfiguration.java | 104 ---- .../ReactorKafkaBinderHealthIndicator.java | 94 ---- ...fkaBinderHealthIndicatorConfiguration.java | 75 --- .../ReceiverOptionsCustomizer.java | 53 -- .../reactorkafka/SenderOptionsCustomizer.java | 53 -- .../binder/reactorkafka/package-info.java | 20 - .../main/resources/META-INF/spring.binders | 2 - .../main/resources/META-INF/spring.factories | 2 - ...eactorKafkaBinderHealthIndicatorTests.java | 100 ---- .../ReactorKafkaBinderIntegrationTests.java | 260 ---------- .../ReactorKafkaBinderObservationTests.java | 163 ------- .../reactorkafka/ReactorKafkaBinderTests.java | 353 -------------- .../src/test/resources/logback.xml | 20 - docs/modules/ROOT/nav.adoc | 1 - .../pages/kafka/kafka_reactive_binder.adoc | 223 --------- 18 files changed, 2104 deletions(-) delete mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/pom.xml delete mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java delete mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderConfiguration.java delete mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderHealthIndicator.java delete mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderHealthIndicatorConfiguration.java delete mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReceiverOptionsCustomizer.java delete mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/SenderOptionsCustomizer.java delete mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/package-info.java delete mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/resources/META-INF/spring.binders delete mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/resources/META-INF/spring.factories delete mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderHealthIndicatorTests.java delete mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderIntegrationTests.java delete mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderObservationTests.java delete mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderTests.java delete mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/resources/logback.xml delete mode 100644 docs/modules/ROOT/pages/kafka/kafka_reactive_binder.adoc diff --git a/binders/kafka-binder/pom.xml b/binders/kafka-binder/pom.xml index 96cf1a1b4..aa728ea3d 100644 --- a/binders/kafka-binder/pom.xml +++ b/binders/kafka-binder/pom.xml @@ -29,7 +29,6 @@ spring-cloud-starter-stream-kafka spring-cloud-stream-binder-kafka-core spring-cloud-stream-binder-kafka-streams - spring-cloud-stream-binder-kafka-reactive diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/pom.xml b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/pom.xml deleted file mode 100644 index 10470d574..000000000 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/pom.xml +++ /dev/null @@ -1,122 +0,0 @@ - - - 4.0.0 - - spring-cloud-stream-binder-kafka-reactive - jar - spring-cloud-stream-binder-kafka-reactive - Kafka binder implementation - - - org.springframework.cloud - spring-cloud-stream-binder-kafka-parent - 5.0.0-SNAPSHOT - - - - - 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.cloud - spring-cloud-stream - - - org.springframework.boot - spring-boot-autoconfigure - true - - - org.apache.kafka - kafka-clients - - - org.springframework.kafka - spring-kafka - - - io.projectreactor.kafka - reactor-kafka - - - org.springframework.boot - spring-boot-test - test - - - org.springframework.kafka - spring-kafka-test - test - - - org.springframework.boot - spring-boot-starter-test - test - - - - org.apache.kafka - kafka-clients - test - - - org.apache.kafka - kafka_2.13 - - - org.apache.kafka - kafka_2.13 - test - - - org.awaitility - awaitility - test - - - io.micrometer - context-propagation - optional - - - - io.micrometer - micrometer-tracing-integration-test - test - - - io.opentelemetry - * - - - com.wavefront - * - - - io.zipkin.reporter2 - * - - - io.micrometer - micrometer-tracing-bridge-otel - - - io.micrometer - micrometer-tracing-reporter-wavefront - - - - - - diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java deleted file mode 100644 index efda3fd65..000000000 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java +++ /dev/null @@ -1,458 +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.Arrays; -import java.util.Collection; -import java.util.List; -import java.util.Map; -import java.util.UUID; -import java.util.concurrent.ConcurrentHashMap; -import java.util.regex.Pattern; - -import io.micrometer.observation.ObservationRegistry; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.apache.kafka.clients.consumer.ConsumerRecord; -import org.apache.kafka.clients.producer.Producer; -import org.apache.kafka.clients.producer.ProducerRecord; -import org.apache.kafka.common.PartitionInfo; -import reactor.core.publisher.Flux; -import reactor.kafka.receiver.KafkaReceiver; -import reactor.kafka.receiver.ReceiverOptions; -import reactor.kafka.receiver.ReceiverRecord; -import reactor.kafka.sender.KafkaSender; -import reactor.kafka.sender.SenderOptions; -import reactor.kafka.sender.SenderRecord; -import reactor.kafka.sender.SenderResult; - -import org.springframework.beans.factory.ObjectProvider; -import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder; -import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider; -import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; -import org.springframework.cloud.stream.binder.ExtendedProducerProperties; -import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder; -import org.springframework.cloud.stream.binder.kafka.common.BinderHeaderMapper; -import org.springframework.cloud.stream.binder.kafka.common.TopicInformation; -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.KafkaExtendedBindingProperties; -import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties; -import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner; -import org.springframework.cloud.stream.binder.kafka.support.ConsumerConfigCustomizer; -import org.springframework.cloud.stream.binder.kafka.support.ProducerConfigCustomizer; -import org.springframework.cloud.stream.binder.kafka.utils.BindingUtils; -import org.springframework.cloud.stream.provisioning.ConsumerDestination; -import org.springframework.cloud.stream.provisioning.ProducerDestination; -import org.springframework.context.Lifecycle; -import org.springframework.context.support.AbstractApplicationContext; -import org.springframework.integration.IntegrationMessageHeaderAccessor; -import org.springframework.integration.channel.FluxMessageChannel; -import org.springframework.integration.core.MessageProducer; -import org.springframework.integration.endpoint.MessageProducerSupport; -import org.springframework.integration.handler.AbstractMessageHandler; -import org.springframework.integration.support.MessageBuilder; -import org.springframework.kafka.core.DefaultKafkaConsumerFactory; -import org.springframework.kafka.core.DefaultKafkaProducerFactory; -import org.springframework.kafka.support.KafkaHeaders; -import org.springframework.kafka.support.converter.KafkaMessageHeaders; -import org.springframework.kafka.support.converter.MessageConverter; -import org.springframework.kafka.support.converter.MessagingMessageConverter; -import org.springframework.kafka.support.converter.RecordMessageConverter; -import org.springframework.lang.Nullable; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.MessageHandler; -import org.springframework.messaging.support.GenericMessage; -import org.springframework.util.Assert; -import org.springframework.util.StringUtils; - -/** - * @author Gary Russell - * @author Byungjun You - * @author Omer Celik - * @author Soby Chacko - * @since 4.0 - * - * @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 class ReactorKafkaBinder - extends AbstractMessageChannelBinder, - ExtendedProducerProperties, KafkaTopicProvisioner> - implements - ExtendedPropertiesBinder { - - private static final Log logger = LogFactory.getLog(ReactorKafkaBinder.class); - - private final KafkaBinderConfigurationProperties configurationProperties; - - private KafkaExtendedBindingProperties extendedBindingProperties = new KafkaExtendedBindingProperties(); - - private ConsumerConfigCustomizer consumerConfigCustomizer; - - private ProducerConfigCustomizer producerConfigCustomizer; - - private ReceiverOptionsCustomizer receiverOptionsCustomizer = (name, opts) -> opts; - - private SenderOptionsCustomizer senderOptionsCustomizer = (name, opts) -> opts; - - private final Map topicsInUse = new ConcurrentHashMap<>(); - - private final Map messageProducers = new ConcurrentHashMap<>(); - - private final ObservationRegistry observationRegistry; - - public ReactorKafkaBinder(KafkaBinderConfigurationProperties configurationProperties, - KafkaTopicProvisioner provisioner, @Nullable ObservationRegistry observationRegistry) { - - super(new String[0], provisioner, null, null); - this.configurationProperties = configurationProperties; - this.observationRegistry = observationRegistry; - } - - public void setConsumerConfigCustomizer(ConsumerConfigCustomizer consumerConfigCustomizer) { - this.consumerConfigCustomizer = consumerConfigCustomizer; - } - - public void setProducerConfigCustomizer(ProducerConfigCustomizer producerConfigCustomizer) { - this.producerConfigCustomizer = producerConfigCustomizer; - } - - @SuppressWarnings({ "rawtypes", "unchecked" }) - public void receiverOptionsCustomizers(ObjectProvider customizers) { - if (customizers.getIfUnique() != null) { - this.receiverOptionsCustomizer = customizers.getIfUnique(); - } - else { - List list = customizers.orderedStream().toList(); - ReceiverOptionsCustomizer customizer = (name, opts) -> { - ReceiverOptions last = null; - for (ReceiverOptionsCustomizer cust: list) { - last = (ReceiverOptions) cust.apply(name, opts); - } - return last; - }; - if (!list.isEmpty()) { - this.receiverOptionsCustomizer = customizer; - } - } - } - - @SuppressWarnings({ "rawtypes", "unchecked" }) - public void senderOptionsCustomizers(ObjectProvider customizers) { - if (customizers.getIfUnique() != null) { - this.senderOptionsCustomizer = customizers.getIfUnique(); - } - else { - List list = customizers.orderedStream().toList(); - SenderOptionsCustomizer customizer = (name, opts) -> { - SenderOptions last = null; - for (SenderOptionsCustomizer cust: list) { - last = (SenderOptions) cust.apply(name, opts); - } - return last; - }; - if (!list.isEmpty()) { - this.senderOptionsCustomizer = customizer; - } - } - } - - @Override - protected MessageHandler createProducerMessageHandler(ProducerDestination destination, - ExtendedProducerProperties producerProperties, MessageChannel errorChannel) - throws Exception { - - Map configs = BindingUtils.createProducerConfigs(producerProperties, - this.configurationProperties); - if (this.producerConfigCustomizer != null) { - this.producerConfigCustomizer.configure(configs, producerProperties.getBindingName(), - destination.getName()); - } - Map props = BindingUtils.createProducerConfigs(producerProperties, - this.configurationProperties); - DefaultKafkaProducerFactory producerFactory = new DefaultKafkaProducerFactory<>(props); - Collection partitions = provisioningProvider.getPartitionsForTopic( - producerProperties.getPartitionCount(), false, () -> { - Producer producer = producerFactory.createProducer(); - List partitionsFor = producer - .partitionsFor(destination.getName()); - producer.close(); - return partitionsFor; - }, destination.getName()); - - this.topicsInUse.put(destination.getName(), - new TopicInformation(null, partitions, false)); - - SenderOptions opts = this.senderOptionsCustomizer.apply(producerProperties.getBindingName(), - SenderOptions.create(configs)); - if (this.configurationProperties.isEnableObservation() && this.observationRegistry != null) { - opts = opts.withObservation(this.observationRegistry); - } - // TODO bean for converter; MCB doesn't use one on the producer side. - RecordMessageConverter converter = new MessagingMessageConverter(); - AbstractApplicationContext applicationContext = getApplicationContext(); - FluxMessageChannel resultChannel = null; - String channelName = producerProperties.getExtension().getRecordMetadataChannel(); - if (channelName != null && applicationContext.containsBean(channelName)) { - resultChannel = applicationContext.getBean(channelName, FluxMessageChannel.class); - } - return new ReactorMessageHandler(opts, converter, destination.getName(), resultChannel); - } - - Map getTopicsInUse() { - return this.topicsInUse; - } - - - @Override - protected MessageProducer createConsumerEndpoint(ConsumerDestination destination, String group, - ExtendedConsumerProperties properties) { - - boolean anonymous = !StringUtils.hasText(group); - String consumerGroup = anonymous ? "anonymous." + UUID.randomUUID() : group; - Map configs = BindingUtils.createConsumerConfigs(anonymous, consumerGroup, properties, - this.configurationProperties); - - String destinations = destination.getName(); - if (this.consumerConfigCustomizer != null) { - this.consumerConfigCustomizer.configure(configs, properties.getBindingName(), destinations); - } - - MessageConverter converter = BindingUtils.getConsumerMessageConverter(getApplicationContext(), properties, - this.configurationProperties); - Assert.isInstanceOf(RecordMessageConverter.class, converter); - /* - * No need to check multiplex here because, if false, the topics are bound one-at-a-time; - * it is still required by the provisioner, however. - */ - List destList = Arrays.stream(StringUtils.commaDelimitedListToStringArray(destinations)) - .map(String::trim) - .toList(); - ReceiverOptions opts = ReceiverOptions.create(configs) - .addAssignListener(parts -> logger.info("Assigned: " + parts)); - if (properties.getExtension().isDestinationIsPattern()) { - opts = opts.subscription(Pattern.compile(destinations)); - } - else { - opts = opts.subscription(destList); - } - opts = this.receiverOptionsCustomizer.apply(properties.getBindingName(), opts); - ReceiverOptions finalOpts = opts; - - Map props = BindingUtils.createConsumerConfigs(anonymous, consumerGroup, properties, - this.configurationProperties); - - DefaultKafkaConsumerFactory factory = new DefaultKafkaConsumerFactory<>(props); - int partitionCount = properties.getInstanceCount() * properties.getConcurrency(); - boolean groupManagement = properties.getExtension().isAutoRebalanceEnabled(); - if (!properties.isMultiplex()) { - provisioningProvider.getListenedPartitions(consumerGroup, properties, factory, - partitionCount, properties.getExtension().isDestinationIsPattern(), - groupManagement, destination.getName(), topicsInUse); - } - else { - for (String name : StringUtils - .commaDelimitedListToStringArray(destination.getName())) { - provisioningProvider.getListenedPartitions(consumerGroup, properties, factory, - partitionCount, properties.getExtension().isDestinationIsPattern(), - groupManagement, name.trim(), topicsInUse); - } - } - - class ReactorMessageProducer extends MessageProducerSupport { - - private final List> receivers = new ArrayList<>(); - - ReactorMessageProducer() { - for (int i = 0; i < properties.getConcurrency(); i++) { - this.receivers.add(KafkaReceiver.create(finalOpts)); - } - } - - @SuppressWarnings("unchecked") - @Override - protected void doStart() { - List>> fluxes = new ArrayList<>(); - int concurrency = properties.getConcurrency(); - boolean autoCommit = properties.getExtension().isReactiveAutoCommit(); - boolean atMostOnce = properties.getExtension().isReactiveAtMostOnce(); - Assert.state(!(autoCommit && atMostOnce), - "Cannot set both reactiveAutoCommit and reactiveAtMostOnce"); - for (int i = 0; i < concurrency; i++) { - Flux> receive = null; - KafkaReceiver kafkaReceiver = this.receivers.get(i); - if (atMostOnce) { - receive = kafkaReceiver - .receiveAtmostOnce(); - } - else if (!autoCommit) { - receive = kafkaReceiver - .receive(); - } - if (autoCommit) { - fluxes.add(kafkaReceiver - .receiveAutoAck() - .map(inner -> new GenericMessage<>(inner))); - } - else { - fluxes.add(receive - .map(record -> { - Message message = (Message) ((RecordMessageConverter) converter) - .toMessage(record, null, null, null); - return addAckHeaderIfNeeded(atMostOnce, record, message); - })); - } - } - if (concurrency == 1) { - subscribeToPublisher(fluxes.get(0)); - } - else { - subscribeToPublisher(Flux.merge(fluxes)); - } - } - - private Message addAckHeaderIfNeeded(boolean autoCommit, ConsumerRecord record, - Message message) { - - if (!autoCommit) { - if (message.getHeaders() instanceof KafkaMessageHeaders headers) { - headers.getRawHeaders().put(KafkaHeaders.ACKNOWLEDGMENT, - ((ReceiverRecord) record).receiverOffset()); - } - else { - message = MessageBuilder.fromMessage(message) - .setHeader(KafkaHeaders.ACKNOWLEDGMENT, - ((ReceiverRecord) record).receiverOffset()) - .build(); - } - } - return message; - } - - } - ReactorMessageProducer reactorMessageProducer = new ReactorMessageProducer(); - this.messageProducers.put(consumerGroup, reactorMessageProducer); - return reactorMessageProducer; - } - - public Map getMessageProducers() { - return this.messageProducers; - } - - @Override - public KafkaConsumerProperties getExtendedConsumerProperties(String channelName) { - return this.extendedBindingProperties.getExtendedConsumerProperties(channelName); - } - - @Override - public KafkaProducerProperties getExtendedProducerProperties(String channelName) { - return this.extendedBindingProperties.getExtendedProducerProperties(channelName); - } - - @Override - public String getDefaultsPrefix() { - return this.extendedBindingProperties.getDefaultsPrefix(); - } - - @Override - public Class getExtendedPropertiesEntryClass() { - return this.extendedBindingProperties.getExtendedPropertiesEntryClass(); - } - - public void setExtendedBindingProperties( - KafkaExtendedBindingProperties extendedBindingProperties) { - - this.extendedBindingProperties = extendedBindingProperties; - } - - private static class ReactorMessageHandler extends AbstractMessageHandler implements Lifecycle { - - private final RecordMessageConverter converter; - - private final String topic; - - private final SenderOptions senderOptions; - - @Nullable - private final FluxMessageChannel results; - - private volatile KafkaSender sender; - - private volatile boolean running; - - ReactorMessageHandler(SenderOptions opts, RecordMessageConverter converter, - String topic, @Nullable FluxMessageChannel results) { - - this.senderOptions = opts; - this.converter = converter; - ((MessagingMessageConverter) converter).setHeaderMapper(new BinderHeaderMapper()); - this.topic = topic; - this.results = results; - } - - @Override - protected void handleMessageInternal(Message message) { - if (this.sender != null) { - Object correlation = message.getHeaders().get(IntegrationMessageHeaderAccessor.CORRELATION_ID); - if (correlation == null) { - correlation = UUID.randomUUID(); - } - @SuppressWarnings("unchecked") - SenderRecord sr = SenderRecord.create( - (ProducerRecord) converter.fromMessage(message, topic), correlation); - Flux> result = sender.send(Flux.just(sr)).contextCapture(); - result.subscribe(res -> { - if (this.results != null) { - this.results.send(MessageBuilder.withPayload(res) - .copyHeaders(message.getHeaders()) - .build()); - } - }); - } - } - - @Override - public synchronized void start() { - if (!this.running) { - this.sender = KafkaSender.create(this.senderOptions); - this.running = true; - } - } - - @Override - public synchronized void stop() { - if (this.running) { - KafkaSender theSender = this.sender; - this.sender = null; - theSender.close(); - this.running = false; - } - } - - @Override - public boolean isRunning() { - return this.running; - } - - } -} diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderConfiguration.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderConfiguration.java deleted file mode 100644 index 38dd1c085..000000000 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderConfiguration.java +++ /dev/null @@ -1,104 +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 io.micrometer.observation.ObservationRegistry; - -import org.springframework.beans.factory.ObjectProvider; -import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; -import org.springframework.boot.autoconfigure.kafka.KafkaConnectionDetails; -import org.springframework.boot.autoconfigure.kafka.KafkaProperties; -import org.springframework.boot.context.properties.ConfigurationProperties; -import org.springframework.boot.context.properties.EnableConfigurationProperties; -import org.springframework.cloud.stream.binder.Binder; -import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; -import org.springframework.cloud.stream.binder.kafka.properties.KafkaExtendedBindingProperties; -import org.springframework.cloud.stream.binder.kafka.provisioning.AdminClientConfigCustomizer; -import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner; -import org.springframework.cloud.stream.binder.kafka.support.ConsumerConfigCustomizer; -import org.springframework.cloud.stream.binder.kafka.support.ProducerConfigCustomizer; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import org.springframework.context.annotation.Import; - -/** - * Binder configuration for ReactorKafka. - * - * @author Gary Russell - * @author Chris Bono - * @author Soby Chacko - * - * @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) -@Configuration(proxyBeanMethods = false) -@ConditionalOnMissingBean(Binder.class) -@EnableConfigurationProperties({ KafkaProperties.class, KafkaExtendedBindingProperties.class }) -@Import({ ReactorKafkaBinderHealthIndicatorConfiguration.class }) -public class ReactorKafkaBinderConfiguration { - - /** - * @ConfigurationProperties is declared on the @Bean method for Spring Boot to ignore - * constructor binding on KafkaBinderConfigurationProperties. If constructor binding is - * used, it ignores all the JavaBeans style properties when generating configuration metadata. - * - * See the following issues for more details: - * - * https://github.com/spring-cloud/spring-cloud-stream/issues/2640 - * https://github.com/spring-projects/spring-boot/issues/34031 - * - * @param kafkaProperties Spring Kafka properties autoconfigured by Spring Boot - */ - @Bean - @ConfigurationProperties(prefix = "spring.cloud.stream.kafka.binder") - KafkaBinderConfigurationProperties configurationProperties( - KafkaProperties kafkaProperties, ObjectProvider kafkaConnectionDetails) { - return new KafkaBinderConfigurationProperties(kafkaProperties, kafkaConnectionDetails); - } - - @Bean - KafkaTopicProvisioner provisioningProvider( - KafkaBinderConfigurationProperties configurationProperties, - ObjectProvider adminClientConfigCustomizer, - KafkaProperties kafkaProperties, ObjectProvider kafkaConnectionDetails) { - return new KafkaTopicProvisioner(configurationProperties, kafkaProperties, - kafkaConnectionDetails.getIfAvailable(), adminClientConfigCustomizer.getIfUnique()); - } - - @Bean - ReactorKafkaBinder reactorKafkaBinder(KafkaBinderConfigurationProperties configurationProperties, - KafkaTopicProvisioner provisioningProvider, - KafkaExtendedBindingProperties extendedBindingProperties, - ObjectProvider consumerConfigCustomizer, - ObjectProvider producerConfigCustomizer, - ObjectProvider receiverOptionsCustomizers, - ObjectProvider senderOptionsptionsCustomizers, - ObjectProvider observationRegistryObjectProvider) { - ReactorKafkaBinder reactorKafkaBinder = new ReactorKafkaBinder(configurationProperties, provisioningProvider, - observationRegistryObjectProvider.getIfUnique()); - reactorKafkaBinder.setExtendedBindingProperties(extendedBindingProperties); - reactorKafkaBinder.setConsumerConfigCustomizer(consumerConfigCustomizer.getIfUnique()); - reactorKafkaBinder.setProducerConfigCustomizer(producerConfigCustomizer.getIfUnique()); - reactorKafkaBinder.receiverOptionsCustomizers(receiverOptionsCustomizers); - reactorKafkaBinder.senderOptionsCustomizers(senderOptionsptionsCustomizers); - return reactorKafkaBinder; - } - -} diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderHealthIndicator.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderHealthIndicator.java deleted file mode 100644 index dca4a7480..000000000 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderHealthIndicator.java +++ /dev/null @@ -1,94 +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.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; - -import org.springframework.boot.actuate.health.Health; -import org.springframework.boot.actuate.health.Status; -import org.springframework.cloud.stream.binder.kafka.common.AbstractKafkaBinderHealthIndicator; -import org.springframework.cloud.stream.binder.kafka.common.TopicInformation; -import org.springframework.integration.endpoint.MessageProducerSupport; -import org.springframework.kafka.core.ConsumerFactory; -import org.springframework.scheduling.concurrent.CustomizableThreadFactory; - -/** - * {@link org.springframework.boot.actuate.health.HealthIndicator} for Reactor Kafka Binder. - * - * @author Soby Chacko - * - * @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 class ReactorKafkaBinderHealthIndicator extends AbstractKafkaBinderHealthIndicator { - - private final ReactorKafkaBinder binder; - - public ReactorKafkaBinderHealthIndicator(ReactorKafkaBinder binder, ConsumerFactory consumerFactory) { - super(consumerFactory); - this.binder = binder; - } - - @Override - protected ExecutorService createHealthBinderExecutorService() { - return Executors.newSingleThreadExecutor( - new CustomizableThreadFactory("reactor-kafka-binder-health-")); - } - - @Override - protected Map getTopicsInUse() { - return this.binder.getTopicsInUse(); - } - - @Override - protected Health buildBinderSpecificHealthDetails() { - Map messageProducerSupportInfo = binder.getMessageProducers(); - if (messageProducerSupportInfo.isEmpty()) { - return Health.unknown().build(); - } - - Status status = Status.UP; - List> messageProducers = new ArrayList<>(); - - Map 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/ReactorKafkaBinderHealthIndicatorConfiguration.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderHealthIndicatorConfiguration.java deleted file mode 100644 index d7177dc80..000000000 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderHealthIndicatorConfiguration.java +++ /dev/null @@ -1,75 +0,0 @@ -/* - * Copyright 2024-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.HashMap; -import java.util.Map; - -import org.apache.kafka.clients.consumer.ConsumerConfig; -import org.apache.kafka.common.serialization.ByteArrayDeserializer; - -import org.springframework.boot.actuate.autoconfigure.health.ConditionalOnEnabledHealthIndicator; -import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; -import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import org.springframework.kafka.core.ConsumerFactory; -import org.springframework.kafka.core.DefaultKafkaConsumerFactory; -import org.springframework.util.ObjectUtils; - -/** - * @author Soby Chacko - * @since 4.1.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) -@Configuration(proxyBeanMethods = false) -@ConditionalOnClass(name = "org.springframework.boot.actuate.health.HealthIndicator") -@ConditionalOnEnabledHealthIndicator("binders") -public class ReactorKafkaBinderHealthIndicatorConfiguration { - - @Bean - public ReactorKafkaBinderHealthIndicator reactorKafkaBinderHealthIndicator( - ReactorKafkaBinder reactorKafkaBinder, - KafkaBinderConfigurationProperties configurationProperties) { - Map props = new HashMap<>(); - props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, - ByteArrayDeserializer.class); - props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, - ByteArrayDeserializer.class); - Map 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.