From 2b1be3754dee7163edd0051162c72e6a44dcb795 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Thu, 26 Sep 2019 11:55:27 -0400 Subject: [PATCH] Remove deprecations Remove deprecated fields, methods and classes in preparation for the 3.0 GA Release, both in Kafka and Kafka Streams binders. Resolves #746 --- .../properties/KafkaAdminProperties.java | 39 --- .../KafkaBinderConfigurationProperties.java | 283 ------------------ .../properties/KafkaConsumerProperties.java | 23 -- .../properties/KafkaProducerProperties.java | 24 -- ...msApplicationSupportAutoConfiguration.java | 48 --- ...kaStreamsApplicationSupportProperties.java | 72 ----- .../main/resources/META-INF/spring.factories | 1 - .../StreamToGlobalKTableFunctionTests.java | 3 - ...serializationErrorHandlerByKafkaTests.java | 9 +- .../binder/kstream/integTest-1.properties | 2 - .../stream/binder/kafka/AdminConfigTests.java | 19 +- 11 files changed, 4 insertions(+), 519 deletions(-) delete mode 100644 spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaAdminProperties.java delete mode 100644 spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsApplicationSupportAutoConfiguration.java delete mode 100644 spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsApplicationSupportProperties.java diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaAdminProperties.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaAdminProperties.java deleted file mode 100644 index 5161d38fb..000000000 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaAdminProperties.java +++ /dev/null @@ -1,39 +0,0 @@ -/* - * Copyright 2018-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.properties; - -import java.util.Map; - -/** - * Properties for configuring topics. - * - * @author Gary Russell - * @since 2.0 - * @deprecated in favor of {@link KafkaTopicProperties} - */ -@Deprecated -public class KafkaAdminProperties extends KafkaTopicProperties { - - public Map getConfiguration() { - return getProperties(); - } - - public void setConfiguration(Map configuration) { - setProperties(configuration); - } - -} diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java index fc2965d37..3488a74d6 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java +++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java @@ -32,7 +32,6 @@ import org.apache.kafka.clients.producer.ProducerConfig; import org.springframework.boot.autoconfigure.kafka.KafkaProperties; import org.springframework.boot.context.properties.ConfigurationProperties; -import org.springframework.boot.context.properties.DeprecatedConfigurationProperty; import org.springframework.cloud.stream.binder.HeaderMode; import org.springframework.cloud.stream.binder.ProducerProperties; import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties.CompressionType; @@ -64,8 +63,6 @@ public class KafkaBinderConfigurationProperties { private final KafkaProperties kafkaProperties; - private String[] zkNodes = new String[] { "localhost" }; - /** * Arbitrary kafka properties that apply to both producers and consumers. */ @@ -81,48 +78,22 @@ public class KafkaBinderConfigurationProperties { */ private Map producerProperties = new HashMap<>(); - private String defaultZkPort = "2181"; - private String[] brokers = new String[] { "localhost" }; private String defaultBrokerPort = "9092"; private String[] headers = new String[] {}; - private int offsetUpdateTimeWindow = 10000; - - private int offsetUpdateCount; - - private int offsetUpdateShutdownTimeout = 2000; - - private int maxWait = 100; - private boolean autoCreateTopics = true; private boolean autoAddPartitions; - private int socketBufferSize = 2097152; - - /** - * ZK session timeout in milliseconds. - */ - private int zkSessionTimeout = 10000; - - /** - * ZK Connection timeout in milliseconds. - */ - private int zkConnectionTimeout = 10000; - private String requiredAcks = "1"; private short replicationFactor = 1; - private int fetchSize = 1024 * 1024; - private int minPartitionCount = 1; - private int queueSize = 8192; - /** * Time to wait to get partition information in seconds; default 60. */ @@ -149,17 +120,6 @@ public class KafkaBinderConfigurationProperties { return this.transaction; } - /** - * No longer used. - * @return the connection String - * @deprecated connection to zookeeper is no longer necessary - */ - @DeprecatedConfigurationProperty(reason = "Not used since 2.0") - @Deprecated - public String getZkConnectionString() { - return toConnectionString(this.zkNodes, this.defaultZkPort); - } - public String getKafkaConnectionString() { return toConnectionString(this.brokers, this.defaultBrokerPort); } @@ -172,72 +132,6 @@ public class KafkaBinderConfigurationProperties { return this.headers; } - /** - * No longer used. - * @return the window. - * @deprecated No longer used by the binder - */ - @Deprecated - @DeprecatedConfigurationProperty(reason = "Not used since 2.0") - public int getOffsetUpdateTimeWindow() { - return this.offsetUpdateTimeWindow; - } - - /** - * No longer used. - * @return the count. - * @deprecated No longer used by the binder - */ - @Deprecated - @DeprecatedConfigurationProperty(reason = "Not used since 2.0") - public int getOffsetUpdateCount() { - return this.offsetUpdateCount; - } - - /** - * No longer used. - * @return the timeout. - * @deprecated No longer used by the binder - */ - @Deprecated - @DeprecatedConfigurationProperty(reason = "Not used since 2.0") - public int getOffsetUpdateShutdownTimeout() { - return this.offsetUpdateShutdownTimeout; - } - - /** - * Zookeeper nodes. - * @return the nodes. - * @deprecated connection to zookeeper is no longer necessary - */ - @Deprecated - @DeprecatedConfigurationProperty(reason = "No longer necessary since 2.0") - public String[] getZkNodes() { - return this.zkNodes; - } - - /** - * Zookeeper nodes. - * @param zkNodes the nodes. - * @deprecated connection to zookeeper is no longer necessary - */ - @Deprecated - @DeprecatedConfigurationProperty(reason = "No longer necessary since 2.0") - public void setZkNodes(String... zkNodes) { - this.zkNodes = zkNodes; - } - - /** - * Zookeeper port. - * @param defaultZkPort the port. - * @deprecated connection to zookeeper is no longer necessary - */ - @Deprecated - @DeprecatedConfigurationProperty(reason = "No longer necessary since 2.0") - public void setDefaultZkPort(String defaultZkPort) { - this.defaultZkPort = defaultZkPort; - } - public String[] getBrokers() { return this.brokers; } @@ -254,83 +148,6 @@ public class KafkaBinderConfigurationProperties { this.headers = headers; } - /** - * No longer used. - * @param offsetUpdateTimeWindow the window. - * @deprecated No longer used by the binder - */ - @Deprecated - @DeprecatedConfigurationProperty(reason = "Not used since 2.0") - public void setOffsetUpdateTimeWindow(int offsetUpdateTimeWindow) { - this.offsetUpdateTimeWindow = offsetUpdateTimeWindow; - } - - /** - * No longer used. - * @param offsetUpdateCount the count. - * @deprecated No longer used by the binder - */ - @Deprecated - @DeprecatedConfigurationProperty(reason = "Not used since 2.0") - public void setOffsetUpdateCount(int offsetUpdateCount) { - this.offsetUpdateCount = offsetUpdateCount; - } - - /** - * No longer used. - * @param offsetUpdateShutdownTimeout the timeout. - * @deprecated No longer used by the binder - */ - @Deprecated - @DeprecatedConfigurationProperty(reason = "Not used since 2.0") - public void setOffsetUpdateShutdownTimeout(int offsetUpdateShutdownTimeout) { - this.offsetUpdateShutdownTimeout = offsetUpdateShutdownTimeout; - } - - /** - * Zookeeper session timeout. - * @return the timeout. - * @deprecated connection to zookeeper is no longer necessary - */ - @Deprecated - @DeprecatedConfigurationProperty(reason = "No longer necessary since 2.0") - public int getZkSessionTimeout() { - return this.zkSessionTimeout; - } - - /** - * Zookeeper session timeout. - * @param zkSessionTimeout the timout - * @deprecated connection to zookeeper is no longer necessary - */ - @Deprecated - @DeprecatedConfigurationProperty(reason = "No longer necessary since 2.0") - public void setZkSessionTimeout(int zkSessionTimeout) { - this.zkSessionTimeout = zkSessionTimeout; - } - - /** - * Zookeeper connection timeout. - * @return the timout. - * @deprecated connection to zookeeper is no longer necessary - */ - @Deprecated - @DeprecatedConfigurationProperty(reason = "No longer necessary since 2.0") - public int getZkConnectionTimeout() { - return this.zkConnectionTimeout; - } - - /** - * Zookeeper connection timeout. - * @param zkConnectionTimeout the timeout. - * @deprecated connection to zookeeper is no longer necessary - */ - @Deprecated - @DeprecatedConfigurationProperty(reason = "No longer necessary since 2.0") - public void setZkConnectionTimeout(int zkConnectionTimeout) { - this.zkConnectionTimeout = zkConnectionTimeout; - } - /** * Converts an array of host values to a comma-separated String. It will append the * default port value, if not already specified. @@ -351,28 +168,6 @@ public class KafkaBinderConfigurationProperties { return StringUtils.arrayToCommaDelimitedString(fullyFormattedHosts); } - /** - * No longer used. - * @return the wait. - * @deprecated No longer used by the binder - */ - @Deprecated - @DeprecatedConfigurationProperty(reason = "Not used since 2.0") - public int getMaxWait() { - return this.maxWait; - } - - /** - * No longer user. - * @param maxWait the wait. - * @deprecated No longer used by the binder - */ - @Deprecated - @DeprecatedConfigurationProperty(reason = "Not used since 2.0") - public void setMaxWait(int maxWait) { - this.maxWait = maxWait; - } - public String getRequiredAcks() { return this.requiredAcks; } @@ -389,28 +184,6 @@ public class KafkaBinderConfigurationProperties { this.replicationFactor = replicationFactor; } - /** - * No longer used. - * @return the size. - * @deprecated No longer used by the binder - */ - @Deprecated - @DeprecatedConfigurationProperty(reason = "Not used since 2.0") - public int getFetchSize() { - return this.fetchSize; - } - - /** - * No longer used. - * @param fetchSize the size. - * @deprecated No longer used by the binder - */ - @Deprecated - @DeprecatedConfigurationProperty(reason = "Not used since 2.0") - public void setFetchSize(int fetchSize) { - this.fetchSize = fetchSize; - } - public int getMinPartitionCount() { return this.minPartitionCount; } @@ -427,28 +200,6 @@ public class KafkaBinderConfigurationProperties { this.healthTimeout = healthTimeout; } - /** - * No longer used. - * @return the queue size. - * @deprecated No longer used by the binder - */ - @Deprecated - @DeprecatedConfigurationProperty(reason = "Not used since 2.0") - public int getQueueSize() { - return this.queueSize; - } - - /** - * No longer used. - * @param queueSize the queue size. - * @deprecated No longer used by the binder - */ - @Deprecated - @DeprecatedConfigurationProperty(reason = "Not used since 2.0") - public void setQueueSize(int queueSize) { - this.queueSize = queueSize; - } - public boolean isAutoCreateTopics() { return this.autoCreateTopics; } @@ -465,30 +216,6 @@ public class KafkaBinderConfigurationProperties { this.autoAddPartitions = autoAddPartitions; } - /** - * No longer used; set properties such as this via {@link #getConfiguration() - * configuration}. - * @return the size. - * @deprecated No longer used by the binder - */ - @Deprecated - @DeprecatedConfigurationProperty(reason = "Not used since 2.0, set properties such as this via 'configuration'") - public int getSocketBufferSize() { - return this.socketBufferSize; - } - - /** - * No longer used; set properties such as this via {@link #getConfiguration() - * configuration}. - * @param socketBufferSize the size. - * @deprecated No longer used by the binder - */ - @Deprecated - @DeprecatedConfigurationProperty(reason = "Not used since 2.0, set properties such as this via 'configuration'") - public void setSocketBufferSize(int socketBufferSize) { - this.socketBufferSize = socketBufferSize; - } - public Map getConfiguration() { return this.configuration; } @@ -800,16 +527,6 @@ public class KafkaBinderConfigurationProperties { this.kafkaProducerProperties.setConfiguration(configuration); } - @SuppressWarnings("deprecation") - public KafkaAdminProperties getAdmin() { - return this.kafkaProducerProperties.getAdmin(); - } - - @SuppressWarnings("deprecation") - public void setAdmin(KafkaAdminProperties admin) { - this.kafkaProducerProperties.setAdmin(admin); - } - public KafkaTopicProperties getTopic() { return this.kafkaProducerProperties.getTopic(); } diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java index 5641fbb14..406a1f100 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java +++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java @@ -265,29 +265,6 @@ public class KafkaConsumerProperties { this.destinationIsPattern = destinationIsPattern; } - /** - * No longer used; get properties such as this via {@link #getTopic()}. - * @return Kafka admin properties - * @deprecated No longer used - */ - @Deprecated - @DeprecatedConfigurationProperty(reason = "Not used since 2.1.1, set properties such as this via 'topic'") - @SuppressWarnings("deprecation") - public KafkaAdminProperties getAdmin() { - // Temporary workaround to copy the topic properties to the admin one. - final KafkaAdminProperties kafkaAdminProperties = new KafkaAdminProperties(); - kafkaAdminProperties.setReplicationFactor(this.topic.getReplicationFactor()); - kafkaAdminProperties.setReplicasAssignments(this.topic.getReplicasAssignments()); - kafkaAdminProperties.setConfiguration(this.topic.getProperties()); - return kafkaAdminProperties; - } - - @Deprecated - @SuppressWarnings("deprecation") - public void setAdmin(KafkaAdminProperties admin) { - this.topic = admin; - } - public KafkaTopicProperties getTopic() { return this.topic; } diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java index 068c0c8a7..640315313 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java +++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java @@ -21,7 +21,6 @@ import java.util.Map; import javax.validation.constraints.NotNull; -import org.springframework.boot.context.properties.DeprecatedConfigurationProperty; import org.springframework.expression.Expression; /** @@ -121,29 +120,6 @@ public class KafkaProducerProperties { this.configuration = configuration; } - /** - * No longer used; get properties such as this via {@link #getTopic()}. - * @return Kafka admin properties - * @deprecated No longer used - */ - @Deprecated - @DeprecatedConfigurationProperty(reason = "Not used since 2.1.1, set properties such as this via 'topic'") - @SuppressWarnings("deprecation") - public KafkaAdminProperties getAdmin() { - // Temporary workaround to copy the topic properties to the admin one. - final KafkaAdminProperties kafkaAdminProperties = new KafkaAdminProperties(); - kafkaAdminProperties.setReplicationFactor(this.topic.getReplicationFactor()); - kafkaAdminProperties.setReplicasAssignments(this.topic.getReplicasAssignments()); - kafkaAdminProperties.setConfiguration(this.topic.getProperties()); - return kafkaAdminProperties; - } - - @Deprecated - @SuppressWarnings("deprecation") - public void setAdmin(KafkaAdminProperties admin) { - this.topic = admin; - } - public KafkaTopicProperties getTopic() { return this.topic; } 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 deleted file mode 100644 index 91387e3b2..000000000 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsApplicationSupportAutoConfiguration.java +++ /dev/null @@ -1,48 +0,0 @@ -/* - * Copyright 2017-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; - -import org.apache.kafka.streams.kstream.TimeWindows; - -import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; -import org.springframework.boot.context.properties.EnableConfigurationProperties; -import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsApplicationSupportProperties; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; - -/** - * Application support configuration for Kafka Streams binder. - * - * @deprecated Features provided on this class can be directly configured in the application itself using Kafka Streams. - * @author Soby Chacko - */ -@Configuration -@EnableConfigurationProperties(KafkaStreamsApplicationSupportProperties.class) -@Deprecated -public class KafkaStreamsApplicationSupportAutoConfiguration { - - @Bean - @ConditionalOnProperty("spring.cloud.stream.kafka.streams.timeWindow.length") - public TimeWindows configuredTimeWindow( - KafkaStreamsApplicationSupportProperties processorProperties) { - return processorProperties.getTimeWindow().getAdvanceBy() > 0 - ? TimeWindows.of(processorProperties.getTimeWindow().getLength()) - .advanceBy(processorProperties.getTimeWindow().getAdvanceBy()) - : TimeWindows.of(processorProperties.getTimeWindow().getLength()); - } - -} diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsApplicationSupportProperties.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsApplicationSupportProperties.java deleted file mode 100644 index 04eeaa4b1..000000000 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsApplicationSupportProperties.java +++ /dev/null @@ -1,72 +0,0 @@ -/* - * Copyright 2017-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.properties; - -import org.springframework.boot.context.properties.ConfigurationProperties; - -/** - * {@link ConfigurationProperties} that can be used by end user Kafka Stream applications. - * This class provides convenient ways to access the commonly used kafka stream properties - * from the user application. For example, windowing operations are common use cases in - * stream processing and one can provide window specific properties at runtime and use - * those properties in the applications using this class. - * - * @deprecated The properties exposed by this class can be used directly on Kafka Streams API in the application. - * @author Soby Chacko - */ -@ConfigurationProperties("spring.cloud.stream.kafka.streams") -@Deprecated -public class KafkaStreamsApplicationSupportProperties { - - private TimeWindow timeWindow; - - public TimeWindow getTimeWindow() { - return this.timeWindow; - } - - public void setTimeWindow(TimeWindow timeWindow) { - this.timeWindow = timeWindow; - } - - /** - * Properties required by time windows. - */ - public static class TimeWindow { - - private int length; - - private int advanceBy; - - public int getLength() { - return this.length; - } - - public void setLength(int length) { - this.length = length; - } - - public int getAdvanceBy() { - return this.advanceBy; - } - - public void setAdvanceBy(int advanceBy) { - this.advanceBy = advanceBy; - } - - } - -} 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 e42ab3c94..e1ad75254 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 @@ -1,5 +1,4 @@ org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsBinderSupportAutoConfiguration,\ - org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsApplicationSupportAutoConfiguration,\ org.springframework.cloud.stream.binder.kafka.streams.function.KafkaStreamsFunctionAutoConfiguration diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToGlobalKTableFunctionTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToGlobalKTableFunctionTests.java index dc935f3db..355be2893 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToGlobalKTableFunctionTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToGlobalKTableFunctionTests.java @@ -38,8 +38,6 @@ import org.junit.Test; import org.springframework.boot.SpringApplication; import org.springframework.boot.WebApplicationType; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.context.properties.EnableConfigurationProperties; -import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsApplicationSupportProperties; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; @@ -175,7 +173,6 @@ public class StreamToGlobalKTableFunctionTests { } @EnableAutoConfiguration - @EnableConfigurationProperties(KafkaStreamsApplicationSupportProperties.class) public static class OrderEnricherApplication { @Bean diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DeserializationErrorHandlerByKafkaTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DeserializationErrorHandlerByKafkaTests.java index d9953c900..2570952fb 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DeserializationErrorHandlerByKafkaTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DeserializationErrorHandlerByKafkaTests.java @@ -34,15 +34,12 @@ import org.junit.ClassRule; import org.junit.Test; import org.junit.runner.RunWith; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.boot.test.mock.mockito.SpyBean; import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.annotation.StreamListener; import org.springframework.cloud.stream.binder.kafka.streams.annotations.KafkaStreamsProcessor; -import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsApplicationSupportProperties; import org.springframework.context.annotation.PropertySource; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; @@ -197,12 +194,8 @@ public abstract class DeserializationErrorHandlerByKafkaTests { @EnableBinding(KafkaStreamsProcessor.class) @EnableAutoConfiguration @PropertySource("classpath:/org/springframework/cloud/stream/binder/kstream/integTest-1.properties") - @EnableConfigurationProperties(KafkaStreamsApplicationSupportProperties.class) public static class WordCountProcessorApplication { - @Autowired - private TimeWindows timeWindows; - @StreamListener("input") @SendTo("output") public KStream process(KStream input) { @@ -212,7 +205,7 @@ public abstract class DeserializationErrorHandlerByKafkaTests { value -> Arrays.asList(value.toLowerCase().split("\\W+"))) .map((key, value) -> new KeyValue<>(value, value)) .groupByKey(Serialized.with(Serdes.String(), Serdes.String())) - .windowedBy(timeWindows).count(Materialized.as("foo-WordCounts-x")) + .windowedBy(TimeWindows.of(5000)).count(Materialized.as("foo-WordCounts-x")) .toStream().map((key, value) -> new KeyValue<>(null, "Count for " + key.key() + " : " + value)); } diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/resources/org/springframework/cloud/stream/binder/kstream/integTest-1.properties b/spring-cloud-stream-binder-kafka-streams/src/test/resources/org/springframework/cloud/stream/binder/kstream/integTest-1.properties index 4a0189c19..6d983a0b1 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/resources/org/springframework/cloud/stream/binder/kstream/integTest-1.properties +++ b/spring-cloud-stream-binder-kafka-streams/src/test/resources/org/springframework/cloud/stream/binder/kstream/integTest-1.properties @@ -4,5 +4,3 @@ spring.cloud.stream.bindings.output.contentType=application/json spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000 spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde=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.timeWindow.length=5000 -spring.cloud.stream.kafka.streams.timeWindow.advanceBy=0 diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AdminConfigTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AdminConfigTests.java index 6bef51132..8bbad166f 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AdminConfigTests.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AdminConfigTests.java @@ -44,9 +44,9 @@ import static org.assertj.core.api.Assertions.assertThat; @SpringBootTest(classes = { KafkaBinderConfiguration.class, BindingServiceConfiguration.class }) @TestPropertySource(properties = { - "spring.cloud.stream.kafka.bindings.input.consumer.admin.replication-factor=2", - "spring.cloud.stream.kafka.bindings.input.consumer.admin.replicas-assignments.0=0,1", - "spring.cloud.stream.kafka.bindings.input.consumer.admin.configuration.message.format.version=0.9.0.0", + "spring.cloud.stream.kafka.bindings.input.consumer.topic.replication-factor=2", + "spring.cloud.stream.kafka.bindings.input.consumer.topic.replicas-assignments.0=0,1", + "spring.cloud.stream.kafka.bindings.input.consumer.topic.properties.message.format.version=0.9.0.0", "spring.cloud.stream.kafka.bindings.secondInput.consumer.topic.replication-factor=3", "spring.cloud.stream.kafka.bindings.secondInput.consumer.topic.replicas-assignments.0=0,1", "spring.cloud.stream.kafka.bindings.secondInput.consumer.topic.properties.message.format.version=0.9.1.0", @@ -60,19 +60,6 @@ public class AdminConfigTests { @Autowired private KafkaMessageChannelBinder binder; - @Test - public void testDeprecatedAdminConfigurationToMapTopicProperties() { - final KafkaConsumerProperties consumerProps = this.binder - .getExtendedConsumerProperties("input"); - final KafkaTopicProperties kafkaTopicProperties = consumerProps.getTopic(); - - assertThat(kafkaTopicProperties.getReplicationFactor()).isEqualTo((short) 2); - assertThat(kafkaTopicProperties.getReplicasAssignments().get(0)) - .isEqualTo(Arrays.asList(0, 1)); - assertThat(kafkaTopicProperties.getProperties().get("message.format.version")) - .isEqualTo("0.9.0.0"); - } - @Test public void testConsumerTopicProperties() { final KafkaConsumerProperties consumerProperties = this.binder