Bump kafka-clients to 2.2.0

- fix deprecations

Author: garyrussell

build.gradle

@@ -78,7 +78,7 @@ subprojects { subproject ->
 		junit4Version = '4.12'
 		junitJupiterVersion = '5.4.0'
 		junitPlatformVersion = '1.4.0'
-		kafkaVersion = '2.1.1'
+		kafkaVersion = '2.2.0'
 		log4jVersion = '2.11.2'
 		mockitoVersion = '2.24.0'
 		scalaVersion = '2.11'

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

@@ -94,7 +94,7 @@ public class EmbeddedKafkaBroker implements InitializingBean, DisposableBean {
 	 */
 	public static final String BROKER_LIST_PROPERTY = "spring.embedded.kafka.brokers.property";
 
-	private static final int DEFAULT_ADMIN_TIMEOUT = 30;
+	private static final Duration DEFAULT_ADMIN_TIMEOUT = Duration.ofSeconds(10);
 
 	private final int count;
 
@@ -116,7 +116,7 @@ public class EmbeddedKafkaBroker implements InitializingBean, DisposableBean {
 
 	private int[] kafkaPorts;
 
-	private int adminTimeout = DEFAULT_ADMIN_TIMEOUT;
+	private Duration adminTimeout = DEFAULT_ADMIN_TIMEOUT;
 
 	private String brokerListProperty;
 
@@ -157,12 +157,12 @@ public EmbeddedKafkaBroker(int count, boolean controlledShutdown, int partitions
 	/**
 	 * Specify the properties to configure Kafka Broker before start, e.g.
 	 * {@code auto.create.topics.enable}, {@code transaction.state.log.replication.factor} etc.
-	 * @param brokerProperties the properties to use for configuring Kafka Broker(s).
+	 * @param properties the properties to use for configuring Kafka Broker(s).
 	 * @return this for chaining configuration.
 	 * @see KafkaConfig
 	 */
-	public EmbeddedKafkaBroker brokerProperties(Map<String, String> brokerProperties) {
-		this.brokerProperties.putAll(brokerProperties);
+	public EmbeddedKafkaBroker brokerProperties(Map<String, String> properties) {
+		this.brokerProperties.putAll(properties);
 		return this;
 	}
 
@@ -180,13 +180,13 @@ public EmbeddedKafkaBroker brokerProperty(String property, Object value) {
 	/**
 	 * Set explicit ports on which the kafka brokers will listen. Useful when running an
 	 * embedded broker that you want to access from other processes.
-	 * @param kafkaPorts the ports.
+	 * @param ports the ports.
 	 * @return the {@link EmbeddedKafkaBroker}.
 	 */
-	public EmbeddedKafkaBroker kafkaPorts(int... kafkaPorts) {
-		Assert.isTrue(kafkaPorts.length == this.count, "A port must be provided for each instance ["
-				+ this.count + "], provided: " + Arrays.toString(kafkaPorts) + ", use 0 for a random port");
-		this.kafkaPorts = kafkaPorts;
+	public EmbeddedKafkaBroker kafkaPorts(int... ports) {
+		Assert.isTrue(ports.length == this.count, "A port must be provided for each instance ["
+				+ this.count + "], provided: " + Arrays.toString(ports) + ", use 0 for a random port");
+		this.kafkaPorts = ports;
 		return this;
 	}
 
@@ -197,7 +197,7 @@ public EmbeddedKafkaBroker kafkaPorts(int... kafkaPorts) {
 	 * @since 2.2
 	 */
 	public void setAdminTimeout(int adminTimeout) {
-		this.adminTimeout = adminTimeout;
+		this.adminTimeout = Duration.ofSeconds(adminTimeout);
 	}
 
 	@Override
@@ -247,42 +247,42 @@ private Properties createBrokerProperties(int i) {
 	/**
 	 * Add topics to the existing broker(s) using the configured number of partitions.
 	 * The broker(s) must be running.
-	 * @param topics the topics.
+	 * @param topicsToAdd the topics.
 	 */
-	public void addTopics(String... topics) {
+	public void addTopics(String... topicsToAdd) {
 		Assert.notNull(this.zookeeper, "Broker must be started before this method can be called");
-		HashSet<String> set = new HashSet<>(Arrays.asList(topics));
+		HashSet<String> set = new HashSet<>(Arrays.asList(topicsToAdd));
 		createKafkaTopics(set);
 		this.topics.addAll(set);
 	}
 
 	/**
 	 * Add topics to the existing broker(s).
 	 * The broker(s) must be running.
-	 * @param topics the topics.
+	 * @param topicsToAdd the topics.
 	 * @since 2.2
 	 */
-	public void addTopics(NewTopic... topics) {
+	public void addTopics(NewTopic... topicsToAdd) {
 		Assert.notNull(this.zookeeper, "Broker must be started before this method can be called");
-		for (NewTopic topic : topics) {
+		for (NewTopic topic : topicsToAdd) {
 			Assert.isTrue(this.topics.add(topic.name()), () -> "topic already exists: " + topic);
 			Assert.isTrue(topic.replicationFactor() <= this.count
 							&& (topic.replicasAssignments() == null
 							|| topic.replicasAssignments().size() <= this.count),
 					() -> "Embedded kafka does not support the requested replication factor: " + topic);
 		}
 
-		doWithAdmin(admin -> createTopics(admin, Arrays.asList(topics)));
+		doWithAdmin(admin -> createTopics(admin, Arrays.asList(topicsToAdd)));
 	}
 
 	/**
 	 * Create topics in the existing broker(s) using the configured number of partitions.
-	 * @param topics the topics.
+	 * @param topicsToCreate the topics.
 	 */
-	private void createKafkaTopics(Set<String> topics) {
+	private void createKafkaTopics(Set<String> topicsToCreate) {
 		doWithAdmin(admin -> {
 			createTopics(admin,
-					topics.stream()
+					topicsToCreate.stream()
 							.map(t -> new NewTopic(t, this.partitionsPerTopic, (short) this.count))
 							.collect(Collectors.toList()));
 		});
@@ -291,7 +291,7 @@ private void createKafkaTopics(Set<String> topics) {
 	private void createTopics(AdminClient admin, List<NewTopic> newTopics) {
 		CreateTopicsResult createTopics = admin.createTopics(newTopics);
 		try {
-			createTopics.all().get(this.adminTimeout, TimeUnit.SECONDS);
+			createTopics.all().get(this.adminTimeout.getSeconds(), TimeUnit.SECONDS);
 		}
 		catch (Exception e) {
 			throw new KafkaException(e);
@@ -312,7 +312,7 @@ public void doWithAdmin(java.util.function.Consumer<AdminClient> callback) {
 		}
 		finally {
 			if (admin != null) {
-				admin.close(this.adminTimeout, TimeUnit.SECONDS);
+				admin.close(this.adminTimeout);
 			}
 		}
 	}
@@ -453,16 +453,16 @@ public void consumeFromAnEmbeddedTopic(Consumer<?, ?> consumer, String topic) {
 	/**
 	 * Subscribe a consumer to one or more of the embedded topics.
 	 * @param consumer the consumer.
-	 * @param topics the topics.
+	 * @param topicsToConsume the topics.
 	 */
-	public void consumeFromEmbeddedTopics(Consumer<?, ?> consumer, String... topics) {
-		HashSet<String> diff = new HashSet<>(Arrays.asList(topics));
+	public void consumeFromEmbeddedTopics(Consumer<?, ?> consumer, String... topicsToConsume) {
+		HashSet<String> diff = new HashSet<>(Arrays.asList(topicsToConsume));
 		diff.removeAll(new HashSet<>(this.topics));
 		assertThat(this.topics)
 				.as("topic(s):'" + diff + "' are not in embedded topic list")
-				.containsAll(new HashSet<>(Arrays.asList(topics)));
+				.containsAll(new HashSet<>(Arrays.asList(topicsToConsume)));
 		final AtomicBoolean assigned = new AtomicBoolean();
-		consumer.subscribe(Arrays.asList(topics), new ConsumerRebalanceListener() {
+		consumer.subscribe(Arrays.asList(topicsToConsume), new ConsumerRebalanceListener() {
 
 			@Override
 			public void onPartitionsRevoked(Collection<TopicPartition> partitions) {

spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java

@@ -16,6 +16,7 @@
 
 package org.springframework.kafka.core;
 
+import java.time.Duration;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Iterator;
@@ -85,7 +86,7 @@
 public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>, ApplicationContextAware,
 		ApplicationListener<ContextStoppedEvent>, DisposableBean {
 
-	private static final int DEFAULT_PHYSICAL_CLOSE_TIMEOUT = 30;
+	private static final Duration DEFAULT_PHYSICAL_CLOSE_TIMEOUT = Duration.ofSeconds(30);
 
 	private static final Log logger = LogFactory.getLog(DefaultKafkaProducerFactory.class); // NOSONAR
 
@@ -103,7 +104,7 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 
 	private Serializer<V> valueSerializer;
 
-	private int physicalCloseTimeout = DEFAULT_PHYSICAL_CLOSE_TIMEOUT;
+	private Duration physicalCloseTimeout = DEFAULT_PHYSICAL_CLOSE_TIMEOUT;
 
 	private String transactionIdPrefix;
 
@@ -153,7 +154,7 @@ public void setValueSerializer(@Nullable Serializer<V> valueSerializer) {
 	 * @since 1.0.7
 	 */
 	public void setPhysicalCloseTimeout(int physicalCloseTimeout) {
-		this.physicalCloseTimeout = physicalCloseTimeout;
+		this.physicalCloseTimeout = Duration.ofSeconds(physicalCloseTimeout);
 	}
 
 	/**
@@ -216,16 +217,16 @@ public boolean transactionCapable() {
 
 	@SuppressWarnings("resource")
 	@Override
-	public void destroy() throws Exception { //NOSONAR
+	public void destroy() {
 		CloseSafeProducer<K, V> producerToClose = this.producer;
 		this.producer = null;
 		if (producerToClose != null) {
-			producerToClose.delegate.close(this.physicalCloseTimeout, TimeUnit.SECONDS);
+			producerToClose.delegate.close(this.physicalCloseTimeout);
 		}
 		producerToClose = this.cache.poll();
 		while (producerToClose != null) {
 			try {
-				producerToClose.delegate.close(this.physicalCloseTimeout, TimeUnit.SECONDS);
+				producerToClose.delegate.close(this.physicalCloseTimeout);
 			}
 			catch (Exception e) {
 				logger.error("Exception while closing producer", e);
@@ -234,7 +235,7 @@ public void destroy() throws Exception { //NOSONAR
 		}
 		synchronized (this.consumerProducers) {
 			this.consumerProducers.forEach(
-				(k, v) -> v.delegate.close(this.physicalCloseTimeout, TimeUnit.SECONDS));
+				(k, v) -> v.delegate.close(this.physicalCloseTimeout));
 			this.consumerProducers.clear();
 		}
 	}
@@ -337,11 +338,11 @@ Producer<K, V> createTransactionalProducerForPartition() {
 		}
 	}
 
-	private void removeConsumerProducer(CloseSafeProducer<K, V> producer) {
+	private void removeConsumerProducer(CloseSafeProducer<K, V> producerToRemove) {
 		synchronized (this.consumerProducers) {
 			Iterator<Entry<String, CloseSafeProducer<K, V>>> iterator = this.consumerProducers.entrySet().iterator();
 			while (iterator.hasNext()) {
-				if (iterator.next().getValue().equals(producer)) {
+				if (iterator.next().getValue().equals(producerToRemove)) {
 					iterator.remove();
 					break;
 				}
@@ -380,12 +381,12 @@ protected BlockingQueue<CloseSafeProducer<K, V>> getCache() {
 	}
 
 	@Override
-	public void closeProducerFor(String transactionIdSuffix) {
+	public void closeProducerFor(String suffix) {
 		if (this.producerPerConsumerPartition) {
 			synchronized (this.consumerProducers) {
-				CloseSafeProducer<K, V> removed = this.consumerProducers.remove(transactionIdSuffix);
+				CloseSafeProducer<K, V> removed = this.consumerProducers.remove(suffix);
 				if (removed != null) {
-					removed.delegate.close(this.physicalCloseTimeout, TimeUnit.SECONDS);
+					removed.delegate.close(this.physicalCloseTimeout);
 				}
 			}
 		}
@@ -524,22 +525,29 @@ public void abortTransaction() throws ProducerFencedException {
 
 		@Override
 		public void close() {
-			close(0, null);
+			close(null);
 		}
 
 		@Override
+		@SuppressWarnings("deprecation")
+		@Deprecated
 		public void close(long timeout, @Nullable TimeUnit unit) {
+			close(Duration.ofMillis(unit.toMillis(timeout)));
+		}
+
+		@Override
+		public void close(Duration timeout) {
 			if (this.cache != null) {
 				if (this.txFailed) {
 					if (logger.isWarnEnabled()) {
 						logger.warn("Error during transactional operation; producer removed from cache; possible cause: "
 							+ "broker restarted during transaction: " + this);
 					}
-					if (unit == null) {
+					if (timeout == null) {
 						this.delegate.close();
 					}
 					else {
-						this.delegate.close(timeout, unit);
+						this.delegate.close(timeout);
 					}
 					if (this.removeConsumerProducer != null) {
 						this.removeConsumerProducer.accept(this);
@@ -550,11 +558,11 @@ public void close(long timeout, @Nullable TimeUnit unit) {
 						synchronized (this) {
 							if (!this.cache.contains(this)
 									&& !this.cache.offer(this)) {
-								if (unit == null) {
+								if (timeout == null) {
 									this.delegate.close();
 								}
 								else {
-									this.delegate.close(timeout, unit);
+									this.delegate.close(timeout);
 								}
 							}
 						}