Upgrade to kafka 2.5.1

Author: garyrussell

build.gradle

@@ -56,7 +56,7 @@ ext {
 	jaywayJsonPathVersion = '2.4.0'
 	junit4Version = '4.13'
 	junitJupiterVersion = '5.6.2'
-	kafkaVersion = '2.5.0'
+	kafkaVersion = '2.5.1'
 	log4jVersion = '2.13.2'
 	micrometerVersion = '1.5.2'
 	mockitoVersion = '3.3.3'

spring-kafka/src/test/java/org/springframework/kafka/config/StreamsBuilderFactoryBeanTests.java

@@ -26,9 +26,13 @@
 import java.nio.file.Paths;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.regex.Pattern;
 
 import org.apache.kafka.streams.StreamsBuilder;
 import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.KTable;
+import org.apache.kafka.streams.kstream.Materialized;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.condition.DisabledOnOs;
@@ -76,7 +80,6 @@ public static void setup() throws IOException {
 
 	@Test
 	public void testCleanupStreams() throws IOException {
-		Files.createDirectory(Paths.get(stateStoreDir.toString(), APPLICATION_ID));
 		Path stateStore = Files.createDirectory(Paths.get(stateStoreDir.toString(), APPLICATION_ID, "0_0"));
 		assertThat(stateStore).exists();
 		streamsBuilderFactoryBean.stop();
@@ -97,6 +100,8 @@ protected StreamsBuilder createInstance() {
 			}
 		};
 		streamsBuilderFactoryBean.afterPropertiesSet();
+		StreamsBuilder builder = streamsBuilderFactoryBean.getObject();
+		builder.stream(Pattern.compile("foo"));
 		streamsBuilderFactoryBean.start();
 		StreamsBuilder streamsBuilder = streamsBuilderFactoryBean.getObject();
 		verify(streamsBuilder).build(kafkaStreamsConfiguration.asProperties());
@@ -125,6 +130,13 @@ public KafkaStreamsConfiguration kStreamsConfigs() {
 			return new KafkaStreamsConfiguration(props);
 		}
 
+		@Bean
+		public KTable<?, ?> table(StreamsBuilder builder) {
+			KStream<Object, Object> stream = builder.stream(Pattern.compile("foo"));
+			return stream.groupByKey()
+					.count(Materialized.as("store"));
+
+		}
 	}
 
 }

spring-kafka/src/test/java/org/springframework/kafka/config/StreamsBuilderFactoryLateConfigTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2020 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.
@@ -21,6 +21,7 @@
 import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
 
 import java.util.Properties;
+import java.util.regex.Pattern;
 
 import org.apache.kafka.streams.StreamsConfig;
 import org.junit.jupiter.api.Test;
@@ -70,11 +71,12 @@ public void testStreamBuilderFactoryCannotBeInstantiatedWhenAutoStart() {
 	}
 
 	@Test
-	public void testStreamsBuilderFactoryWithConfigProvidedLater() {
+	public void testStreamsBuilderFactoryWithConfigProvidedLater() throws Exception {
 		Properties props = new Properties();
 		props.put(StreamsConfig.APPLICATION_ID_CONFIG, APPLICATION_ID);
 		props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, this.brokerAddresses);
 		streamsBuilderFactoryBean.setStreamsConfiguration(props);
+		streamsBuilderFactoryBean.getObject().stream(Pattern.compile("foo"));
 
 		assertThat(streamsBuilderFactoryBean.isRunning()).isFalse();
 		streamsBuilderFactoryBean.start();