GH-265: Add Kafka Publisher and Sink (#469)
* GH-265: Add Kafka Consumer and Sink Fixes https://github.com/spring-cloud/stream-applications/issues/265 * Make `kafkaPublisher` based on the Spring for Apache Kafka auto-configuration and Spring Integration channel adapter, essentially `KafkaProducerMessageHandler` * Make this `KafkaPublisherConfiguration` as an auto-configuration by itself. * Expose those simple properties required by the `KafkaProducerMessageHandlerSpec` * Add `kafka-sink` module based on the `kafkaPublisher` * Add `kafka-sink` into apps metadata properties
This commit is contained in:
Artem Bilan
31
consumer/kafka-publisher/README.adoc
Normal file
31
consumer/kafka-publisher/README.adoc
Normal file
@@ -0,0 +1,31 @@
|
||||
# Apache Kafka Publisher (Consumer function)
|
||||
|
||||
A `Consumer<Message<?>>` that allows to publish messages to Apache Kafka topic.
|
||||
|
||||
|
||||
## Beans for injection
|
||||
|
||||
The `KafkaPublisherConfiguration` is an auto-configuration, so no need to import anything.
|
||||
|
||||
The `Consumer<Message<?>> kafkaPublisher` bean can be injection into target service for producing data into Kafka topic.
|
||||
|
||||
## Configuration Options
|
||||
|
||||
All configuration properties are prefixed with `kafka.publisher`.
|
||||
|
||||
For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/consumer/kafka/KafkaPublisherProperties.java[KafkaPublisherProperties].
|
||||
Also, this artifact fully depends on Spring for Apache Kafka auto-configuration and injects a `KafkaTemplate` from there.
|
||||
|
||||
A `ComponentCustomizer<KafkaProducerMessageHandlerSpec<?, ?, ?>>` bean can be added in the target project to provide any custom options for the `KafkaProducerMessageHandlerSpec` configuration used by the `kafkaPublisher`.
|
||||
|
||||
The `KafkaPublisherConfiguration` also exposes 3 `PublishSubscribeChannel`: `kafkaPublisherSuccessChannel`, `kafkaPublisherFailureChannel`, `kafkaPublisherFuturesChannel`.
|
||||
They are mapped to respective options of the `KafkaProducerMessageHandler`.
|
||||
They may be subscribed in the target project any possible Spring Integration way.
|
||||
See more information about `KafkaProducerMessageHandler` configuration and behavior in Spring Integration https://docs.spring.io/spring-integration/docs/current/reference/html/kafka.html#kafka-outbound[documentation].
|
||||
|
||||
## Tests
|
||||
|
||||
|
||||
## Other usage
|
||||
|
||||
See this https://github.com/spring-cloud/stream-applications/blob/master/applications/sink/kafka-sink/README.adoc[README] where this consumer is used to create a Spring Cloud Stream application where it makes an Apache Kafka sink.
|
||||
34
consumer/kafka-publisher/pom.xml
Normal file
34
consumer/kafka-publisher/pom.xml
Normal file
@@ -0,0 +1,34 @@
|
||||
<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 http://maven.apache.org/maven-v4_0_0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<parent>
|
||||
<artifactId>spring-functions-parent</artifactId>
|
||||
<groupId>org.springframework.cloud.fn</groupId>
|
||||
<version>4.0.0-SNAPSHOT</version>
|
||||
<relativePath>../../spring-functions-parent/pom.xml</relativePath>
|
||||
</parent>
|
||||
|
||||
<artifactId>kafka-publisher</artifactId>
|
||||
<name>kafka-publisher</name>
|
||||
<description>Apache Kafka Publisher(Consumer Function)</description>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.integration</groupId>
|
||||
<artifactId>spring-integration-kafka</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework.kafka</groupId>
|
||||
<artifactId>spring-kafka-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.integration</groupId>
|
||||
<artifactId>spring-integration-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
</project>
|
||||
@@ -0,0 +1,121 @@
|
||||
/*
|
||||
* Copyright 2023-2023 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.cloud.fn.consumer.kafka;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.function.Consumer;
|
||||
|
||||
import org.springframework.boot.autoconfigure.AutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.boot.context.properties.PropertyMapper;
|
||||
import org.springframework.cloud.fn.common.config.ComponentCustomizer;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.integration.channel.PublishSubscribeChannel;
|
||||
import org.springframework.integration.expression.ValueExpression;
|
||||
import org.springframework.integration.kafka.dsl.Kafka;
|
||||
import org.springframework.integration.kafka.dsl.KafkaProducerMessageHandlerSpec;
|
||||
import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler;
|
||||
import org.springframework.kafka.core.KafkaTemplate;
|
||||
import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
|
||||
import org.springframework.lang.Nullable;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
|
||||
/**
|
||||
* A configuration for Apache Kafka Publisher (Consumer function).
|
||||
* Uses a {@link KafkaProducerMessageHandlerSpec} to publish a message to Kafka topic.
|
||||
*
|
||||
* @author Artem Bilan
|
||||
*
|
||||
* @since 4.0
|
||||
*/
|
||||
@AutoConfiguration(after = KafkaAutoConfiguration.class)
|
||||
@EnableConfigurationProperties(KafkaPublisherProperties.class)
|
||||
public class KafkaPublisherConfiguration {
|
||||
|
||||
/**
|
||||
* The function to produce messages to the Kafka topic.
|
||||
* @param kafkaProducerMessageHandler the handler to publish messages to Kafka.
|
||||
* @return the consumer for accepting message for producing to Kafka.
|
||||
*/
|
||||
@Bean
|
||||
public Consumer<Message<?>> kafkaPublisher(KafkaProducerMessageHandler<?, ?> kafkaProducerMessageHandler) {
|
||||
return kafkaProducerMessageHandler::handleMessage;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public KafkaProducerMessageHandler<?, ?> kafkaProducerMessageHandlerSpec(KafkaTemplate<?, ?> kafkaTemplate,
|
||||
KafkaPublisherProperties kafkaPublisherProperties,
|
||||
PublishSubscribeChannel kafkaPublisherSuccessChannel,
|
||||
PublishSubscribeChannel kafkaPublisherFailureChannel,
|
||||
PublishSubscribeChannel kafkaPublisherFuturesChannel,
|
||||
@Nullable ComponentCustomizer<KafkaProducerMessageHandlerSpec<?, ?, ?>> kafkaProducerSpecComponentCustomizer) {
|
||||
|
||||
var kafkaProducerMessageHandlerSpec = Kafka.outboundChannelAdapter(kafkaTemplate);
|
||||
|
||||
PropertyMapper mapper = PropertyMapper.get().alwaysApplyingWhenNonNull();
|
||||
|
||||
mapper.from(kafkaPublisherProperties.getTopic()).to(kafkaProducerMessageHandlerSpec::topic);
|
||||
mapper.from(kafkaPublisherProperties.getTopicExpression()).to(kafkaProducerMessageHandlerSpec::topicExpression);
|
||||
mapper.from(kafkaPublisherProperties.getKey()).to(kafkaProducerMessageHandlerSpec::messageKey);
|
||||
mapper.from(kafkaPublisherProperties.getKeyExpression()).to(kafkaProducerMessageHandlerSpec::messageKeyExpression);
|
||||
mapper.from(kafkaPublisherProperties.getPartition()).to(kafkaProducerMessageHandlerSpec::partitionId);
|
||||
mapper.from(kafkaPublisherProperties.getPartitionExpression()).to(kafkaProducerMessageHandlerSpec::partitionIdExpression);
|
||||
mapper.from(kafkaPublisherProperties.getTimestamp()).as(ValueExpression::new).to(kafkaProducerMessageHandlerSpec::timestampExpression);
|
||||
mapper.from(kafkaPublisherProperties.getTimestampExpression()).to(kafkaProducerMessageHandlerSpec::timestampExpression);
|
||||
mapper.from(kafkaPublisherProperties.getSendTimeout()).as(Duration::toMillis).to(kafkaProducerMessageHandlerSpec::sendTimeout);
|
||||
mapper.from(kafkaPublisherProperties.isUseTemplateConverter()).to(kafkaProducerMessageHandlerSpec::useTemplateConverter);
|
||||
|
||||
kafkaProducerMessageHandlerSpec.headerMapper(new DefaultKafkaHeaderMapper(kafkaPublisherProperties.getMappedHeaders()));
|
||||
|
||||
kafkaProducerMessageHandlerSpec.sendSuccessChannel(kafkaPublisherSuccessChannel);
|
||||
kafkaProducerMessageHandlerSpec.sendFailureChannel(kafkaPublisherFailureChannel);
|
||||
kafkaProducerMessageHandlerSpec.futuresChannel(kafkaPublisherFuturesChannel);
|
||||
|
||||
if (kafkaProducerSpecComponentCustomizer != null) {
|
||||
kafkaProducerSpecComponentCustomizer.customize(kafkaProducerMessageHandlerSpec);
|
||||
}
|
||||
|
||||
return kafkaProducerMessageHandlerSpec.get();
|
||||
}
|
||||
|
||||
/**
|
||||
* @see KafkaProducerMessageHandler#setSendSuccessChannel(MessageChannel)
|
||||
*/
|
||||
@Bean
|
||||
public PublishSubscribeChannel kafkaPublisherSuccessChannel() {
|
||||
return new PublishSubscribeChannel();
|
||||
}
|
||||
|
||||
/**
|
||||
* @see KafkaProducerMessageHandler#setSendFailureChannel(MessageChannel)
|
||||
*/
|
||||
@Bean
|
||||
public PublishSubscribeChannel kafkaPublisherFailureChannel() {
|
||||
return new PublishSubscribeChannel();
|
||||
}
|
||||
|
||||
/**
|
||||
* @see KafkaProducerMessageHandler#setFuturesChannel(MessageChannel)
|
||||
*/
|
||||
@Bean
|
||||
public PublishSubscribeChannel kafkaPublisherFuturesChannel() {
|
||||
return new PublishSubscribeChannel();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,190 @@
|
||||
/*
|
||||
* Copyright 2023-2023 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.cloud.fn.consumer.kafka;
|
||||
|
||||
import java.time.Duration;
|
||||
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
import org.springframework.expression.Expression;
|
||||
|
||||
/**
|
||||
* Properties for the Kafka Publisher (Consumer function).
|
||||
*
|
||||
* @author Artem Bilan
|
||||
*
|
||||
* @since 4.0
|
||||
*/
|
||||
@ConfigurationProperties("kafka.publisher")
|
||||
public class KafkaPublisherProperties {
|
||||
|
||||
/**
|
||||
* Kafka topic - overridden by topicExpression, if supplied. Defaults to KafkaTemplate.getDefaultTopic()
|
||||
*/
|
||||
private String topic;
|
||||
|
||||
/**
|
||||
* A SpEL expression that evaluates to a Kafka topic.
|
||||
*/
|
||||
private Expression topicExpression;
|
||||
|
||||
/**
|
||||
* Kafka record key - overridden by keyExpression, if supplied.
|
||||
*/
|
||||
private String key;
|
||||
|
||||
/**
|
||||
* A SpEL expression that evaluates to a Kafka record key.
|
||||
*/
|
||||
private Expression keyExpression;
|
||||
|
||||
/**
|
||||
* Kafka topic partition - overridden by partitionExpression, if supplied.
|
||||
*/
|
||||
private Integer partition;
|
||||
|
||||
/**
|
||||
* A SpEL expression that evaluates to a Kafka topic partition.
|
||||
*/
|
||||
private Expression partitionExpression;
|
||||
|
||||
/**
|
||||
* Kafka record timestamp - overridden by timestampExpression, if supplied.
|
||||
*/
|
||||
private Long timestamp;
|
||||
|
||||
/**
|
||||
* A SpEL expression that evaluates to a Kafka record timestamp.
|
||||
*/
|
||||
private Expression timestampExpression;
|
||||
|
||||
/**
|
||||
* True if Kafka producer handler should operation in a sync mode.
|
||||
*/
|
||||
private boolean sync;
|
||||
|
||||
/**
|
||||
* How long Kafka producer handler should wait for send operation results. Defaults to 10 seconds.
|
||||
*/
|
||||
private Duration sendTimeout = Duration.ofSeconds(10);
|
||||
|
||||
/**
|
||||
* Headers that will be mapped.
|
||||
*/
|
||||
private String[] mappedHeaders = { "*" };
|
||||
|
||||
/**
|
||||
* Whether to use the template's message converter to create a Kafka record.
|
||||
*/
|
||||
private boolean useTemplateConverter;
|
||||
|
||||
public String getTopic() {
|
||||
return this.topic;
|
||||
}
|
||||
|
||||
public void setTopic(String topic) {
|
||||
this.topic = topic;
|
||||
}
|
||||
|
||||
public Expression getTopicExpression() {
|
||||
return this.topicExpression;
|
||||
}
|
||||
|
||||
public void setTopicExpression(Expression topicExpression) {
|
||||
this.topicExpression = topicExpression;
|
||||
}
|
||||
|
||||
public String getKey() {
|
||||
return this.key;
|
||||
}
|
||||
|
||||
public void setKey(String key) {
|
||||
this.key = key;
|
||||
}
|
||||
|
||||
public Expression getKeyExpression() {
|
||||
return this.keyExpression;
|
||||
}
|
||||
|
||||
public void setKeyExpression(Expression keyExpression) {
|
||||
this.keyExpression = keyExpression;
|
||||
}
|
||||
|
||||
public Integer getPartition() {
|
||||
return this.partition;
|
||||
}
|
||||
|
||||
public void setPartition(Integer partition) {
|
||||
this.partition = partition;
|
||||
}
|
||||
|
||||
public Expression getPartitionExpression() {
|
||||
return this.partitionExpression;
|
||||
}
|
||||
|
||||
public void setPartitionExpression(Expression partitionExpression) {
|
||||
this.partitionExpression = partitionExpression;
|
||||
}
|
||||
|
||||
public Long getTimestamp() {
|
||||
return this.timestamp;
|
||||
}
|
||||
|
||||
public void setTimestamp(Long timestamp) {
|
||||
this.timestamp = timestamp;
|
||||
}
|
||||
|
||||
public Expression getTimestampExpression() {
|
||||
return this.timestampExpression;
|
||||
}
|
||||
|
||||
public void setTimestampExpression(Expression timestampExpression) {
|
||||
this.timestampExpression = timestampExpression;
|
||||
}
|
||||
|
||||
public boolean isSync() {
|
||||
return this.sync;
|
||||
}
|
||||
|
||||
public void setSync(boolean sync) {
|
||||
this.sync = sync;
|
||||
}
|
||||
|
||||
public Duration getSendTimeout() {
|
||||
return this.sendTimeout;
|
||||
}
|
||||
|
||||
public void setSendTimeout(Duration sendTimeout) {
|
||||
this.sendTimeout = sendTimeout;
|
||||
}
|
||||
|
||||
public String[] getMappedHeaders() {
|
||||
return this.mappedHeaders;
|
||||
}
|
||||
|
||||
public void setMappedHeaders(String[] mappedHeaders) {
|
||||
this.mappedHeaders = mappedHeaders;
|
||||
}
|
||||
|
||||
public boolean isUseTemplateConverter() {
|
||||
return this.useTemplateConverter;
|
||||
}
|
||||
|
||||
public void setUseTemplateConverter(boolean useTemplateConverter) {
|
||||
this.useTemplateConverter = useTemplateConverter;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
org.springframework.cloud.fn.consumer.kafka.KafkaPublisherConfiguration
|
||||
@@ -0,0 +1,168 @@
|
||||
/*
|
||||
* Copyright 2023-2023 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.cloud.fn.consumer.kafka;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.Arrays;
|
||||
import java.util.Map;
|
||||
import java.util.function.Consumer;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import org.apache.kafka.clients.consumer.ConsumerRecord;
|
||||
import org.apache.kafka.clients.producer.RecordMetadata;
|
||||
import org.apache.kafka.common.TopicPartition;
|
||||
import org.apache.kafka.common.header.Header;
|
||||
import org.junit.jupiter.api.BeforeAll;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import reactor.core.publisher.Sinks;
|
||||
|
||||
import org.springframework.boot.autoconfigure.AutoConfigurations;
|
||||
import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
|
||||
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
|
||||
import org.springframework.cloud.fn.common.config.SpelExpressionConverterConfiguration;
|
||||
import org.springframework.context.ApplicationContext;
|
||||
import org.springframework.integration.channel.PublishSubscribeChannel;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.kafka.KafkaException;
|
||||
import org.springframework.kafka.core.ConsumerFactory;
|
||||
import org.springframework.kafka.core.KafkaTemplate;
|
||||
import org.springframework.kafka.support.KafkaHeaders;
|
||||
import org.springframework.kafka.test.EmbeddedKafkaBroker;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageHandlingException;
|
||||
import org.springframework.messaging.MessageHeaders;
|
||||
import org.springframework.messaging.support.GenericMessage;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
|
||||
|
||||
/**
|
||||
* @author Artem Bilan
|
||||
*
|
||||
* @since 4.0
|
||||
*/
|
||||
public class KafkaPublisherConfigurationTests {
|
||||
|
||||
static final EmbeddedKafkaBroker EMBEDDED_KAFKA =
|
||||
new EmbeddedKafkaBroker(1, true, 1)
|
||||
.brokerListProperty("spring.kafka.bootstrap-servers");
|
||||
|
||||
final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
|
||||
.withConfiguration(AutoConfigurations.of(
|
||||
KafkaAutoConfiguration.class,
|
||||
KafkaPublisherConfiguration.class,
|
||||
SpelExpressionConverterConfiguration.class));
|
||||
|
||||
@BeforeAll
|
||||
static void initializeEmbeddedKafka() {
|
||||
EMBEDDED_KAFKA.afterPropertiesSet();
|
||||
}
|
||||
|
||||
@Test
|
||||
void defaultTopicReceivesTheRecord() {
|
||||
String defaultTopic = "DEFAULT_TOPIC";
|
||||
this.contextRunner.withPropertyValues("spring.kafka.template.defaultTopic=" + defaultTopic)
|
||||
.run((context) -> {
|
||||
KafkaTemplate<?, ?> kafkaTemplate = obtainKafkaTemplate(context);
|
||||
Consumer<Message<?>> kafkaPublisher = getKafkaPublisher(context);
|
||||
String testData = "test data";
|
||||
kafkaPublisher.accept(new GenericMessage<>(testData));
|
||||
ConsumerRecord<?, ?> receive = kafkaTemplate.receive(defaultTopic, 0, 0, Duration.ofSeconds(10));
|
||||
assertThat(receive).extracting(ConsumerRecord::value).isEqualTo(testData);
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void wrongPartitionViaProperties() {
|
||||
this.contextRunner.withPropertyValues(
|
||||
"spring.kafka.producer.properties[max.block.ms]=1000",
|
||||
"kafka.publisher.topic=topic1",
|
||||
"kafka.publisher.partition=1", // Our broker allows only one partition for auto-created topic
|
||||
"kafka.publisher.sync=true")
|
||||
.run((context) -> {
|
||||
Consumer<Message<?>> kafkaConsumer = getKafkaPublisher(context);
|
||||
assertThatExceptionOfType(MessageHandlingException.class)
|
||||
.isThrownBy(() -> kafkaConsumer.accept(new GenericMessage<>("test data")))
|
||||
.withCauseInstanceOf(KafkaException.class)
|
||||
.withStackTraceContaining("Topic topic1 not present in metadata after 1000 ms.");
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void successChannelInteractionAndMappedHeaders() {
|
||||
this.contextRunner.withPropertyValues("kafka.publisher.topicExpression=headers.topic",
|
||||
"kafka.publisher.mappedHeaders=mapped")
|
||||
.run((context) -> {
|
||||
KafkaTemplate<?, ?> kafkaTemplate = obtainKafkaTemplate(context);
|
||||
Consumer<Message<?>> kafkaConsumer = getKafkaPublisher(context);
|
||||
|
||||
PublishSubscribeChannel kafkaConsumerSuccessChannel =
|
||||
context.getBean("kafkaPublisherSuccessChannel", PublishSubscribeChannel.class);
|
||||
|
||||
Sinks.One<Message<?>> successSend = Sinks.one();
|
||||
|
||||
kafkaConsumerSuccessChannel.subscribe(successSend::tryEmitValue);
|
||||
|
||||
String testTopic = "topic2";
|
||||
String testData = "some other data";
|
||||
Message<String> testMessage =
|
||||
MessageBuilder.withPayload(testData)
|
||||
.setHeader("topic", testTopic)
|
||||
.setHeader("mapped", "mapped value")
|
||||
.setHeader("not mapped", "not mapped")
|
||||
.build();
|
||||
|
||||
kafkaConsumer.accept(testMessage);
|
||||
|
||||
ConsumerRecord<?, ?> receive = kafkaTemplate.receive(testTopic, 0, 0, Duration.ofSeconds(10));
|
||||
assertThat(receive).extracting(ConsumerRecord::value).isEqualTo(testData);
|
||||
Map<String, String> headers =
|
||||
Arrays.stream(receive.headers().toArray())
|
||||
.collect(Collectors.toMap(Header::key, (header) -> new String(header.value())));
|
||||
assertThat(headers)
|
||||
.containsEntry("mapped", "mapped value")
|
||||
.doesNotContainKeys("topic", "not mapped");
|
||||
|
||||
Message<?> successMessage = successSend.asMono().block(Duration.ofSeconds(10));
|
||||
|
||||
assertThat(successMessage)
|
||||
.satisfies(message -> {
|
||||
assertThat(message.getPayload()).isEqualTo(testData);
|
||||
MessageHeaders messageHeaders = message.getHeaders();
|
||||
assertThat(messageHeaders)
|
||||
.containsKeys("topic", "mapped", "not mapped", KafkaHeaders.RECORD_METADATA);
|
||||
assertThat(messageHeaders.get(KafkaHeaders.RECORD_METADATA))
|
||||
.isInstanceOf(RecordMetadata.class)
|
||||
.extracting("topicPartition")
|
||||
.isEqualTo(new TopicPartition(testTopic, 0));
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private static KafkaTemplate<?, ?> obtainKafkaTemplate(ApplicationContext applicationContext) {
|
||||
KafkaTemplate<?, ?> kafkaTemplate = applicationContext.getBean(KafkaTemplate.class);
|
||||
kafkaTemplate.setConsumerFactory(applicationContext.getBean(ConsumerFactory.class));
|
||||
return kafkaTemplate;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private static Consumer<Message<?>> getKafkaPublisher(ApplicationContext applicationContext) {
|
||||
return (Consumer<Message<?>>) applicationContext.getBean("kafkaPublisher");
|
||||
}
|
||||
|
||||
}
|
||||
@@ -18,6 +18,7 @@
|
||||
<module>file-consumer</module>
|
||||
<module>ftp-consumer</module>
|
||||
<module>jdbc-consumer</module>
|
||||
<module>kafka-publisher</module>
|
||||
<module>log-consumer</module>
|
||||
<module>mongodb-consumer</module>
|
||||
<module>mqtt-consumer</module>
|
||||
|
||||
@@ -147,6 +147,11 @@
|
||||
<artifactId>jdbc-consumer</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud.fn</groupId>
|
||||
<artifactId>kafka-publisher</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud.fn</groupId>
|
||||
<artifactId>log-consumer</artifactId>
|
||||
|
||||
Reference in New Issue
Block a user