diff --git a/docs/src/main/asciidoc/kafka-streams.adoc b/docs/src/main/asciidoc/kafka-streams.adoc index bc221f173..1eb8da96b 100644 --- a/docs/src/main/asciidoc/kafka-streams.adoc +++ b/docs/src/main/asciidoc/kafka-streams.adoc @@ -252,7 +252,28 @@ The input from the three partial functions which are `KStream`, `GlobalKTable`, Input bindings are named as `enrichOrder-in-0`, `enrichOrder-in-1` and `enrichOrder-in-2` respectively. Output binding is named as `enrichOrder-out-0`. With curried functions, you can virtually have any number of inputs. However, keep in mind that, anything more than a smaller number of inputs and partially applied functions for them as above in Java might lead to unreadable code. -Therefore if your Kafka Streams application requires more than a reasonably smaller number of input bindings and you want to use this functional model, then you may want to rethink your design and decompose the application appropriately. +Therefore if your Kafka Streams application requires more than a reasonably smaller number of input bindings, and you want to use this functional model, then you may want to rethink your design and decompose the application appropriately. + +===== Output Bindings + +Kafka Streams binder allows types of either `KStream` or `KTable` as output bindings. +Behind the scenes, the binder uses the `to` method on `KStream` to send the resultant records to the output topic. +If the application provides a `KTable` as output in the function, the binder still uses this technique by delegating to the `to` method of `KStream`. + +For example both functions below will work: + +``` +@Bean +public Function, KTable> foo() { + return KStream::toTable; + }; +} + +@Bean +public Function, KStream> bar() { + return KTable::toStream; +} +``` ===== Multiple Output Bindings @@ -383,8 +404,7 @@ The default output binding for this example becomes `curriedFoobar-out-0`. ====== Special note on using `KTable` as output in function composition -When using function composition, for intermediate functions, you can use `KTable` as output. -For instance, lets say you have the following two functions. +Lets say you have the following two functions. ``` @Bean @@ -399,10 +419,7 @@ public Function, KStream> bar() { } ``` -You can compose them as `foo|bar` although foo's output is `KTable`. -In normal case, when you use `foo` as standalone, this will not work, as the binder does not support `KTable` as the final output. -Note that in the example above, bar's output is still a `KStream`. -We are only able to use `foo` which has a `KTable` output, since we are composing with another function that has `KStream` as its output. +You can compose them as `foo|bar`, but keep in mind that the second function (`bar` in this case) must have a `KTable` as input since the first function (`foo`) has `KTable` as output. ==== Imperative programming model. diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.java index 359de55df..acc1197d4 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.java @@ -44,13 +44,10 @@ import org.springframework.util.StringUtils; * @author Soby Chacko */ class KTableBinder extends - // @checkstyle:off AbstractBinder, ExtendedConsumerProperties, ExtendedProducerProperties> implements ExtendedPropertiesBinder, KafkaStreamsConsumerProperties, KafkaStreamsProducerProperties> { - // @checkstyle:on - private final KafkaStreamsBinderConfigurationProperties binderConfigurationProperties; private final KafkaTopicProvisioner kafkaTopicProvisioner; @@ -111,9 +108,7 @@ class KTableBinder extends @Override protected Binding> doBindProducer(String name, KTable outboundBindTarget, - // @checkstyle:off ExtendedProducerProperties properties) { - // @checkstyle:on throw new UnsupportedOperationException( "No producer level binding is allowed for KTable"); } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java index 48e502e29..599910a66 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java @@ -39,6 +39,7 @@ import org.apache.kafka.streams.StreamsBuilder; import org.apache.kafka.streams.StreamsConfig; import org.apache.kafka.streams.Topology; import org.apache.kafka.streams.kstream.KStream; +import org.apache.kafka.streams.kstream.KTable; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; @@ -298,7 +299,13 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro outboundResolvableType, (Object[]) result, streamsBuilderFactoryBean); } else { - handleSingleKStreamOutbound(resolvableTypes, outboundResolvableType, (KStream) result, outboundDefinitionIterator); + if (KTable.class.isAssignableFrom(result.getClass())) { + handleSingleKStreamOutbound(resolvableTypes, outboundResolvableType != null ? + outboundResolvableType : resolvableType.getGeneric(1), ((KTable) result).toStream(), outboundDefinitionIterator); + } + else { + handleSingleKStreamOutbound(resolvableTypes, outboundResolvableType, (KStream) result, outboundDefinitionIterator); + } } } } @@ -337,8 +344,14 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro outboundResolvableType, (Object[]) result, streamsBuilderFactoryBean); } else { - handleSingleKStreamOutbound(resolvableTypes, outboundResolvableType != null ? - outboundResolvableType : resolvableType.getGeneric(1), (KStream) result, outboundDefinitionIterator); + if (KTable.class.isAssignableFrom(result.getClass())) { + handleSingleKStreamOutbound(resolvableTypes, outboundResolvableType != null ? + outboundResolvableType : resolvableType.getGeneric(1), ((KTable) result).toStream(), outboundDefinitionIterator); + } + else { + handleSingleKStreamOutbound(resolvableTypes, outboundResolvableType != null ? + outboundResolvableType : resolvableType.getGeneric(1), (KStream) result, outboundDefinitionIterator); + } } } } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBindableProxyFactory.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBindableProxyFactory.java index 4a87bf488..80806e92b 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBindableProxyFactory.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBindableProxyFactory.java @@ -145,7 +145,8 @@ public class KafkaStreamsBindableProxyFactory extends AbstractBindableProxyFacto } if (outboundArgument != null && outboundArgument.getRawClass() != null && (!outboundArgument.isArray() && - outboundArgument.getRawClass().isAssignableFrom(KStream.class))) { + (outboundArgument.getRawClass().isAssignableFrom(KStream.class) || + outboundArgument.getRawClass().isAssignableFrom(KTable.class)))) { //Allowing both KStream and KTable on the outbound. // if the type is array, we need to do a late binding as we don't know the number of // output bindings at this point in the flow. @@ -157,12 +158,15 @@ public class KafkaStreamsBindableProxyFactory extends AbstractBindableProxyFacto if (outputBindingsIter.hasNext()) { outputBinding = outputBindingsIter.next(); } - } else { outputBinding = String.format("%s-%s-0", this.functionName, FunctionConstants.DEFAULT_OUTPUT_SUFFIX); } Assert.isTrue(outputBinding != null, "output binding is not inferred."); + // We will only allow KStream targets on the outbound. If the user provides a KTable, + // we still use the KStreamBinder to send it through the outbound. + // In that case before sending, we do a cast from KTable to KStream. + // See KafkaStreamsFunctionsProcessor#setupFunctionInvokerForKafkaStreams for details. KafkaStreamsBindableProxyFactory.this.outputHolders.put(outputBinding, new BoundTargetHolder(getBindingTargetFactory(KStream.class) .createOutput(outputBinding), true));