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

@@ -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.<binding-name>.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.

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;
 	}
 

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<Object, Object> outboundBindTarget, Serde<Object> keySerde,
 					Serde<Object> valueSerde, KafkaStreamsProducerProperties properties) {
 		final Produced<Object, Object> 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(),

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;
+	}
 }

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;
+	}
 }

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<String, Object> streamConfigGlobalProperties = (Map<String, Object>) context.getBean("streamConfigGlobalProperties");