GH-1085: Allow KTable binding on the outbound

At the moment, Kafka Streams binder only allows KStream bindings on the outbound.
There is a delegation mechanism in which we stil can use KStream for output binding
while allowing the applications to provide a KTable type as the function return type.

Update docs.

Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1085

Resolves #1105

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<KStream<String, String>, KTable<String, String>> foo() {
+    return KStream::toTable;
+    };
+}
+
+@Bean
+public Function<KTable<String, String>, KStream<String, String>> 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<KTable<String, String>, KStream<String, String>> 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.
 

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<KTable<Object, Object>, ExtendedConsumerProperties<KafkaStreamsConsumerProperties>, ExtendedProducerProperties<KafkaStreamsProducerProperties>>
 		implements
 		ExtendedPropertiesBinder<KTable<Object, Object>, KafkaStreamsConsumerProperties, KafkaStreamsProducerProperties> {
 
-	// @checkstyle:on
-
 	private final KafkaStreamsBinderConfigurationProperties binderConfigurationProperties;
 
 	private final KafkaTopicProvisioner kafkaTopicProvisioner;
@@ -111,9 +108,7 @@ class KTableBinder extends
 	@Override
 	protected Binding<KTable<Object, Object>> doBindProducer(String name,
 			KTable<Object, Object> outboundBindTarget,
-			// @checkstyle:off
 			ExtendedProducerProperties<KafkaStreamsProducerProperties> properties) {
-		// @checkstyle:on
 		throw new UnsupportedOperationException(
 				"No producer level binding is allowed for KTable");
 	}

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

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