MINOR: Fix logged timeout in KafkaProducer.close()
The log line says `ms`, but the actual value could represent a
different time unit depending on what the user provided.
Reviewers: Jason Gustafson Ismael Juma
@@ -1078,21 +1078,24 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
if (timeout < 0)
throw new IllegalArgumentException("The timeout cannot be negative.");
- log.info("Closing the Kafka producer with timeoutMillis = {} ms.", timeUnit.toMillis(timeout));
+ long timeoutMs = timeUnit.toMillis(timeout);
+ log.info("Closing the Kafka producer with timeoutMillis = {} ms.", timeoutMs);
+
// this will keep track of the first encountered exception
AtomicReference<Throwable> firstException = new AtomicReference<>();
boolean invokedFromCallback = Thread.currentThread() == this.ioThread;
- if (timeout > 0) {
+ if (timeoutMs > 0) {
if (invokedFromCallback) {
log.warn("Overriding close timeout {} ms to 0 ms in order to prevent useless blocking due to self-join. " +
- "This means you have incorrectly invoked close with a non-zero timeout from the producer call-back.", timeout);
+ "This means you have incorrectly invoked close with a non-zero timeout from the producer call-back.",
+ timeoutMs);
} else {
// Try to close gracefully.
if (this.sender != null)
this.sender.initiateClose();
if (this.ioThread != null) {
try {
- this.ioThread.join(timeUnit.toMillis(timeout));
+ this.ioThread.join(timeoutMs);
} catch (InterruptedException t) {
firstException.compareAndSet(null, new InterruptException(t));
log.error("Interrupted while joining ioThread", t);
@@ -1103,7 +1106,7 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
if (this.sender != null && this.ioThread != null && this.ioThread.isAlive()) {
log.info("Proceeding to force close the producer since pending requests could not be completed " +
- "within timeout {} ms.", timeout);
+ "within timeout {} ms.", timeoutMs);
this.sender.forceClose();
// Only join the sender thread when not calling from callback.
if (!invokedFromCallback) {
