[ https://issues.apache.org/jira/browse/KAFKA-9998?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17111249#comment-17111249 ]
Chia-Ping Tsai commented on KAFKA-9998: --------------------------------------- It seems the blocking is caused by your custom callback? If we don't join the thread before completing close(), it causes resource leak. > KafkaProducer.close(timeout) still may block indefinitely > --------------------------------------------------------- > > Key: KAFKA-9998 > URL: https://issues.apache.org/jira/browse/KAFKA-9998 > Project: Kafka > Issue Type: Bug > Affects Versions: 2.4.1 > Reporter: radai rosenblatt > Priority: Major > > looking at KafkaProducer.close(timeout), we have this: > {code:java} > private void close(Duration timeout, boolean swallowException) { > long timeoutMs = timeout.toMillis(); > if (timeoutMs < 0) > throw new IllegalArgumentException("The timeout cannot be negative."); > 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 (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.", > timeoutMs); > } else { > // Try to close gracefully. > if (this.sender != null) > this.sender.initiateClose(); > if (this.ioThread != null) { > try { > this.ioThread.join(timeoutMs); <---- GRACEFUL JOIN > } catch (InterruptedException t) { > firstException.compareAndSet(null, new > InterruptException(t)); > log.error("Interrupted while joining ioThread", t); > } > } > } > } > 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.", timeoutMs); > this.sender.forceClose(); > // Only join the sender thread when not calling from callback. > if (!invokedFromCallback) { > try { > this.ioThread.join(); <----- UNBOUNDED JOIN > } catch (InterruptedException e) { > firstException.compareAndSet(null, new InterruptException(e)); > } > } > } > ... > } > {code} > specifically in our case the ioThread was running a (very) long running > user-provided callback which was preventing the producer from closing within > the given timeout. > > I think the 2nd join() call should either be _VERY_ short (since we're > already past the timeout at that stage) ir should not happen at all. -- This message was sent by Atlassian Jira (v8.3.4#803005)