From d30252a66bef12cf1578606c08bed29d3b386799 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 7 Jun 2023 22:01:45 -0400 Subject: [PATCH] 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 --- consumer/kafka-publisher/README.adoc | 31 +++ consumer/kafka-publisher/pom.xml | 34 ++++ .../kafka/KafkaPublisherConfiguration.java | 121 +++++++++++ .../kafka/KafkaPublisherProperties.java | 190 ++++++++++++++++++ ...ot.autoconfigure.AutoConfiguration.imports | 1 + .../KafkaPublisherConfigurationTests.java | 168 ++++++++++++++++ consumer/pom.xml | 1 + function-dependencies/pom.xml | 5 + 8 files changed, 551 insertions(+) create mode 100644 consumer/kafka-publisher/README.adoc create mode 100644 consumer/kafka-publisher/pom.xml create mode 100644 consumer/kafka-publisher/src/main/java/org/springframework/cloud/fn/consumer/kafka/KafkaPublisherConfiguration.java create mode 100644 consumer/kafka-publisher/src/main/java/org/springframework/cloud/fn/consumer/kafka/KafkaPublisherProperties.java create mode 100644 consumer/kafka-publisher/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports create mode 100644 consumer/kafka-publisher/src/test/java/org/springframework/cloud/fn/consumer/kafka/KafkaPublisherConfigurationTests.java diff --git a/consumer/kafka-publisher/README.adoc b/consumer/kafka-publisher/README.adoc new file mode 100644 index 00000000..4a70f3b3 --- /dev/null +++ b/consumer/kafka-publisher/README.adoc @@ -0,0 +1,31 @@ +# Apache Kafka Publisher (Consumer function) + +A `Consumer>` 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> 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>` 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. diff --git a/consumer/kafka-publisher/pom.xml b/consumer/kafka-publisher/pom.xml new file mode 100644 index 00000000..33bd815b --- /dev/null +++ b/consumer/kafka-publisher/pom.xml @@ -0,0 +1,34 @@ + + 4.0.0 + + + spring-functions-parent + org.springframework.cloud.fn + 4.0.0-SNAPSHOT + ../../spring-functions-parent/pom.xml + + + kafka-publisher + kafka-publisher + Apache Kafka Publisher(Consumer Function) + + + + org.springframework.integration + spring-integration-kafka + + + + org.springframework.kafka + spring-kafka-test + test + + + org.springframework.integration + spring-integration-test + test + + + + diff --git a/consumer/kafka-publisher/src/main/java/org/springframework/cloud/fn/consumer/kafka/KafkaPublisherConfiguration.java b/consumer/kafka-publisher/src/main/java/org/springframework/cloud/fn/consumer/kafka/KafkaPublisherConfiguration.java new file mode 100644 index 00000000..6138f10b --- /dev/null +++ b/consumer/kafka-publisher/src/main/java/org/springframework/cloud/fn/consumer/kafka/KafkaPublisherConfiguration.java @@ -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> kafkaPublisher(KafkaProducerMessageHandler kafkaProducerMessageHandler) { + return kafkaProducerMessageHandler::handleMessage; + } + + @Bean + public KafkaProducerMessageHandler kafkaProducerMessageHandlerSpec(KafkaTemplate kafkaTemplate, + KafkaPublisherProperties kafkaPublisherProperties, + PublishSubscribeChannel kafkaPublisherSuccessChannel, + PublishSubscribeChannel kafkaPublisherFailureChannel, + PublishSubscribeChannel kafkaPublisherFuturesChannel, + @Nullable ComponentCustomizer> 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(); + } + +} diff --git a/consumer/kafka-publisher/src/main/java/org/springframework/cloud/fn/consumer/kafka/KafkaPublisherProperties.java b/consumer/kafka-publisher/src/main/java/org/springframework/cloud/fn/consumer/kafka/KafkaPublisherProperties.java new file mode 100644 index 00000000..4994cab6 --- /dev/null +++ b/consumer/kafka-publisher/src/main/java/org/springframework/cloud/fn/consumer/kafka/KafkaPublisherProperties.java @@ -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; + } + +} diff --git a/consumer/kafka-publisher/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/consumer/kafka-publisher/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports new file mode 100644 index 00000000..9be4c43d --- /dev/null +++ b/consumer/kafka-publisher/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports @@ -0,0 +1 @@ +org.springframework.cloud.fn.consumer.kafka.KafkaPublisherConfiguration diff --git a/consumer/kafka-publisher/src/test/java/org/springframework/cloud/fn/consumer/kafka/KafkaPublisherConfigurationTests.java b/consumer/kafka-publisher/src/test/java/org/springframework/cloud/fn/consumer/kafka/KafkaPublisherConfigurationTests.java new file mode 100644 index 00000000..124af477 --- /dev/null +++ b/consumer/kafka-publisher/src/test/java/org/springframework/cloud/fn/consumer/kafka/KafkaPublisherConfigurationTests.java @@ -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> 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> 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> kafkaConsumer = getKafkaPublisher(context); + + PublishSubscribeChannel kafkaConsumerSuccessChannel = + context.getBean("kafkaPublisherSuccessChannel", PublishSubscribeChannel.class); + + Sinks.One> successSend = Sinks.one(); + + kafkaConsumerSuccessChannel.subscribe(successSend::tryEmitValue); + + String testTopic = "topic2"; + String testData = "some other data"; + Message 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 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> getKafkaPublisher(ApplicationContext applicationContext) { + return (Consumer>) applicationContext.getBean("kafkaPublisher"); + } + +} diff --git a/consumer/pom.xml b/consumer/pom.xml index 4e1aa926..377226f7 100644 --- a/consumer/pom.xml +++ b/consumer/pom.xml @@ -18,6 +18,7 @@ file-consumer ftp-consumer jdbc-consumer + kafka-publisher log-consumer mongodb-consumer mqtt-consumer diff --git a/function-dependencies/pom.xml b/function-dependencies/pom.xml index 1fa623cd..ba462d47 100644 --- a/function-dependencies/pom.xml +++ b/function-dependencies/pom.xml @@ -147,6 +147,11 @@ jdbc-consumer ${project.version} + + org.springframework.cloud.fn + kafka-publisher + ${project.version} + org.springframework.cloud.fn log-consumer