Upgrade to 2.4 clients; bump to 2.4.0.B-S

- Remove references to ZkClient
- deprecation of `Consumer.committed(TopicPartition tp)`
- Addition of `onPartitionsLost` to `ConsumerRebalanceListener`
- since `onPartitionsLost` calls `onPartitionsRevoked` by default with a
  possible partial list, remove the revoked partitions from the assignments
- fix some reflection uses in tests (`groupId` field and `NewTopic`)
- fix `EKIT` `listen12` listener - we now always get the full batch
- `ConsumerRebalanceListener.onPartitionsRevoked` is no longer called with an
  empty collection
- deprecation of some `TopologyDriver` helper classes
- update what's new and change log appendix
- add `unregisterSeekCallback` to `ConsumerSeekAware`
  - `AbstractCSA` now removes TL there since we may have only a partial revoke

* Fix what's new

* GH-1277: isAckAfterHandle default true

Resolves: #1277

* Fix SBFB Javadocs

* Update version

* Fix test for isAckAfterHandle default true

* Upgrade to 2.4.0 Clients

* Remove mavenLocal() repo

Author: garyrussell

build.gradle

@@ -45,7 +45,7 @@ ext {
 	jaywayJsonPathVersion = '2.4.0'
 	junit4Version = '4.12'
 	junitJupiterVersion = '5.5.2'
-	kafkaVersion = '2.3.1'
+	kafkaVersion = '2.4.0'
 	log4jVersion = '2.12.1'
 	micrometerVersion = '1.3.2'
 	mockitoVersion = '3.0.0'

gradle.properties

@@ -1 +1 @@
-version=2.3.5.BUILD-SNAPSHOT
+version=2.4.0.BUILD-SNAPSHOT

spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java

@@ -34,8 +34,6 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 
-import org.I0Itec.zkclient.ZkClient;
-import org.I0Itec.zkclient.exception.ZkInterruptedException;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.AdminClientConfig;
@@ -66,7 +64,6 @@
 import kafka.server.NotRunning;
 import kafka.utils.CoreUtils;
 import kafka.utils.TestUtils;
-import kafka.utils.ZKStringSerializer$;
 import kafka.zk.ZkFourLetterWords;
 import kafka.zookeeper.ZooKeeperClient;
 
@@ -133,8 +130,6 @@ public class EmbeddedKafkaBroker implements InitializingBean, DisposableBean {
 
 	private volatile ZooKeeperClient zooKeeperClient;
 
-	private volatile ZkClient zkClient;
-
 	public EmbeddedKafkaBroker(int count) {
 		this(count, false);
 	}
@@ -298,7 +293,8 @@ private Properties createBrokerProperties(int i) {
 				scala.Option.apply(null),
 				scala.Option.apply(null),
 				scala.Option.apply(null),
-				true, false, 0, false, 0, false, 0, scala.Option.apply(null), 1, false);
+				true, false, 0, false, 0, false, 0, scala.Option.apply(null), 1, false,
+				this.partitionsPerTopic, (short) this.count);
 	}
 
 	/**
@@ -395,19 +391,11 @@ public void destroy() {
 				// do nothing
 			}
 		}
-		try {
-			synchronized (this) {
-				if (this.zooKeeperClient != null) {
-					this.zooKeeperClient.close();
-				}
-				if (this.zkClient != null) {
-					this.zkClient.close();
-				}
+		synchronized (this) {
+			if (this.zooKeeperClient != null) {
+				this.zooKeeperClient.close();
 			}
 		}
-		catch (ZkInterruptedException e) {
-			// do nothing
-		}
 		try {
 			this.zookeeper.shutdown();
 			this.zkConnect = null;
@@ -433,20 +421,6 @@ public EmbeddedZookeeper getZookeeper() {
 		return this.zookeeper;
 	}
 
-	/**
-	 * Return the ZkClient.
-	 * @return the client.
-	 * @deprecated in favor of {@link #getZooKeeperClient()}.
-	 */
-	@Deprecated
-	public synchronized ZkClient getZkClient() {
-		if (this.zkClient == null) {
-			this.zkClient = new ZkClient(this.zkConnect, ZK_SESSION_TIMEOUT, ZK_CONNECTION_TIMEOUT,
-					ZKStringSerializer$.MODULE$);
-		}
-		return this.zkClient;
-	}
-
 	/**
 	 * Return the ZooKeeperClient.
 	 * @return the client.

spring-kafka/src/main/java/org/springframework/kafka/config/StreamsBuilderFactoryBean.java

@@ -126,7 +126,7 @@ public StreamsBuilderFactoryBean(KafkaStreamsConfiguration streamsConfig) {
 	}
 
 	/**
-	 * Set {@link StreamsConfig} on this factory.
+	 * Set the streams configuration {@link Properties} on this factory.
 	 * @param streamsConfig the streams configuration.
 	 * @since 2.2
 	 */

