From 7cae3aa54ff0292b7aec92a2f23e622a9424b0a7 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Thu, 1 Jul 2021 19:35:20 -0400 Subject: [PATCH] GH-1096: Named components in Kafka Streams Support KIP-307 in Kafka Streams binder where the input (source) and output (sink) bindings are customized with user-provided names. Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1096 Resolves #1098 --- docs/src/main/asciidoc/kafka-streams.adoc | 12 +++++++++++- .../AbstractKafkaStreamsBinderProcessor.java | 3 +++ .../stream/binder/kafka/streams/KStreamBinder.java | 3 +++ .../properties/KafkaStreamsConsumerProperties.java | 13 +++++++++++++ .../properties/KafkaStreamsProducerProperties.java | 13 +++++++++++++ .../KafkaStreamsBinderWordCountFunctionTests.java | 6 +++++- 6 files changed, 48 insertions(+), 2 deletions(-) diff --git a/docs/src/main/asciidoc/kafka-streams.adoc b/docs/src/main/asciidoc/kafka-streams.adoc index 74e7b0d20..bc221f173 100644 --- a/docs/src/main/asciidoc/kafka-streams.adoc +++ b/docs/src/main/asciidoc/kafka-streams.adoc @@ -1953,12 +1953,17 @@ flag to enable/disable native encoding + Default: `true`. -streamPartitionerBeanName: +streamPartitionerBeanName:: Custom outbound partitioner bean name to be used at the consumer. Applications can provide custom `StreamPartitioner` as a Spring bean and the name of this bean can be provided to the producer to use instead of the default one. + Default: See the discussion above on outbound partition support. +producedAs:: +Custom name for the sink component to which the processor is producing to. ++ +Deafult: `none` (generated by Kafka Streams) + ==== Kafka Streams Consumer Properties The following properties are available for Kafka Streams consumers and must be prefixed with `spring.cloud.stream.kafka.streams.bindings..consumer.` @@ -2028,6 +2033,11 @@ Event type header key on each incoming records through this binding. + Default: `event_type` +consumedAs:: +Custom name for the source component from which the processor is consuming from. ++ +Deafult: `none` (generated by Kafka Streams) + ==== Special note on concurrency In Kafka Streams, you can control of the number of threads a processor can create using the `num.stream.threads` property. diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java index e27915751..adab0f42c 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java @@ -571,6 +571,9 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application if (timestampExtractor != null) { consumed.withTimestampExtractor(timestampExtractor); } + if (StringUtils.hasText(kafkaStreamsConsumerProperties.getConsumedAs())) { + consumed.withName(kafkaStreamsConsumerProperties.getConsumedAs()); + } return consumed; } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java index a60f33126..3ae3354cb 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java @@ -191,6 +191,9 @@ class KStreamBinder extends KStream outboundBindTarget, Serde keySerde, Serde valueSerde, KafkaStreamsProducerProperties properties) { final Produced produced = Produced.with(keySerde, valueSerde); + if (StringUtils.hasText(properties.getProducedAs())) { + produced.withName(properties.getProducedAs()); + } StreamPartitioner streamPartitioner = null; if (!StringUtils.isEmpty(properties.getStreamPartitionerBeanName())) { streamPartitioner = getApplicationContext().getBean(properties.getStreamPartitionerBeanName(), diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsConsumerProperties.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsConsumerProperties.java index a9a39b6cb..17b10a962 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsConsumerProperties.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsConsumerProperties.java @@ -66,6 +66,11 @@ public class KafkaStreamsConsumerProperties extends KafkaConsumerProperties { */ private String eventTypeHeaderKey = "event_type"; + /** + * Custom name for the source component from which the processor is consuming from. + */ + private String consumedAs; + public String getApplicationId() { return this.applicationId; } @@ -129,4 +134,12 @@ public class KafkaStreamsConsumerProperties extends KafkaConsumerProperties { public void setEventTypeHeaderKey(String eventTypeHeaderKey) { this.eventTypeHeaderKey = eventTypeHeaderKey; } + + public String getConsumedAs() { + return consumedAs; + } + + public void setConsumedAs(String consumedAs) { + this.consumedAs = consumedAs; + } } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsProducerProperties.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsProducerProperties.java index e5df57aef..152621c5e 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsProducerProperties.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsProducerProperties.java @@ -41,6 +41,11 @@ public class KafkaStreamsProducerProperties extends KafkaProducerProperties { */ private String streamPartitionerBeanName; + /** + * Custom name for the sink component to which the processor is producing to. + */ + private String producedAs; + public String getKeySerde() { return this.keySerde; } @@ -64,4 +69,12 @@ public class KafkaStreamsProducerProperties extends KafkaProducerProperties { public void setStreamPartitionerBeanName(String streamPartitionerBeanName) { this.streamPartitionerBeanName = streamPartitionerBeanName; } + + public String getProducedAs() { + return producedAs; + } + + public void setProducedAs(String producedAs) { + this.producedAs = producedAs; + } } diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java index 327b6b715..ae81fc128 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java @@ -108,7 +108,7 @@ public class KafkaStreamsBinderWordCountFunctionTests { "--spring.jmx.enabled=false", "--spring.cloud.stream.bindings.process-in-0.destination=words", "--spring.cloud.stream.bindings.process-out-0.destination=counts", - "--spring.cloud.stream.kafka.streams.default.consumer.application-id=testKstreamWordCountFunction", + "--spring.cloud.stream.kafka.streams.binder.application-id=testKstreamWordCountFunction", "--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000", "--spring.cloud.stream.kafka.streams.binder.consumerProperties.request.timeout.ms=29000", //for testing ...binder.consumerProperties "--spring.cloud.stream.kafka.streams.binder.producerProperties.max.block.ms=90000", //for testing ...binder.producerProperties @@ -116,6 +116,8 @@ public class KafkaStreamsBinderWordCountFunctionTests { "=org.apache.kafka.common.serialization.Serdes$StringSerde", "--spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde" + "=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.consumedAs=custom-consumer", + "--spring.cloud.stream.kafka.streams.bindings.process-out-0.producer.producedAs=custom-producer", "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) { receiveAndValidate("words", "counts"); @@ -136,6 +138,8 @@ public class KafkaStreamsBinderWordCountFunctionTests { final String topology2 = kafkaStreamsTopologyEndpoint.kafkaStreamsTopology("testKstreamWordCountFunction"); assertThat(topology1).isNotEmpty(); assertThat(topology1).isEqualTo(topology2); + assertThat(topology1.contains("Source: custom-consumer")).isTrue(); + assertThat(topology1.contains("Sink: custom-producer")).isTrue(); //verify that ...binder.consumerProperties and ...binder.producerProperties work. Map streamConfigGlobalProperties = (Map) context.getBean("streamConfigGlobalProperties");