GH-2293: Initial Commit Reactor-Kafka Binder
Note that it requires the message channel to be a `FluxMessageChannel`. It currently only supports `Consumer<ReceiverRecord<?, ?>>` (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.
This commit is contained in:
Gary Russell
committed by
Soby Chacko
Soby Chacko
parent
2f6de9263e
commit
f3ccc7b72d
@@ -34,6 +34,7 @@
|
||||
<module>spring-cloud-starter-stream-kafka</module>
|
||||
<module>spring-cloud-stream-binder-kafka-core</module>
|
||||
<module>spring-cloud-stream-binder-kafka-streams</module>
|
||||
<module>spring-cloud-stream-binder-kafka-reactive</module>
|
||||
<module>docs</module>
|
||||
</modules>
|
||||
|
||||
|
||||
@@ -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;
|
||||
@@ -0,0 +1,90 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<artifactId>spring-cloud-stream-binder-kafka-reactive</artifactId>
|
||||
<packaging>jar</packaging>
|
||||
<name>spring-cloud-stream-binder-kafka-reactive</name>
|
||||
<description>Kafka binder implementation</description>
|
||||
|
||||
<parent>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
|
||||
<version>4.0.0-SNAPSHOT</version>
|
||||
</parent>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka-core</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-actuator</artifactId>
|
||||
<optional>true</optional>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-configuration-processor</artifactId>
|
||||
<optional>true</optional>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-autoconfigure</artifactId>
|
||||
<optional>true</optional>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.kafka</groupId>
|
||||
<artifactId>kafka-clients</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.kafka</groupId>
|
||||
<artifactId>spring-kafka</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>io.projectreactor.kafka</groupId>
|
||||
<artifactId>reactor-kafka</artifactId>
|
||||
<version>1.3.8</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.kafka</groupId>
|
||||
<artifactId>spring-kafka-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binder-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<!-- Following dependencies are needed to support Kafka 1.1.0 client-->
|
||||
<dependency>
|
||||
<groupId>org.apache.kafka</groupId>
|
||||
<artifactId>kafka-clients</artifactId>
|
||||
<classifier>test</classifier>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.kafka</groupId>
|
||||
<artifactId>kafka_2.13</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.kafka</groupId>
|
||||
<artifactId>kafka_2.13</artifactId>
|
||||
<classifier>test</classifier>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.awaitility</groupId>
|
||||
<artifactId>awaitility</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
</project>
|
||||
@@ -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<ExtendedConsumerProperties<KafkaConsumerProperties>,
|
||||
ExtendedProducerProperties<KafkaProducerProperties>, KafkaTopicProvisioner>
|
||||
implements
|
||||
ExtendedPropertiesBinder<MessageChannel, KafkaConsumerProperties, KafkaProducerProperties> {
|
||||
|
||||
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<KafkaProducerProperties> producerProperties, MessageChannel errorChannel)
|
||||
throws Exception {
|
||||
|
||||
Map<String, Object> configs = createProducerConfigs(producerProperties);
|
||||
// TODO: Move config customizers to core
|
||||
// if (this.producerConfigCustomizer != null) {
|
||||
// this.producerConfigCustomizer.configure(props, bindingNameHolder.get(), destination);
|
||||
// bindingNameHolder.remove();
|
||||
// }
|
||||
|
||||
SenderOptions<Object, Object> 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<KafkaConsumerProperties> properties) throws Exception {
|
||||
|
||||
boolean anonymous = !StringUtils.hasText(group);
|
||||
String consumerGroup = anonymous ? "anonymous." + UUID.randomUUID().toString() : group;
|
||||
Map<String, Object> 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<Object, Object> opts = ReceiverOptions.create(configs)
|
||||
.addAssignListener(parts -> System.out.println("Assigned: " + parts))
|
||||
.subscription(Collections.singletonList(destination.getName()));
|
||||
|
||||
return new MessageProducerSupport() {
|
||||
|
||||
private final KafkaReceiver<Object, Object> receiver = KafkaReceiver.create(opts);
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
@Override
|
||||
protected void doStart() {
|
||||
Flux<Message<Object>> flux = receiver
|
||||
.receive()
|
||||
.map(record -> (Message<Object>) converter.toMessage(record, null, null, null));
|
||||
subscribeToPublisher(flux);
|
||||
}
|
||||
|
||||
};
|
||||
}
|
||||
|
||||
/*
|
||||
* TODO: Copied from Kafka binder - refactor to core
|
||||
*/
|
||||
private Map<String, Object> createConsumerConfigs(boolean anonymous, String consumerGroup,
|
||||
ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties) {
|
||||
|
||||
Map<String, Object> 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<String, Object> 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<String, String> 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<String, Object> createProducerConfigs(
|
||||
ExtendedProducerProperties<KafkaProducerProperties> producerProperties) {
|
||||
|
||||
Map<String, Object> 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<String, Object> 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<String, String> 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<? extends BinderSpecificPropertiesProvider> getExtendedPropertiesEntryClass() {
|
||||
return this.extendedBindingProperties.getExtendedPropertiesEntryClass();
|
||||
}
|
||||
|
||||
private static class ReactorMessageHandler extends AbstractMessageHandler implements Lifecycle {
|
||||
|
||||
private final RecordMessageConverter converter;
|
||||
|
||||
private final String topic;
|
||||
|
||||
private final SenderOptions<Object, Object> senderOptions;
|
||||
|
||||
private volatile KafkaSender<Object, Object> sender;
|
||||
|
||||
private volatile boolean running;
|
||||
|
||||
ReactorMessageHandler(SenderOptions<Object, Object> 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<RecordMetadata> sink = Sinks.one();
|
||||
if (sendResultHeader instanceof AtomicReference) {
|
||||
@SuppressWarnings("unchecked")
|
||||
AtomicReference<Mono<RecordMetadata>> result =
|
||||
(AtomicReference<Mono<RecordMetadata>>) sendResultHeader;
|
||||
result.set(sink.asMono());
|
||||
}
|
||||
if (this.sender != null) {
|
||||
UUID uuid = UUID.randomUUID();
|
||||
@SuppressWarnings("unchecked")
|
||||
SenderRecord<Object, Object, UUID> sr = SenderRecord.create(
|
||||
(ProducerRecord<Object, Object>) converter.fromMessage(message, topic), uuid);
|
||||
Flux<SenderResult<UUID>> 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<Object, Object> theSender = this.sender;
|
||||
this.sender = null;
|
||||
theSender.close();
|
||||
this.running = false;
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isRunning() {
|
||||
return this.running;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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> 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;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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;
|
||||
@@ -0,0 +1,2 @@
|
||||
reactorKafka:\
|
||||
org.springframework.cloud.stream.binder.reactorkafka.ReactorKafkaBinderConfiguration
|
||||
@@ -0,0 +1,2 @@
|
||||
org.springframework.boot.env.EnvironmentPostProcessor:\
|
||||
org.springframework.cloud.stream.binder.kafka.common.KafkaBinderEnvironmentPostProcessor
|
||||
@@ -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<String, String> consumer;
|
||||
|
||||
@BeforeAll
|
||||
public static void setUp() {
|
||||
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("group", "false",
|
||||
embeddedKafka);
|
||||
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
|
||||
DefaultKafkaConsumerFactory<String, String> 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<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
|
||||
DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
|
||||
try {
|
||||
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf, true);
|
||||
template.setDefaultTopic("words");
|
||||
template.sendDefault("foobar");
|
||||
|
||||
ConsumerRecord<String, String> cr = KafkaTestUtils.getSingleRecord(consumer, "uppercased-words");
|
||||
assertThat(cr.value().equals("FOOBAR")).isTrue();
|
||||
}
|
||||
finally {
|
||||
pf.destroy();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@EnableAutoConfiguration
|
||||
public static class ReactiveKafkaApplication {
|
||||
|
||||
@Bean
|
||||
public Function<Flux<String>, Flux<String>> uppercase() {
|
||||
return s -> s.map(String::toUpperCase);
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
@@ -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<Message<?>> sub = new Subscriber<Message<?>>() {
|
||||
|
||||
@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<KafkaConsumerProperties> props =
|
||||
new ExtendedConsumerProperties<KafkaConsumerProperties>(ext);
|
||||
|
||||
Binding<MessageChannel> 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<KafkaProducerProperties> props =
|
||||
new ExtendedProducerProperties<KafkaProducerProperties>(ext);
|
||||
|
||||
Binding<MessageChannel> bindProducer = binder.bindProducer("testP", outbound, props);
|
||||
AtomicReference<Mono<RecordMetadata>> 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();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
<configuration>
|
||||
<appender name="stdout" class="ch.qos.logback.core.ConsoleAppender">
|
||||
<encoder>
|
||||
<pattern>%d{ISO8601} %5p %t %c{2}:%L - %m%n</pattern>
|
||||
</encoder>
|
||||
</appender>
|
||||
<logger name="org.apache.kafka" level="WARN"/>
|
||||
<logger name="reactor.kafka" level="DEBUG"/>
|
||||
<logger name="org.springframework.integration.kafka" level="INFO"/>
|
||||
<logger name="org.springframework.kafka" level="INFO"/>
|
||||
<logger name="org.springframework.cloud.stream" level="INFO" />
|
||||
<logger name="org.springframework.integration.channel" level="INFO" />
|
||||
<logger name="kafka.server.ReplicaFetcherThread" level="ERROR"/>
|
||||
<logger name="kafka.server.LogDirFailureChannel" level="FATAL"/>
|
||||
<logger name="kafka.server.BrokerMetadataCheckpoint" level="ERROR"/>
|
||||
<logger name="kafka.utils.CoreUtils$" level="ERROR"/>
|
||||
<root level="WARN">
|
||||
<appender-ref ref="stdout"/>
|
||||
</root>
|
||||
</configuration>
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user