spring-kafka/src/main/java/org/springframework/kafka/core/reactive/ReactiveKafkaConsumerTemplate.java

@@ -146,10 +146,22 @@ public Mono<Long> position(TopicPartition partition) {
 		return doOnConsumer(consumer -> consumer.position(partition));
 	}
 
+	/**
+	 * Get the committed {@link OffsetAndMetadata} for the partition.
+	 * @param partition the partition.
+	 * @return the {@link OffsetAndMetadata}.
+	 * @deprecated in favor of {@link #committed(Set)}.
+	 */
+	@Deprecated
+	@SuppressWarnings("deprecation")
 	public Mono<OffsetAndMetadata> committed(TopicPartition partition) {
 		return doOnConsumer(consumer -> consumer.committed(partition));
 	}
 
+	public Mono<Map<TopicPartition, OffsetAndMetadata>> committed(Set<TopicPartition> partitions) {
+		return doOnConsumer(consumer -> consumer.committed(partitions));
+	}
+
 	public Flux<PartitionInfo> partitionsFromConsumerFor(String topic) {
 		Mono<List<PartitionInfo>> partitions = doOnConsumer(c -> c.partitionsFor(topic));
 		return partitions.flatMapIterable(Function.identity());

spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractConsumerSeekAware.java

@@ -56,6 +56,10 @@ public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, Consumer
 	@Override
 	public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
 		partitions.forEach(tp -> this.callbacks.remove(tp));
+	}
+
+	@Override
+	public void unregisterSeekCallback() {
 		this.callbackForThread.remove();
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java

@@ -465,6 +465,12 @@ public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
 						getGroupId() + ": partitions assigned: " + partitions);
 			}
 
+			@Override
+			public void onPartitionsLost(Collection<TopicPartition> partitions) {
+				AbstractMessageListenerContainer.this.logger.info(() ->
+				getGroupId() + ": partitions lost: " + partitions);
+			}
+
 		};
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareRebalanceListener.java

@@ -53,6 +53,16 @@ default void onPartitionsRevokedAfterCommit(Consumer<?, ?> consumer, Collection<
 		// do nothing
 	}
 
+	/**
+	 * The same as {@link #onPartitionsLost(Collection)} with an additional consumer parameter.
+	 * @param consumer the consumer.
+	 * @param partitions the partitions.
+	 * @since 2.4
+	 */
+	default void onPartitionsLost(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
+		// do nothing
+	}
+
 	/**
 	 * The same as {@link #onPartitionsAssigned(Collection)} with the additional consumer
 	 * parameter.
@@ -73,4 +83,9 @@ default void onPartitionsAssigned(Collection<TopicPartition> partitions) {
 		throw new UnsupportedOperationException("Listener container should never call this");
 	}
 
+	@Override
+	default void onPartitionsLost(Collection<TopicPartition> partitions) {
+		throw new UnsupportedOperationException("Listener container should never call this");
+	}
+
 }

spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerSeekAware.java

@@ -72,6 +72,15 @@ default void onIdleContainer(Map<TopicPartition, Long> assignments, ConsumerSeek
 		// do nothing
 	}
 
+	/**
+	 * Called when the listener consumer terminates allowing implementations to clean up
+	 * state, such as thread locals.
+	 * @since 2.4
+	 */
+	default void unregisterSeekCallback() {
+		// do nothing
+	}
+
 	/**
 	 * A callback that a listener can invoke to seek to a specific offset.
 	 */

spring-kafka/src/main/java/org/springframework/kafka/listener/GenericErrorHandler.java

@@ -63,8 +63,7 @@ default void clearThreadState() {
 	 * @since 2.3.2
 	 */
 	default boolean isAckAfterHandle() {
-		// TODO: Default true in the next release.
-		return false;
+		return true;
 	}
 
 }