Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-640: Upgrade kafka version to 1.1.0

Browse Source 
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
This commit is contained in:
Gary Russell
2018-04-03 17:09:10 -04:00
committed by Artem Bilan
parent b048aaa8f0
commit 9f724ea27a
13 changed files with 357 additions and 168 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
build.gradle
View File

@@ -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)
}
}

37
spring-kafka-test/src/main/java/org/springframework/kafka/test/rule/KafkaEmbedded.java
View File

@@ -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 {

246
spring-kafka-test/src/main/java/org/springframework/kafka/test/rule/Log4j2LevelAdjuster.java Normal file
View File

@@ -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<String> 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<Class<?>, 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<String, Level> 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);
}
}

108
spring-kafka-test/src/main/java/org/springframework/kafka/test/rule/Log4jLevelAdjuster.java
View File

@@ -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 &#064;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<String> cats = new LinkedHashSet<String>(Arrays.asList(categories));
cats.add(getClass().getPackage().getName());
this.categories = new ArrayList<String>(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<Class<?>, Level> oldLevels = new HashMap<Class<?>, Level>();
for (Class<?> cls : classes) {
oldLevels.put(cls, LogManager.getLogger(cls).getEffectiveLevel());
LogManager.getLogger(cls).setLevel(level);
}
Map<String, Level> oldCatLevels = new HashMap<String, Level>();
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));
}
}
}
};
}
}

23
spring-kafka-test/src/test/resources/log4j2-test.xml Normal file
View File

@@ -0,0 +1,23 @@
<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="WARN">
<Appenders>
<Console name="STDOUT" target="SYSTEM_OUT">
<PatternLayout pattern="%d %p [%t] [%c] - %m%n" />
</Console>
</Appenders>
<Loggers>
<Logger name="org.springframework.kafka.test" level="warn"/>
<Logger name="org.apache.kafka.clients" level="warn"/>
<Logger name="org.apache.kafka.clients.NetworkClient" level="error"/>
<Logger name="org.apache.kafka.common.network.Selector" level="error"/>
<Logger name="kafka.server.ReplicaFetcherThread" level="error"/>
<Logger name="kafka.server.LogDirFailureChannel" level="fatal"/>
<Logger name="kafka.server.BrokerMetadataCheckpoint" level="error"/>
<Logger name="kafka.utils.CoreUtils$" level="error"/>
<Logger name="org.apache.kafka.clients.producer.internals.TransactionManager" level="warn"/>
<Logger name="org.apache.zookeeper.server.ZooKeeperServer" level="fatal"/>
<Root level="warn">
<AppenderRef ref="STDOUT" />
</Root>
</Loggers>
</Configuration>

5
spring-kafka/src/test/java/org/springframework/kafka/core/StreamsBuilderFactoryBeanTests.java
View File

@@ -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 {

4
spring-kafka/src/test/java/org/springframework/kafka/kstream/KafkaStreamsBranchTests.java
View File

@@ -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<String, String> createConsumer() {

30
spring-kafka/src/test/java/org/springframework/kafka/kstream/KafkaStreamsJsonSerializationTests.java
View File

@@ -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<JsonObjectKey> jsonObjectKeySerde = new JsonSerde<>(JsonObjectKey.class).setUseTypeMapperForKey(true);
public static final JsonSerde<JsonObjectKey> jsonObjectKeySerde =
new JsonSerde<>(JsonObjectKey.class).setUseTypeMapperForKey(true);
public static final JsonSerde<JsonObjectValue> 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<JsonObjectKey, JsonObjectValue> outputTopicRecords = KafkaTestUtils.getRecords(objectOutputTopicConsumer);
ConsumerRecords<JsonObjectKey, JsonObjectValue> outputTopicRecords =
KafkaTestUtils.getRecords(this.objectOutputTopicConsumer);
assertThat(outputTopicRecords.count()).isEqualTo(1);
ConsumerRecord<JsonObjectKey, JsonObjectValue> output = outputTopicRecords.iterator().next();
@@ -112,7 +125,7 @@ public class KafkaStreamsJsonSerializationTests {
DefaultKafkaConsumerFactory<K, V> kafkaConsumerFactory =
new DefaultKafkaConsumerFactory<>(consumerProps, keySerde.deserializer(), valueSerde.deserializer());
Consumer<K, V> 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;
}
}
}

5
spring-kafka/src/test/java/org/springframework/kafka/kstream/KafkaStreamsTests.java
View File

@@ -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<Integer, String> kStream(StreamsBuilder kStreamBuilder) {
KStream<Integer, String> stream = kStreamBuilder.stream(STREAMING_TOPIC1);
stream.mapValues(String::toUpperCase)
.mapValues(Foo::new)
stream.mapValues((ValueMapper<String, String>) String::toUpperCase)
.mapValues((ValueMapper<String, Foo>) Foo::new)
.through(FOOS, Produced.with(Serdes.Integer(), new JsonSerde<Foo>() {
}))

1
spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java
View File

@@ -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")

2
spring-kafka/src/test/java/org/springframework/kafka/security/jaas/KafkaJaasLoginModuleInitializerTests.java
View File

@@ -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<AppConfigurationEntry> appConfigurationEntries = context.configurationEntries();
assertThat(appConfigurationEntries).hasSize(1);

11
spring-kafka/src/test/resources/log4j.properties
View File

@@ -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

24
spring-kafka/src/test/resources/log4j2-test.xml Normal file
View File

@@ -0,0 +1,24 @@
<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="WARN">
<Appenders>
<Console name="STDOUT" target="SYSTEM_OUT">
<PatternLayout pattern="%d %p [%t] [%c] - %m%n" />
</Console>
</Appenders>
<Loggers>
<Logger name="org.springframework.kafka" level="warn"/>
<Logger name="org.springframework.kafka.ReplyingKafkaTemplate" level="warn"/>
<Logger name="org.apache.kafka.clients" level="warn"/>
<Logger name="org.apache.kafka.clients.NetworkClient" level="error"/>
<Logger name="org.apache.kafka.common.network.Selector" level="error"/>
<Logger name="kafka.server.ReplicaFetcherThread" level="error"/>
<Logger name="kafka.server.LogDirFailureChannel" level="fatal"/>
<Logger name="kafka.server.BrokerMetadataCheckpoint" level="error"/>
<Logger name="kafka.utils.CoreUtils$" level="error"/>
<Logger name="org.apache.kafka.clients.producer.internals.TransactionManager" level="warn"/>
<Logger name="org.apache.zookeeper.server.ZooKeeperServer" level="fatal"/>
<Root level="warn">
<AppenderRef ref="STDOUT" />
</Root>
</Loggers>
</Configuration>