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