General polishing and clean up after review

Resolves #692

spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsApplicationSupportAutoConfiguration.java

@@ -36,8 +36,7 @@ import org.springframework.context.annotation.Configuration;
 public class KafkaStreamsApplicationSupportAutoConfiguration {
 
 	@Bean
-	@ConditionalOnProperty("spring.cloud.strea" +
-			"m.kafka.streams.timeWindow.length")
+	@ConditionalOnProperty("spring.cloud.stream.kafka.streams.timeWindow.length")
 	public TimeWindows configuredTimeWindow(
 			KafkaStreamsApplicationSupportProperties processorProperties) {
 		return processorProperties.getTimeWindow().getAdvanceBy() > 0

spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java

@@ -36,7 +36,6 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean
 import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
-import org.springframework.cloud.function.context.FunctionCatalog;
 import org.springframework.cloud.stream.binder.BinderConfiguration;
 import org.springframework.cloud.stream.binder.kafka.streams.function.FunctionDetectorCondition;
 import org.springframework.cloud.stream.binder.kafka.streams.function.KafkaStreamsBindableProxyFactory;
@@ -258,11 +257,11 @@ public class KafkaStreamsBinderSupportAutoConfiguration {
 																	KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue,
 																	KafkaStreamsMessageConversionDelegate kafkaStreamsMessageConversionDelegate,
 																	ObjectProvider<CleanupConfig> cleanupConfig,
-																	FunctionCatalog functionCatalog, KafkaStreamsBindableProxyFactory bindableProxyFactory,
+																	KafkaStreamsBindableProxyFactory bindableProxyFactory,
 																	StreamFunctionProperties streamFunctionProperties) {
 		return new KafkaStreamsFunctionProcessor(bindingServiceProperties, kafkaStreamsExtendedBindingProperties,
 				keyValueSerdeResolver, kafkaStreamsBindingInformationCatalogue, kafkaStreamsMessageConversionDelegate,
-				cleanupConfig.getIfUnique(), functionCatalog, bindableProxyFactory, streamFunctionProperties);
+				cleanupConfig.getIfUnique(), bindableProxyFactory, streamFunctionProperties);
 	}
 
 	@Bean

spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java

@@ -46,7 +46,6 @@ import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.beans.factory.BeanInitializationException;
 import org.springframework.beans.factory.support.BeanDefinitionRegistry;
 import org.springframework.beans.factory.support.RootBeanDefinition;
-import org.springframework.cloud.function.context.FunctionCatalog;
 import org.springframework.cloud.stream.binder.kafka.streams.function.KafkaStreamsBindableProxyFactory;
 import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsConsumerProperties;
 import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsExtendedBindingProperties;
@@ -75,7 +74,6 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro
 	private final KeyValueSerdeResolver keyValueSerdeResolver;
 	private final KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue;
 	private final KafkaStreamsMessageConversionDelegate kafkaStreamsMessageConversionDelegate;
-	private final FunctionCatalog functionCatalog;
 
 	private Set<String> origInputs = new LinkedHashSet<>();
 	private Set<String> origOutputs = new LinkedHashSet<>();
@@ -91,7 +89,6 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro
 										KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue,
 										KafkaStreamsMessageConversionDelegate kafkaStreamsMessageConversionDelegate,
 										CleanupConfig cleanupConfig,
-										FunctionCatalog functionCatalog,
 										KafkaStreamsBindableProxyFactory bindableProxyFactory,
 										StreamFunctionProperties streamFunctionProperties) {
 		super(bindingServiceProperties, kafkaStreamsBindingInformationCatalogue, kafkaStreamsExtendedBindingProperties,
@@ -101,7 +98,6 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro
 		this.keyValueSerdeResolver = keyValueSerdeResolver;
 		this.kafkaStreamsBindingInformationCatalogue = kafkaStreamsBindingInformationCatalogue;
 		this.kafkaStreamsMessageConversionDelegate = kafkaStreamsMessageConversionDelegate;
-		this.functionCatalog = functionCatalog;
 		this.kafkaStreamsBindableProxyFactory = bindableProxyFactory;
 		this.origInputs.addAll(bindableProxyFactory.getInputs());
 		this.origOutputs.addAll(bindableProxyFactory.getOutputs());
@@ -139,15 +135,13 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro
 				outboundResolvableType = iterableResType.getGeneric(i);
 			}
 			else {
-				while (i < inputCount) {
-					if (iterator.hasNext()) {
-						iterableResType = iterableResType.getGeneric(1);
-						if (iterableResType.getRawClass() != null &&
-								functionOrConsumerFound(iterableResType)) {
-							popuateResolvableTypeMap(iterableResType, resolvableTypeMap, iterator);
-						}
-						i++;
+				while (i < inputCount && iterator.hasNext()) {
+					iterableResType = iterableResType.getGeneric(1);
+					if (iterableResType.getRawClass() != null &&
+							functionOrConsumerFound(iterableResType)) {
+						popuateResolvableTypeMap(iterableResType, resolvableTypeMap, iterator);
 					}
+					i++;
 				}
 				outboundResolvableType = iterableResType.getGeneric(1);
 			}

spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBindableProxyFactory.java

@@ -200,21 +200,9 @@ public class KafkaStreamsBindableProxyFactory extends AbstractBindableProxyFacto
 
 	private void bindInput(ResolvableType arg0, String inputName) {
 		if (arg0.getRawClass() != null) {
-			if (arg0.getRawClass().isAssignableFrom(KStream.class)) {
-				KafkaStreamsBindableProxyFactory.this.inputHolders.put(inputName,
-						new BoundTargetHolder(getBindingTargetFactory(KStream.class)
-								.createInput(inputName), true));
-			}
-			else if (arg0.getRawClass().isAssignableFrom(KTable.class)) {
-				KafkaStreamsBindableProxyFactory.this.inputHolders.put(inputName,
-						new BoundTargetHolder(getBindingTargetFactory(KTable.class)
-								.createInput(inputName), true));
-			}
-			else {
-				KafkaStreamsBindableProxyFactory.this.inputHolders.put(inputName,
-						new BoundTargetHolder(getBindingTargetFactory(GlobalKTable.class)
-								.createInput(inputName), true));
-			}
+			KafkaStreamsBindableProxyFactory.this.inputHolders.put(inputName,
+					new BoundTargetHolder(getBindingTargetFactory(arg0.getRawClass())
+							.createInput(inputName), true));
 		}
 
 		BeanDefinitionRegistry registry = (BeanDefinitionRegistry) beanFactory;

spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionWrapperDetector.java

@@ -1,48 +0,0 @@
-/*
- * Copyright 2019-2019 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.stream.binder.kafka.streams.function;
-
-import java.lang.reflect.Type;
-
-import org.apache.kafka.streams.kstream.GlobalKTable;
-import org.apache.kafka.streams.kstream.KStream;
-import org.apache.kafka.streams.kstream.KTable;
-
-import org.springframework.boot.autoconfigure.AutoConfigureBefore;
-import org.springframework.cloud.function.context.WrapperDetector;
-import org.springframework.cloud.stream.config.BinderFactoryAutoConfiguration;
-import org.springframework.context.annotation.Configuration;
-
-/**
- * @author Soby Chacko
- * @since 2.2.0
- */
-@Configuration
-@AutoConfigureBefore(BinderFactoryAutoConfiguration.class)
-public class KafkaStreamsFunctionWrapperDetector implements WrapperDetector {
-	@Override
-	public boolean isWrapper(Type type) {
-		if (type instanceof Class<?>) {
-			Class<?> cls = (Class<?>) type;
-			return KStream.class.isAssignableFrom(cls) ||
-					KTable.class.isAssignableFrom(cls) ||
-					GlobalKTable.class.isAssignableFrom(cls);
-		}
-		return false;
-	}
-}
-

spring-cloud-stream-binder-kafka-streams/src/main/resources/META-INF/spring.factories

@@ -3,7 +3,3 @@ org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
   org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsApplicationSupportAutoConfiguration,\
   org.springframework.cloud.stream.binder.kafka.streams.function.KafkaStreamsFunctionAutoConfiguration
 
-org.springframework.cloud.function.context.WrapperDetector=\
-  org.springframework.cloud.stream.binder.kafka.streams.function.KafkaStreamsFunctionWrapperDetector
-
-