diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsApplicationSupportAutoConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsApplicationSupportAutoConfiguration.java index b81e5c5b6..91387e3b2 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsApplicationSupportAutoConfiguration.java +++ b/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 diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java index 08f6a9968..3160693a4 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java +++ b/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, - 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 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 d30ab98bb..82627f235 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 @@ -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 origInputs = new LinkedHashSet<>(); private Set 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); } 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 0085a9624..d91cc4471 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 @@ -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; diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionWrapperDetector.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionWrapperDetector.java deleted file mode 100644 index aa2a90d40..000000000 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionWrapperDetector.java +++ /dev/null @@ -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; - } -} - diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/resources/META-INF/spring.factories b/spring-cloud-stream-binder-kafka-streams/src/main/resources/META-INF/spring.factories index 3f422823d..e42ab3c94 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/resources/META-INF/spring.factories +++ b/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 - -