showuon commented on code in PR #15133:
URL: https://github.com/apache/kafka/pull/15133#discussion_r1463221572
##########
core/src/test/scala/unit/kafka/utils/TestUtils.scala:
##########
@@ -1396,11 +1396,13 @@ object TestUtils extends Logging {
// Note: Call this method in the test itself, rather than the @AfterEach
method.
// Because of the assert, if assertNoNonDaemonThreads fails, nothing after
would be executed.
def assertNoNonDaemonThreads(threadNamePrefix: String): Unit = {
- val nonDaemonThreads = Thread.getAllStackTraces.keySet.asScala.filter { t
=>
- !t.isDaemon && t.isAlive && t.getName.startsWith(threadNamePrefix)
- }
- val threadCount = nonDaemonThreads.size
- assertEquals(0, threadCount, s"Found unexpected $threadCount NonDaemon
threads=${nonDaemonThreads.map(t => t.getName).mkString(", ")}")
+ var nonDemonThreads: mutable.Set[Thread] = mutable.Set.empty[Thread]
+ waitUntilTrue(() => {
+ nonDemonThreads = Thread.getAllStackTraces.keySet.asScala.filter { t =>
+ !t.isDaemon && t.isAlive && t.getName.startsWith(threadNamePrefix)
+ }
+ 0 == nonDemonThreads.size
+ }, s"Found unexpected ${nonDemonThreads.size} NonDaemon
threads=${nonDemonThreads.map(t => t.getName).mkString(", ")}", 1000)
Review Comment:
> BTW, which threads do you think are closed async during Kafka server
shutdown? AFAIK, we have checks in place to ensure that we wait for proper
close during shutdown
Because in the failed test cases, They are `ReplicaFetcherThread` leaked.
But had a quick look, yes, we did do `awaitShutdown()` for fetcher threads.
Hmm... Let me dig into further tomorrow.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
