From f3ccc7b72de7e7aa5279528b3be4a990225b1100 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 10 Mar 2022 13:35:29 -0500 Subject: [PATCH] GH-2293: Initial Commit Reactor-Kafka Binder Note that it requires the message channel to be a `FluxMessageChannel`. It currently only supports `Consumer>` (no conversion). The function binding should use the message channel directly instead of bridging the direct channel to a reactive channel. Producer Side PoC Binder specific changes in ReactorKafkaBinder Binder configuration More changes to the reactive binder Move KafkaBinderEnvironmentPostProcessor to the core module Change binder key in spring.binders Adding an e2e IT test for the reactive binder Add lifecycle to integration adapters. Remove unnecessary doStop(). Fix import order. --- binders/kafka-binder/pom.xml | 1 + .../KafkaBinderEnvironmentPostProcessor.java | 2 +- .../pom.xml | 90 +++++ .../reactorkafka/ReactorKafkaBinder.java | 321 ++++++++++++++++++ .../ReactorKafkaBinderConfiguration.java | 61 ++++ .../binder/reactorkafka/package-info.java | 20 ++ .../main/resources/META-INF/spring.binders | 2 + .../main/resources/META-INF/spring.factories | 2 + .../ReactorKafkaBinderIntegrationTests.java | 103 ++++++ .../reactorkafka/ReactorKafkaBinderTests.java | 141 ++++++++ .../src/test/resources/logback.xml | 20 ++ .../main/resources/META-INF/spring.factories | 2 +- 12 files changed, 763 insertions(+), 2 deletions(-) rename binders/kafka-binder/{spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka => spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/common}/KafkaBinderEnvironmentPostProcessor.java (98%) create mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/pom.xml create mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java create mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderConfiguration.java create mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/package-info.java create mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/resources/META-INF/spring.binders create mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/resources/META-INF/spring.factories create mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderIntegrationTests.java create mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderTests.java create mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/resources/logback.xml diff --git a/binders/kafka-binder/pom.xml b/binders/kafka-binder/pom.xml index 47e17fbba..bd887917a 100644 --- a/binders/kafka-binder/pom.xml +++ b/binders/kafka-binder/pom.xml @@ -34,6 +34,7 @@ spring-cloud-starter-stream-kafka spring-cloud-stream-binder-kafka-core spring-cloud-stream-binder-kafka-streams + spring-cloud-stream-binder-kafka-reactive docs diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderEnvironmentPostProcessor.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/common/KafkaBinderEnvironmentPostProcessor.java similarity index 98% rename from binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderEnvironmentPostProcessor.java rename to binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/common/KafkaBinderEnvironmentPostProcessor.java index 7fe2ef06b..203afb9b0 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderEnvironmentPostProcessor.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/common/KafkaBinderEnvironmentPostProcessor.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.cloud.stream.binder.kafka; +package org.springframework.cloud.stream.binder.kafka.common; import java.util.HashMap; import java.util.Map; 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 new file mode 100644 index 000000000..b636732ea --- /dev/null +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/pom.xml @@ -0,0 +1,90 @@ + + + 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 + 4.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 + 1.3.8 + + + org.springframework.boot + spring-boot-test + test + + + org.springframework.kafka + spring-kafka-test + test + + + org.springframework.cloud + spring-cloud-stream-binder-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 + + + + 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 new file mode 100644 index 000000000..cec65422f --- /dev/null +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java @@ -0,0 +1,321 @@ +/* + * 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. + */ + +package org.springframework.cloud.stream.binder.reactorkafka; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.clients.producer.RecordMetadata; +import org.apache.kafka.common.serialization.ByteArrayDeserializer; +import org.apache.kafka.common.serialization.ByteArraySerializer; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.core.publisher.Sinks; +import reactor.kafka.receiver.KafkaReceiver; +import reactor.kafka.receiver.ReceiverOptions; +import reactor.kafka.sender.KafkaSender; +import reactor.kafka.sender.SenderOptions; +import reactor.kafka.sender.SenderRecord; +import reactor.kafka.sender.SenderResult; + +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.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.provisioning.ConsumerDestination; +import org.springframework.cloud.stream.provisioning.ProducerDestination; +import org.springframework.context.Lifecycle; +import org.springframework.integration.core.MessageProducer; +import org.springframework.integration.endpoint.MessageProducerSupport; +import org.springframework.integration.handler.AbstractMessageHandler; +import org.springframework.kafka.support.converter.MessagingMessageConverter; +import org.springframework.kafka.support.converter.RecordMessageConverter; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; +import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; +import org.springframework.util.StringUtils; + +/** + * @author Gary Russell + * @since 4.0 + * + */ +public class ReactorKafkaBinder + extends AbstractMessageChannelBinder, + ExtendedProducerProperties, KafkaTopicProvisioner> + implements + ExtendedPropertiesBinder { + + private static final Log log = LogFactory.getLog(ReactorKafkaBinder.class); + + private final KafkaBinderConfigurationProperties configurationProperties; + + private final KafkaExtendedBindingProperties extendedBindingProperties = new KafkaExtendedBindingProperties(); + + public ReactorKafkaBinder(KafkaBinderConfigurationProperties configurationProperties, + KafkaTopicProvisioner provisioner) { + + super(new String[0], provisioner, null, null); + this.configurationProperties = configurationProperties; + } + + + @Override + protected MessageHandler createProducerMessageHandler(ProducerDestination destination, + ExtendedProducerProperties producerProperties, MessageChannel errorChannel) + throws Exception { + + Map configs = createProducerConfigs(producerProperties); + // TODO: Move config customizers to core +// if (this.producerConfigCustomizer != null) { +// this.producerConfigCustomizer.configure(props, bindingNameHolder.get(), destination); +// bindingNameHolder.remove(); +// } + + SenderOptions opts = SenderOptions.create(configs); + // TODO bean for converter. + RecordMessageConverter converter = new MessagingMessageConverter(); + return new ReactorMessageHandler(opts, converter, destination.getName()); + } + + + @Override + protected MessageProducer createConsumerEndpoint(ConsumerDestination destination, String group, + ExtendedConsumerProperties properties) throws Exception { + + boolean anonymous = !StringUtils.hasText(group); + String consumerGroup = anonymous ? "anonymous." + UUID.randomUUID().toString() : group; + Map configs = createConsumerConfigs(anonymous, consumerGroup, properties); + + // TODO: Move config customizers to core +// if (this.consumerConfigCustomizer != null) { +// this.consumerConfigCustomizer.configure(configs, bindingNameHolder.get(), destination); +// } + + RecordMessageConverter converter = new MessagingMessageConverter(); + ReceiverOptions opts = ReceiverOptions.create(configs) + .addAssignListener(parts -> System.out.println("Assigned: " + parts)) + .subscription(Collections.singletonList(destination.getName())); + + return new MessageProducerSupport() { + + private final KafkaReceiver receiver = KafkaReceiver.create(opts); + + @SuppressWarnings("unchecked") + @Override + protected void doStart() { + Flux> flux = receiver + .receive() + .map(record -> (Message) converter.toMessage(record, null, null, null)); + subscribeToPublisher(flux); + } + + }; + } + + /* + * TODO: Copied from Kafka binder - refactor to core + */ + private Map createConsumerConfigs(boolean anonymous, String consumerGroup, + ExtendedConsumerProperties consumerProperties) { + + Map props = new HashMap<>(); + props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, + ByteArrayDeserializer.class); + props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, + ByteArrayDeserializer.class); + props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); + props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100); + props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, + anonymous ? "latest" : "earliest"); + props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup); + + Map mergedConfig = this.configurationProperties + .mergedConsumerConfiguration(); + if (!ObjectUtils.isEmpty(mergedConfig)) { + props.putAll(mergedConfig); + } + if (ObjectUtils.isEmpty(props.get(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG))) { + props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, + this.configurationProperties.getKafkaConnectionString()); + } + Map config = consumerProperties.getExtension().getConfiguration(); + if (!ObjectUtils.isEmpty(config)) { + Assert.state(!config.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG), + ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG + " cannot be overridden at the binding level; " + + "use multiple binders instead"); + props.putAll(config); + } + if (!ObjectUtils.isEmpty(consumerProperties.getExtension().getStartOffset())) { + props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, + consumerProperties.getExtension().getStartOffset().name()); + } + return props; + } + + /* + * TODO: Copied from Kafka binder - refactor to core + */ + private Map createProducerConfigs( + ExtendedProducerProperties producerProperties) { + + Map props = new HashMap<>(); + props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class); + props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, + ByteArraySerializer.class); + props.put(ProducerConfig.ACKS_CONFIG, + String.valueOf(this.configurationProperties.getRequiredAcks())); + Map mergedConfig = this.configurationProperties + .mergedProducerConfiguration(); + if (!ObjectUtils.isEmpty(mergedConfig)) { + props.putAll(mergedConfig); + } + if (ObjectUtils.isEmpty(props.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG))) { + props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, + this.configurationProperties.getKafkaConnectionString()); + } + final KafkaProducerProperties kafkaProducerProperties = producerProperties.getExtension(); + if (ObjectUtils.isEmpty(props.get(ProducerConfig.BATCH_SIZE_CONFIG))) { + props.put(ProducerConfig.BATCH_SIZE_CONFIG, + String.valueOf(kafkaProducerProperties.getBufferSize())); + } + if (ObjectUtils.isEmpty(props.get(ProducerConfig.LINGER_MS_CONFIG))) { + props.put(ProducerConfig.LINGER_MS_CONFIG, + String.valueOf(kafkaProducerProperties.getBatchTimeout())); + } + if (ObjectUtils.isEmpty(props.get(ProducerConfig.COMPRESSION_TYPE_CONFIG))) { + props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, + kafkaProducerProperties.getCompressionType().toString()); + } + Map configs = producerProperties.getExtension().getConfiguration(); + Assert.state(!configs.containsKey(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG), + ProducerConfig.BOOTSTRAP_SERVERS_CONFIG + " cannot be overridden at the binding level; " + + "use multiple binders instead"); + if (!ObjectUtils.isEmpty(configs)) { + props.putAll(configs); + } + if (!ObjectUtils.isEmpty(kafkaProducerProperties.getConfiguration())) { + props.putAll(kafkaProducerProperties.getConfiguration()); + } + return props; + } + + @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(); + } + + private static class ReactorMessageHandler extends AbstractMessageHandler implements Lifecycle { + + private final RecordMessageConverter converter; + + private final String topic; + + private final SenderOptions senderOptions; + + private volatile KafkaSender sender; + + private volatile boolean running; + + ReactorMessageHandler(SenderOptions opts, RecordMessageConverter converter, + String topic) { + + this.senderOptions = opts; + this.converter = converter; + this.topic = topic; + } + + @Override + protected void handleMessageInternal(Message message) { + Object sendResultHeader = message.getHeaders().get("sendResult"); + Sinks.One sink = Sinks.one(); + if (sendResultHeader instanceof AtomicReference) { + @SuppressWarnings("unchecked") + AtomicReference> result = + (AtomicReference>) sendResultHeader; + result.set(sink.asMono()); + } + if (this.sender != null) { + UUID uuid = UUID.randomUUID(); + @SuppressWarnings("unchecked") + SenderRecord sr = SenderRecord.create( + (ProducerRecord) converter.fromMessage(message, topic), uuid); + Flux> result = sender.send(Flux.just(sr)); + result.subscribe(res -> sink.emitValue(res.recordMetadata(), null)); + } + else { + sink.emitError(new IllegalStateException("Handler is not running"), null); + } + } + + @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 new file mode 100644 index 000000000..f398b9e9e --- /dev/null +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderConfiguration.java @@ -0,0 +1,61 @@ +/* + * Copyright 2022-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. + */ + +package org.springframework.cloud.stream.binder.reactorkafka; + +import org.springframework.beans.factory.ObjectProvider; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration; +import org.springframework.boot.autoconfigure.kafka.KafkaProperties; +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.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; + +@Configuration(proxyBeanMethods = false) +@ConditionalOnMissingBean(Binder.class) +@Import({ KafkaAutoConfiguration.class }) +@EnableConfigurationProperties({ KafkaExtendedBindingProperties.class }) +public class ReactorKafkaBinderConfiguration { + + @Bean + KafkaBinderConfigurationProperties configurationProperties( + KafkaProperties kafkaProperties) { + return new KafkaBinderConfigurationProperties(kafkaProperties); + } + + @Bean + KafkaTopicProvisioner provisioningProvider( + KafkaBinderConfigurationProperties configurationProperties, + ObjectProvider adminClientConfigCustomizer, KafkaProperties kafkaProperties) { + return new KafkaTopicProvisioner(configurationProperties, + kafkaProperties, adminClientConfigCustomizer.getIfUnique()); + } + + @Bean + ReactorKafkaBinder reactorKafkaBinder(KafkaBinderConfigurationProperties configurationProperties, + KafkaTopicProvisioner provisioningProvider) { + + ReactorKafkaBinder reactorKafkaBinder = new ReactorKafkaBinder(configurationProperties, provisioningProvider); + return reactorKafkaBinder; + } + +} 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 new file mode 100644 index 000000000..a12de78b6 --- /dev/null +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/package-info.java @@ -0,0 +1,20 @@ +/* + * 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 new file mode 100644 index 000000000..ee8843881 --- /dev/null +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/resources/META-INF/spring.binders @@ -0,0 +1,2 @@ +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 new file mode 100644 index 000000000..dd6e28b8a --- /dev/null +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/resources/META-INF/spring.factories @@ -0,0 +1,2 @@ +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/ReactorKafkaBinderIntegrationTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderIntegrationTests.java new file mode 100644 index 000000000..b548a25c4 --- /dev/null +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderIntegrationTests.java @@ -0,0 +1,103 @@ +/* + * Copyright 2022-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. + */ + +package org.springframework.cloud.stream.binder.reactorkafka; + +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.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Flux; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.test.EmbeddedKafkaBroker; +import org.springframework.kafka.test.condition.EmbeddedKafkaCondition; +import org.springframework.kafka.test.context.EmbeddedKafka; +import org.springframework.kafka.test.utils.KafkaTestUtils; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author Soby Chacko + * @author Gary Russell + */ +@EmbeddedKafka(topics = "uppercased-words") +public class ReactorKafkaBinderIntegrationTests { + + private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); + + private static Consumer consumer; + + @BeforeAll + public static void setUp() { + Map consumerProps = KafkaTestUtils.consumerProps("group", "false", + embeddedKafka); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); + consumer = cf.createConsumer(); + embeddedKafka.consumeFromEmbeddedTopics(consumer, "uppercased-words"); + } + + @Test + void testEndtoEndReactorKafkaBinder() throws Exception { + SpringApplication app = new SpringApplication(ReactiveKafkaApplication.class); + app.setWebApplicationType(WebApplicationType.NONE); + + try (ConfigurableApplicationContext context = app.run( + "--server.port=0", + "--spring.jmx.enabled=false", + "--spring.cloud.stream.function.reactive.uppercase=true", + "--spring.cloud.stream.bindings.uppercase-in-0.destination=words", + "--spring.cloud.stream.bindings.uppercase-out-0.destination=uppercased-words", + "--spring.cloud.stream.kafka.binder.brokers=" + embeddedKafka.getBrokersAsString())) { + + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + try { + KafkaTemplate template = new KafkaTemplate<>(pf, true); + template.setDefaultTopic("words"); + template.sendDefault("foobar"); + + ConsumerRecord cr = KafkaTestUtils.getSingleRecord(consumer, "uppercased-words"); + assertThat(cr.value().equals("FOOBAR")).isTrue(); + } + finally { + pf.destroy(); + } + } + } + + @EnableAutoConfiguration + public static class ReactiveKafkaApplication { + + @Bean + public Function, Flux> uppercase() { + return s -> s.map(String::toUpperCase); + } + + } +} 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 new file mode 100644 index 000000000..82b93997f --- /dev/null +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderTests.java @@ -0,0 +1,141 @@ +/* + * 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. + */ + +package org.springframework.cloud.stream.binder.reactorkafka; + +import java.util.Collections; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.kafka.clients.producer.RecordMetadata; +import org.junit.jupiter.api.Test; +import org.reactivestreams.Subscriber; +import org.reactivestreams.Subscription; +import reactor.core.publisher.Mono; + +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.channel.FluxMessageChannel; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +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.retry.support.RetryTemplate; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; + +/** + * @author Gary Russell + * @since 4.0 + * + */ +@EmbeddedKafka(topics = { "testC", "testP" }) +public class ReactorKafkaBinderTests { + + @SuppressWarnings({ "rawtypes", "unchecked" }) + @Test + void consumerBinding() throws Exception { + KafkaProperties kafkaProperties = new KafkaProperties(); + kafkaProperties.setBootstrapServers( + Collections.singletonList(EmbeddedKafkaCondition.getBroker().getBrokersAsString())); + KafkaBinderConfigurationProperties binderProps = new KafkaBinderConfigurationProperties(kafkaProperties); + KafkaTopicProvisioner provisioner = new KafkaTopicProvisioner(binderProps, kafkaProperties, null); + ReactorKafkaBinder binder = new ReactorKafkaBinder(binderProps, provisioner); + binder.setApplicationContext(mock(GenericApplicationContext.class)); + + CountDownLatch latch = new CountDownLatch(1); + + FluxMessageChannel inbound = new FluxMessageChannel(); + Subscriber> sub = new Subscriber>() { + + @Override + public void onSubscribe(Subscription s) { + s.request(1); + } + + @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); + + Binding consumer = binder.bindConsumer("testC", "foo", inbound, props); + + DefaultKafkaProducerFactory pf = + new DefaultKafkaProducerFactory<>(KafkaTestUtils.producerProps(EmbeddedKafkaCondition.getBroker())); + KafkaTemplate kt = new KafkaTemplate<>(pf); + kt.send("testC", "foo").get(10, TimeUnit.SECONDS); + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + consumer.unbind(); + pf.destroy(); + } + + @Test + void producerBinding() throws InterruptedException { + KafkaProperties kafkaProperties = new KafkaProperties(); + kafkaProperties.setBootstrapServers( + Collections.singletonList(EmbeddedKafkaCondition.getBroker().getBrokersAsString())); + KafkaBinderConfigurationProperties binderProps = new KafkaBinderConfigurationProperties(kafkaProperties); + KafkaTopicProvisioner provisioner = new KafkaTopicProvisioner(binderProps, kafkaProperties, null); + provisioner.setMetadataRetryOperations(new RetryTemplate()); + ReactorKafkaBinder binder = new ReactorKafkaBinder(binderProps, provisioner); + binder.setApplicationContext(new GenericApplicationContext()); + + MessageChannel outbound = new FluxMessageChannel(); + KafkaProducerProperties ext = new KafkaProducerProperties(); + ExtendedProducerProperties props = + new ExtendedProducerProperties(ext); + + Binding bindProducer = binder.bindProducer("testP", outbound, props); + AtomicReference> sendResult = new AtomicReference<>(); + outbound.send(MessageBuilder.withPayload("foo") + .setHeader("sendResult", sendResult) + .build()); + CountDownLatch latch = new CountDownLatch(1); + sendResult.get().doOnNext(rmd -> { + latch.countDown(); + }).subscribe(); + assertThat(latch.await(10, TimeUnit.SECONDS)).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 new file mode 100644 index 000000000..32c681652 --- /dev/null +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/resources/logback.xml @@ -0,0 +1,20 @@ + + + + %d{ISO8601} %5p %t %c{2}:%L - %m%n + + + + + + + + + + + + + + + + diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring.factories b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring.factories index 1d2e61ded..c6b85b146 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring.factories +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring.factories @@ -1,4 +1,4 @@ org.springframework.boot.env.EnvironmentPostProcessor:\ -org.springframework.cloud.stream.binder.kafka.KafkaBinderEnvironmentPostProcessor +org.springframework.cloud.stream.binder.kafka.common.KafkaBinderEnvironmentPostProcessor org.springframework.boot.autoconfigure.EnableAutoConfiguration:\ org.springframework.cloud.stream.binder.kafka.config.ExtendedBindingHandlerMappingsProviderConfiguration