From 9f724ea27a2b403d3dcc846ce5c7c6805cdff108 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 3 Apr 2018 17:09:10 -0400 Subject: [PATCH] GH-640: Upgrade kafka version to 1.1.0 Resolves https://github.com/spring-projects/spring-kafka/issues/640 Polishing - Logging - remove unnecessary excludes - add slf4j over log4j2 - add logging to -test module tests - suppress some noise from embedded kafka Resolves spring-projects/spring-kafka#278 * Ignore `StreamsBuilderFactoryBeanTests` on Windows * Close remaining dangling `KafkaConsumer`s in tests --- build.gradle | 29 +-- .../kafka/test/rule/KafkaEmbedded.java | 37 +-- .../kafka/test/rule/Log4j2LevelAdjuster.java | 246 ++++++++++++++++++ .../kafka/test/rule/Log4jLevelAdjuster.java | 108 -------- .../src/test/resources/log4j2-test.xml | 23 ++ .../core/StreamsBuilderFactoryBeanTests.java | 5 +- .../kstream/KafkaStreamsBranchTests.java | 4 + .../KafkaStreamsJsonSerializationTests.java | 30 ++- .../kafka/kstream/KafkaStreamsTests.java | 5 +- .../listener/TransactionalContainerTests.java | 1 + .../KafkaJaasLoginModuleInitializerTests.java | 2 +- .../src/test/resources/log4j.properties | 11 - .../src/test/resources/log4j2-test.xml | 24 ++ 13 files changed, 357 insertions(+), 168 deletions(-) create mode 100644 spring-kafka-test/src/main/java/org/springframework/kafka/test/rule/Log4j2LevelAdjuster.java delete mode 100644 spring-kafka-test/src/main/java/org/springframework/kafka/test/rule/Log4jLevelAdjuster.java create mode 100644 spring-kafka-test/src/test/resources/log4j2-test.xml delete mode 100644 spring-kafka/src/test/resources/log4j.properties create mode 100644 spring-kafka/src/test/resources/log4j2-test.xml diff --git a/build.gradle b/build.gradle index bd1e6845..07355ab6 100644 --- a/build.gradle +++ b/build.gradle @@ -78,10 +78,10 @@ subprojects { subproject -> junitJupiterVersion = '5.1.0' junitPlatformVersion = '1.1.0' junitVintageVersion = '5.1.0' - kafkaVersion = '1.0.1' + kafkaVersion = '1.1.0' + log4jVersion = '2.11.0' mockitoVersion = '2.15.0' scalaVersion = '2.11' - slf4jVersion = '1.7.25' springRetryVersion = '1.2.2.RELEASE' springVersion = '5.0.4.RELEASE' springDataCommonsVersion = '2.0.4.RELEASE' @@ -107,6 +107,9 @@ subprojects { subproject -> // To avoid compiler warnings about @API annotations in JUnit code testCompileOnly 'org.apiguardian:apiguardian-api:1.0.0' + + testRuntime "org.apache.logging.log4j:log4j-core:$log4jVersion" + testRuntime "org.apache.logging.log4j:log4j-slf4j-impl:$log4jVersion" } // enable all compiler warnings; individual projects may customize further @@ -177,14 +180,8 @@ project ('spring-kafka') { compile "org.springframework:spring-messaging:$springVersion" compile "org.springframework:spring-tx:$springVersion" compile "org.springframework.retry:spring-retry:$springRetryVersion" - compile ("org.apache.kafka:kafka-clients:$kafkaVersion") { - exclude group: 'org.slf4j', module: 'slf4j-api' - } - compile ("org.apache.kafka:kafka-streams:$kafkaVersion") { - optional it - exclude group: 'org.slf4j', module: 'slf4j-api' - exclude group: 'org.slf4j', module: 'slf4j-log4j12' - } + compile "org.apache.kafka:kafka-clients:$kafkaVersion" + compile ("org.apache.kafka:kafka-streams:$kafkaVersion", optional) compile ("com.fasterxml.jackson.core:jackson-core:$jacksonVersion", optional) compile ("com.fasterxml.jackson.core:jackson-databind:$jacksonVersion", optional) @@ -196,8 +193,6 @@ project ('spring-kafka') { testCompile project (":spring-kafka-test") testCompile "org.assertj:assertj-core:$assertjVersion" testCompile "org.springframework:spring-tx:$springVersion" - - testRuntime "org.slf4j:slf4j-log4j12:$slf4jVersion" } } @@ -211,13 +206,8 @@ project ('spring-kafka-test') { compile ("org.apache.kafka:kafka-clients:$kafkaVersion:test") - compile ("org.apache.kafka:kafka_$scalaVersion:$kafkaVersion") { - exclude group: 'org.slf4j', module: 'slf4j-log4j12' - } - - compile ("org.apache.kafka:kafka_$scalaVersion:$kafkaVersion:test") { - exclude group: 'org.slf4j', module: 'slf4j-log4j12' - } + compile "org.apache.kafka:kafka_$scalaVersion:$kafkaVersion" + compile "org.apache.kafka:kafka_$scalaVersion:$kafkaVersion:test" compile ("junit:junit:$junit4Version") { exclude group: 'org.hamcrest', module: 'hamcrest-core' @@ -228,6 +218,7 @@ project ('spring-kafka-test') { compile ("org.hamcrest:hamcrest-all:$hamcrestVersion", optional) compile ("org.assertj:assertj-core:$assertjVersion", optional) + compile ("org.apache.logging.log4j:log4j-core:$log4jVersion", optional) } } diff --git a/spring-kafka-test/src/main/java/org/springframework/kafka/test/rule/KafkaEmbedded.java b/spring-kafka-test/src/main/java/org/springframework/kafka/test/rule/KafkaEmbedded.java index 27f6d6a1..1601e59e 100644 --- a/spring-kafka-test/src/main/java/org/springframework/kafka/test/rule/KafkaEmbedded.java +++ b/spring-kafka-test/src/main/java/org/springframework/kafka/test/rule/KafkaEmbedded.java @@ -46,7 +46,6 @@ import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.security.auth.SecurityProtocol; -import org.apache.kafka.common.utils.AppInfoParser; import org.apache.kafka.common.utils.Time; import org.junit.rules.ExternalResource; @@ -89,27 +88,27 @@ public class KafkaEmbedded extends ExternalResource implements KafkaRule, Initia public static final long METADATA_PROPAGATION_TIMEOUT = 10000L; - private static final String clientVersion; +// private static final String clientVersion; private static final Method testUtilsCreateBrokerConfigMethod; static { - clientVersion = AppInfoParser.getVersion(); - if (clientVersion.startsWith("1.1.")) { - try { - testUtilsCreateBrokerConfigMethod = TestUtils.class.getDeclaredMethod("createBrokerConfig", - int.class, String.class, boolean.class, boolean.class, int.class, - scala.Option.class, scala.Option.class, scala.Option.class, - boolean.class, boolean.class, int.class, boolean.class, int.class, boolean.class, - int.class, scala.Option.class, int.class, boolean.class); - } - catch (NoSuchMethodException | SecurityException e) { - throw new RuntimeException("Failed to determine TestUtils.createBrokerConfig() method"); - } - } - else { +// clientVersion = AppInfoParser.getVersion(); +// if (clientVersion.startsWith("1.1.")) { +// try { +// testUtilsCreateBrokerConfigMethod = TestUtils.class.getDeclaredMethod("createBrokerConfig", +// int.class, String.class, boolean.class, boolean.class, int.class, +// scala.Option.class, scala.Option.class, scala.Option.class, +// boolean.class, boolean.class, int.class, boolean.class, int.class, boolean.class, +// int.class, scala.Option.class, int.class, boolean.class); +// } +// catch (NoSuchMethodException | SecurityException e) { +// throw new RuntimeException("Failed to determine TestUtils.createBrokerConfig() method"); +// } +// } +// else { testUtilsCreateBrokerConfigMethod = null; - } +// } } private final int count; @@ -222,6 +221,8 @@ public class KafkaEmbedded extends ExternalResource implements KafkaRule, Initia brokerConfigProperties.setProperty(KafkaConfig.ReplicaSocketTimeoutMsProp(), "1000"); brokerConfigProperties.setProperty(KafkaConfig.ControllerSocketTimeoutMsProp(), "1000"); brokerConfigProperties.setProperty(KafkaConfig.OffsetsTopicReplicationFactorProp(), "1"); + brokerConfigProperties.setProperty(KafkaConfig.ReplicaHighWatermarkCheckpointIntervalMsProp(), + String.valueOf(Long.MAX_VALUE)); if (this.brokerProperties != null) { this.brokerProperties.forEach(brokerConfigProperties::put); } @@ -251,7 +252,7 @@ public class KafkaEmbedded extends ExternalResource implements KafkaRule, Initia scala.Option.apply(null), scala.Option.apply(null), scala.Option.apply(null), - true, false, 0, false, 0, false, 0, scala.Option.apply(null), 1); + true, false, 0, false, 0, false, 0, scala.Option.apply(null), 1, false); } else { try { diff --git a/spring-kafka-test/src/main/java/org/springframework/kafka/test/rule/Log4j2LevelAdjuster.java b/spring-kafka-test/src/main/java/org/springframework/kafka/test/rule/Log4j2LevelAdjuster.java new file mode 100644 index 00000000..9f9c1c1f --- /dev/null +++ b/spring-kafka-test/src/main/java/org/springframework/kafka/test/rule/Log4j2LevelAdjuster.java @@ -0,0 +1,246 @@ +/* + * Copyright 2018 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 + * + * http://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.kafka.test.rule; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; +import java.util.stream.Stream; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.logging.log4j.Level; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.core.LoggerContext; +import org.apache.logging.log4j.core.config.Configuration; +import org.apache.logging.log4j.core.config.LoggerConfig; +import org.junit.rules.MethodRule; +import org.junit.runners.model.FrameworkMethod; +import org.junit.runners.model.Statement; + +import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; + +/** + * A JUnit method {@link org.junit.Rule} that changes the Log4J 2 logger level for a set of classes + * or packages while a test method is running. Useful for performance or scalability tests + * where we don't want to generate a large log in a tight inner loop, or + * enabling debug logging for a test case. + * + * @author Artem Bilan + * + * @since 2.2 + * + */ +public final class Log4j2LevelAdjuster implements MethodRule { + + private static final Log logger = LogFactory.getLog(Log4j2LevelAdjuster.class); + + private final Class[] classes; + + private final Level level; + + private final String[] categories; + + private Log4j2LevelAdjuster(Level level) { + this(level, null, new String[] { "org.springframework.integration" }); + } + + private Log4j2LevelAdjuster(Level level, Class[] classes, String[] categories) { + Assert.notNull(level, "'level' must be null"); + this.level = level; + this.classes = classes != null ? classes : new Class[0]; + + Stream categoryStream = Stream.of(getClass().getPackage().getName()); + + if (!ObjectUtils.isEmpty(categories)) { + categoryStream = Stream.concat(Arrays.stream(categories), categoryStream); + } + + this.categories = categoryStream.toArray(String[]::new); + } + + @Override + public Statement apply(final Statement base, final FrameworkMethod method, Object target) { + return new Statement() { + + @Override + public void evaluate() throws Throwable { + LoggerContext ctx = (LoggerContext) LogManager.getContext(false); + Configuration config = ctx.getConfiguration(); + + Map, Level> classLevels = new HashMap<>(); + for (Class cls : Log4j2LevelAdjuster.this.classes) { + String className = cls.getName(); + LoggerConfig loggerConfig = config.getLoggerConfig(className); + LoggerConfig specificConfig = loggerConfig; + + // We need a specific configuration for this logger, + // otherwise we would change the level of all other loggers + // having the original configuration as parent as well + + if (!loggerConfig.getName().equals(className)) { + specificConfig = new LoggerConfig(className, Log4j2LevelAdjuster.this.level, true); + specificConfig.setParent(loggerConfig); + config.addLogger(className, specificConfig); + } + + classLevels.put(cls, specificConfig.getLevel()); + specificConfig.setLevel(Log4j2LevelAdjuster.this.level); + } + + Map categoryLevels = new HashMap<>(); + for (String category : Log4j2LevelAdjuster.this.categories) { + LoggerConfig loggerConfig = config.getLoggerConfig(category); + LoggerConfig specificConfig = loggerConfig; + + // We need a specific configuration for this logger, + // otherwise we would change the level of all other loggers + // having the original configuration as parent as well + + if (!loggerConfig.getName().equals(category)) { + specificConfig = new LoggerConfig(category, Log4j2LevelAdjuster.this.level, true); + specificConfig.setParent(loggerConfig); + config.addLogger(category, specificConfig); + } + + categoryLevels.put(category, specificConfig.getLevel()); + specificConfig.setLevel(Log4j2LevelAdjuster.this.level); + } + + ctx.updateLoggers(); + + logger.debug("++++++++++++++++++++++++++++ " + + "Overridden log level setting for: " + Arrays.toString(Log4j2LevelAdjuster.this.classes) + + " and " + Arrays.toString(Log4j2LevelAdjuster.this.categories) + + " for test " + method.getName()); + + try { + base.evaluate(); + } + finally { + logger.debug("++++++++++++++++++++++++++++ " + + "Restoring log level setting for: " + Arrays.toString(Log4j2LevelAdjuster.this.classes) + + " and " + Arrays.toString(Log4j2LevelAdjuster.this.categories) + + " for test " + method.getName()); + + for (Class cls : Log4j2LevelAdjuster.this.classes) { + LoggerConfig loggerConfig = config.getLoggerConfig(cls.getName()); + loggerConfig.setLevel(classLevels.get(cls)); + } + + for (String category : Log4j2LevelAdjuster.this.categories) { + LoggerConfig loggerConfig = config.getLoggerConfig(category); + loggerConfig.setLevel(categoryLevels.get(category)); + } + + ctx.updateLoggers(); + } + } + + }; + } + + /** + * Specify the classes for logging level adjusting configured before. + * A new copy Log4j2LevelAdjuster instance is produced by this method. + * The provided classes parameter overrides existing value in the {@link #classes}. + * @param classes the classes to use for logging level adjusting + * @return a Log4j2LevelAdjuster copy with the provided classes + */ + public Log4j2LevelAdjuster classes(Class... classes) { + return classes(false, classes); + } + + /** + * Specify the classes for logging level adjusting configured before. + * A new copy Log4j2LevelAdjuster instance is produced by this method. + * The provided classes parameter can be merged with existing value in the {@link #classes}. + * @param merge to merge or not with previously configured {@link #classes} + * @param classes the classes to use for logging level adjusting + * @return a Log4j2LevelAdjuster copy with the provided classes + * @since 5.0.2 + */ + public Log4j2LevelAdjuster classes(boolean merge, Class... classes) { + return new Log4j2LevelAdjuster(this.level, + merge ? Stream.of(this.classes, classes).flatMap(Stream::of).toArray(Class[]::new) : classes, + this.categories); + } + + /** + * Specify the categories for logging level adjusting configured before. + * A new copy Log4j2LevelAdjuster instance is produced by this method. + * The provided categories parameter overrides existing value in the {@link #categories}. + * @param categories the categories to use for logging level adjusting + * @return a Log4j2LevelAdjuster copy with the provided categories + */ + public Log4j2LevelAdjuster categories(String... categories) { + return categories(false, categories); + } + + /** + * Specify the categories for logging level adjusting configured before. + * A new copy Log4j2LevelAdjuster instance is produced by this method. + * The provided categories parameter can be merged with existing value in the {@link #categories}. + * @param merge to merge or not with previously configured {@link #categories} + * @param categories the categories to use for logging level adjusting + * @return a Log4j2LevelAdjuster copy with the provided categories + * @since 5.0.2 + */ + public Log4j2LevelAdjuster categories(boolean merge, String... categories) { + return new Log4j2LevelAdjuster(this.level, this.classes, + merge ? Stream.of(this.categories, categories).flatMap(Stream::of).toArray(String[]::new) : categories); + } + + /** + * The factory to produce Log4j2LevelAdjuster instances for {@link Level#TRACE} logging + * with the {@code org.springframework.integration} as default category. + * @return the Log4j2LevelAdjuster instance + */ + public static Log4j2LevelAdjuster trace() { + return forLevel(Level.TRACE); + } + + /** + * The factory to produce Log4j2LevelAdjuster instances for {@link Level#DEBUG} logging + * with the {@code org.springframework.integration} as default category. + * @return the Log4j2LevelAdjuster instance + */ + public static Log4j2LevelAdjuster debug() { + return forLevel(Level.DEBUG); + } + + /** + * The factory to produce Log4j2LevelAdjuster instances for {@link Level#INFO} logging + * with the {@code org.springframework.integration} as default category. + * @return the Log4j2LevelAdjuster instance + */ + public static Log4j2LevelAdjuster info() { + return forLevel(Level.INFO); + } + + /** + * The factory to produce Log4j2LevelAdjuster instances for arbitrary logging {@link Level} + * with the {@code org.springframework.integration} as default category. + * @param level the {@link Level} to use for logging + * @return the Log4j2LevelAdjuster instance + */ + public static Log4j2LevelAdjuster forLevel(Level level) { + return new Log4j2LevelAdjuster(level); + } + +} diff --git a/spring-kafka-test/src/main/java/org/springframework/kafka/test/rule/Log4jLevelAdjuster.java b/spring-kafka-test/src/main/java/org/springframework/kafka/test/rule/Log4jLevelAdjuster.java deleted file mode 100644 index d99bf736..00000000 --- a/spring-kafka-test/src/main/java/org/springframework/kafka/test/rule/Log4jLevelAdjuster.java +++ /dev/null @@ -1,108 +0,0 @@ -/* - * Copyright 2017 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 - * - * http://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.kafka.test.rule; - -import java.util.ArrayList; -import java.util.Arrays; -import java.util.HashMap; -import java.util.LinkedHashSet; -import java.util.Map; -import java.util.Set; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.apache.log4j.Level; -import org.apache.log4j.LogManager; -import org.junit.rules.MethodRule; -import org.junit.runners.model.FrameworkMethod; -import org.junit.runners.model.Statement; - -/** - * A JUnit method @Rule that changes the logger level for a set of classes - * or packages - * while a test method is running. Useful for performance or scalability tests - * where we don't want to generate a large log in a tight inner loop, or - * enabling debug logging for a test case. - * - * @author Dave Syer - * @author Gary Russell - * - * @since 2.0 - */ -public class Log4jLevelAdjuster implements MethodRule { - - private static final Log logger = LogFactory.getLog(Log4jLevelAdjuster.class); - - private final Class[] classes; - - private final Level level; - - private final String[] categories; - - public Log4jLevelAdjuster(Level level, Class... classes) { - this.level = level; - this.classes = classes; - this.categories = new String[0]; - } - - public Log4jLevelAdjuster(Level level, String... categories) { - this.level = level; - this.classes = new Class[0]; - Set cats = new LinkedHashSet(Arrays.asList(categories)); - cats.add(getClass().getPackage().getName()); - this.categories = new ArrayList(cats).toArray(new String[cats.size()]); - } - - @Override - public Statement apply(final Statement base, final FrameworkMethod method, Object target) { - return new Statement() { - - @Override - public void evaluate() throws Throwable { - Map, Level> oldLevels = new HashMap, Level>(); - for (Class cls : classes) { - oldLevels.put(cls, LogManager.getLogger(cls).getEffectiveLevel()); - LogManager.getLogger(cls).setLevel(level); - } - Map oldCatLevels = new HashMap(); - for (String category : categories) { - oldCatLevels.put(category, LogManager.getLogger(category).getEffectiveLevel()); - LogManager.getLogger(category).setLevel(level); - } - logger.debug("++++++++++++++++++++++++++++ " - + "Overridden log level setting for: " + Arrays.asList(classes) + " and " - + Arrays.asList(categories) + " for test " + method.getName()); - try { - base.evaluate(); - } - finally { - logger.debug("++++++++++++++++++++++++++++ " - + "Restoring log level setting for: " + Arrays.asList(classes) + " and " - + Arrays.asList(categories) + " for test " + method.getName()); - // raw Class type used to avoid http://bugs.sun.com/view_bug.do?bug_id=6682380 - for (Class cls : classes) { - LogManager.getLogger(cls).setLevel(oldLevels.get(cls)); - } - for (String category : categories) { - LogManager.getLogger(category).setLevel(oldCatLevels.get(category)); - } - } - } - }; - } - -} diff --git a/spring-kafka-test/src/test/resources/log4j2-test.xml b/spring-kafka-test/src/test/resources/log4j2-test.xml new file mode 100644 index 00000000..f50d0f7d --- /dev/null +++ b/spring-kafka-test/src/test/resources/log4j2-test.xml @@ -0,0 +1,23 @@ + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-kafka/src/test/java/org/springframework/kafka/core/StreamsBuilderFactoryBeanTests.java b/spring-kafka/src/test/java/org/springframework/kafka/core/StreamsBuilderFactoryBeanTests.java index 2c3887a8..2743e107 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/core/StreamsBuilderFactoryBeanTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/core/StreamsBuilderFactoryBeanTests.java @@ -28,12 +28,13 @@ import java.util.Map; import org.apache.kafka.streams.StreamsConfig; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.DisabledOnOs; +import org.junit.jupiter.api.condition.OS; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -import org.springframework.kafka.annotation.EnableKafka; import org.springframework.kafka.annotation.EnableKafkaStreams; import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration; import org.springframework.kafka.test.context.EmbeddedKafka; @@ -49,6 +50,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; @SpringJUnitConfig @DirtiesContext @EmbeddedKafka +@DisabledOnOs(OS.WINDOWS) public class StreamsBuilderFactoryBeanTests { private static final String APPLICATION_ID = "testCleanupStreams"; @@ -77,7 +79,6 @@ public class StreamsBuilderFactoryBeanTests { } @Configuration - @EnableKafka @EnableKafkaStreams public static class KafkaStreamsConfiguration { diff --git a/spring-kafka/src/test/java/org/springframework/kafka/kstream/KafkaStreamsBranchTests.java b/spring-kafka/src/test/java/org/springframework/kafka/kstream/KafkaStreamsBranchTests.java index 7f476d52..6b445a40 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/kstream/KafkaStreamsBranchTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/kstream/KafkaStreamsBranchTests.java @@ -56,6 +56,7 @@ import org.springframework.test.context.junit4.SpringRunner; /** * @author Elliot Kennedy * @author Artem Bilan + * * @since 1.3.3 */ @RunWith(SpringRunner.class) @@ -102,6 +103,9 @@ public class KafkaStreamsBranchTests { assertThat(trueValues).containsExactly("true", "true"); assertThat(falseValues).containsExactly("false"); + + falseConsumer.close(); + trueConsumer.close(); } private Consumer createConsumer() { diff --git a/spring-kafka/src/test/java/org/springframework/kafka/kstream/KafkaStreamsJsonSerializationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/kstream/KafkaStreamsJsonSerializationTests.java index d82dd267..8099a054 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/kstream/KafkaStreamsJsonSerializationTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/kstream/KafkaStreamsJsonSerializationTests.java @@ -34,6 +34,7 @@ import org.apache.kafka.streams.StreamsConfig; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.Printed; import org.apache.kafka.streams.kstream.Produced; +import org.junit.After; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; @@ -61,20 +62,24 @@ import com.fasterxml.jackson.annotation.JsonProperty; /** * @author Elliot Kennedy + * @author Artem Bilan */ @RunWith(SpringRunner.class) @DirtiesContext @EmbeddedKafka(partitions = 1, topics = { - KafkaStreamsJsonSerializationTests.OBJECT_INPUT_TOPIC, - KafkaStreamsJsonSerializationTests.OBJECT_OUTPUT_TOPIC + KafkaStreamsJsonSerializationTests.OBJECT_INPUT_TOPIC, + KafkaStreamsJsonSerializationTests.OBJECT_OUTPUT_TOPIC }) public class KafkaStreamsJsonSerializationTests { public static final String OBJECT_INPUT_TOPIC = "object-input-topic"; + public static final String OBJECT_OUTPUT_TOPIC = "object-output-topic"; - public static final JsonSerde jsonObjectKeySerde = new JsonSerde<>(JsonObjectKey.class).setUseTypeMapperForKey(true); + public static final JsonSerde jsonObjectKeySerde = + new JsonSerde<>(JsonObjectKey.class).setUseTypeMapperForKey(true); + public static final JsonSerde jsonObjectValueSerde = new JsonSerde<>(JsonObjectValue.class); @Autowired @@ -87,14 +92,22 @@ public class KafkaStreamsJsonSerializationTests { @Before public void setup() throws Exception { - objectOutputTopicConsumer = consumer(OBJECT_OUTPUT_TOPIC, jsonObjectKeySerde, jsonObjectValueSerde); + this.objectOutputTopicConsumer = consumer(OBJECT_OUTPUT_TOPIC, jsonObjectKeySerde, jsonObjectValueSerde); + } + + @After + public void teardown() { + if (this.objectOutputTopicConsumer != null) { + this.objectOutputTopicConsumer.close(); + } } @Test - public void testJsonObjectSerialization() throws Exception { + public void testJsonObjectSerialization() { template.send(OBJECT_INPUT_TOPIC, new JsonObjectKey(25), new JsonObjectValue("twenty-five")); - ConsumerRecords outputTopicRecords = KafkaTestUtils.getRecords(objectOutputTopicConsumer); + ConsumerRecords outputTopicRecords = + KafkaTestUtils.getRecords(this.objectOutputTopicConsumer); assertThat(outputTopicRecords.count()).isEqualTo(1); ConsumerRecord output = outputTopicRecords.iterator().next(); @@ -112,7 +125,7 @@ public class KafkaStreamsJsonSerializationTests { DefaultKafkaConsumerFactory kafkaConsumerFactory = new DefaultKafkaConsumerFactory<>(consumerProps, keySerde.deserializer(), valueSerde.deserializer()); Consumer consumer = kafkaConsumerFactory.createConsumer(); - kafkaEmbedded.consumeFromAnEmbeddedTopic(consumer, topic); + this.kafkaEmbedded.consumeFromAnEmbeddedTopic(consumer, topic); return consumer; } @@ -135,6 +148,7 @@ public class KafkaStreamsJsonSerializationTests { "key=" + key + '}'; } + } public static class JsonObjectValue { @@ -160,6 +174,7 @@ public class KafkaStreamsJsonSerializationTests { "value='" + value + '\'' + '}'; } + } @Configuration @@ -205,6 +220,7 @@ public class KafkaStreamsJsonSerializationTests { return testStream; } + } } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/kstream/KafkaStreamsTests.java b/spring-kafka/src/test/java/org/springframework/kafka/kstream/KafkaStreamsTests.java index afe5057b..cd2b5b98 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/kstream/KafkaStreamsTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/kstream/KafkaStreamsTests.java @@ -37,6 +37,7 @@ import org.apache.kafka.streams.kstream.Materialized; import org.apache.kafka.streams.kstream.Printed; import org.apache.kafka.streams.kstream.Produced; import org.apache.kafka.streams.kstream.TimeWindows; +import org.apache.kafka.streams.kstream.ValueMapper; import org.apache.kafka.streams.processor.WallclockTimestampExtractor; import org.apache.kafka.streams.processor.internals.StreamThread; import org.junit.Test; @@ -193,8 +194,8 @@ public class KafkaStreamsTests { @Bean public KStream kStream(StreamsBuilder kStreamBuilder) { KStream stream = kStreamBuilder.stream(STREAMING_TOPIC1); - stream.mapValues(String::toUpperCase) - .mapValues(Foo::new) + stream.mapValues((ValueMapper) String::toUpperCase) + .mapValues((ValueMapper) Foo::new) .through(FOOS, Produced.with(Serdes.Integer(), new JsonSerde() { })) diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java index 661f759f..c024e6d4 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java @@ -356,6 +356,7 @@ public class TransactionalContainerTests { assertThat(consumer.position(new TopicPartition(topic1, 0))).isEqualTo(1); logger.info("Stop testRollbackRecord"); pf.destroy(); + consumer.close(); } @SuppressWarnings("serial") diff --git a/spring-kafka/src/test/java/org/springframework/kafka/security/jaas/KafkaJaasLoginModuleInitializerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/security/jaas/KafkaJaasLoginModuleInitializerTests.java index 942302de..34487699 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/security/jaas/KafkaJaasLoginModuleInitializerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/security/jaas/KafkaJaasLoginModuleInitializerTests.java @@ -62,7 +62,7 @@ public class KafkaJaasLoginModuleInitializerTests { assertThat(kafkaConfiguration).hasSize(1); assertThat(kafkaConfiguration[0].getOptions()).isEqualTo(kafkaConfigurationArray[0].getOptions()); - JaasContext context = JaasContext.load(JaasContext.Type.CLIENT, null, Collections.emptyMap()); + JaasContext context = JaasContext.loadClientContext(Collections.emptyMap()); List appConfigurationEntries = context.configurationEntries(); assertThat(appConfigurationEntries).hasSize(1); diff --git a/spring-kafka/src/test/resources/log4j.properties b/spring-kafka/src/test/resources/log4j.properties deleted file mode 100644 index 56399de4..00000000 --- a/spring-kafka/src/test/resources/log4j.properties +++ /dev/null @@ -1,11 +0,0 @@ -log4j.rootCategory=WARN, stdout - -log4j.appender.stdout=org.apache.log4j.ConsoleAppender -log4j.appender.stdout.layout=org.apache.log4j.PatternLayout -log4j.appender.stdout.layout.ConversionPattern=%d{HH:mm:ss.SSS} %-5p [%t][%c] %m%n -log4j.category.org.springframework.kafka=WARN -log4j.category.org.springframework.kafka.ReplyingKafkaTemplate=WARN -log4j.category.org.apache.kafka.clients=WARN -log4j.category.org.apache.kafka.common.network.Selector=ERROR -log4j.category.kafka.server.ReplicaFetcherThread=ERROR -log4j.category.org.apache.kafka.clients.producer.internals.TransactionManager=WARN diff --git a/spring-kafka/src/test/resources/log4j2-test.xml b/spring-kafka/src/test/resources/log4j2-test.xml new file mode 100644 index 00000000..9f33afab --- /dev/null +++ b/spring-kafka/src/test/resources/log4j2-test.xml @@ -0,0 +1,24 @@ + + + + + + + + + + + + + + + + + + + + + + + +