Deadlock between `KafkaBinderMetrics.close()` and the offset-lag task on context shutdown

Start with KafkaBinderMetrics.close(), stop(), findTotalTopicGroupLag(), and createConsumerFactory(); the issue gives the relevant lock and map operations and a reproducible Maven test in a gist. Read the existing lifecycle and offset lag tests, then run the reproduction with mvn test. Done means shutdown returns reliably and the deadlock check no longer finds the reported cycle.

Version: spring-cloud-stream-binder-kafka 5.0.4 (the class is identical on main). Java 25.

Describe the issue

On application-context shutdown, KafkaBinderMetrics.close() can deadlock with the scheduled offset-lag task. When that happens the JVM never exits.

We hit this in CI. The Gradle test worker calls System.exit() after the tests finish. SpringApplicationShutdownHook then blocks forever in KafkaBinderMetrics.close(), and the build hangs until the pipeline timeout. In our case that is 8 out of 9 full runs on one build host.

The cycle (line numbers from 5.0.4 / main):

close() (shutdown thread)	offset-lag task (scheduler thread)
consumerFactoryLock.lock() (L294)
	metadataConsumers.computeIfAbsent(group, ...) (L209): holds the CHM bin lock
	→ createConsumerFactory(): defaultConsumerFactory == null → consumerFactoryLock.lock() (L255): blocks. lock() is not interruptible, so shutdownNow() does not help.
scheduler.shutdownNow(), awaitTermination(offsetLagMetricsInterval) times out
finally: metadataConsumers.clear() (L313): waits for the CHM bin lock held by the task → deadlock

The cycle needs the task's first createConsumerFactory() call to coincide with close(). That is, the binding was created shortly before shutdown, which is common with short-lived test contexts.

It looks like the cycle came from two changes: GH-3062 started calling close() from stop(), and it moved metadataConsumers.clear() into the locked section. #3014 replaced synchronized with ReentrantLock.

Thread dump from CI (Spring Boot app, Gradle test worker)

Found one Java-level deadlock:
"pool-15-thread-1":
  waiting for ownable synchronizer 0x00000000e30d4008, (a java.util.concurrent.locks.ReentrantLock$NonfairSync),
  which is held by "SpringApplicationShutdownHook"
"SpringApplicationShutdownHook":
  waiting to lock monitor 0x00007fda7000a2b0 (object 0x00000000e4e193d0, a java.util.concurrent.ConcurrentHashMap$ReservationNode),
  which is held by "pool-15-thread-2"
"pool-15-thread-2":
  waiting for ownable synchronizer 0x00000000e30d4008, (a java.util.concurrent.locks.ReentrantLock$NonfairSync),
  which is held by "SpringApplicationShutdownHook"

"SpringApplicationShutdownHook":
	at java.util.concurrent.ConcurrentHashMap.clear(ConcurrentHashMap.java:1214)
	- waiting to lock <0x00000000e4e193d0> (a java.util.concurrent.ConcurrentHashMap$ReservationNode)
	at org.springframework.cloud.stream.binder.kafka.KafkaBinderMetrics.close(KafkaBinderMetrics.java:305)
	at org.springframework.cloud.stream.binder.kafka.KafkaBinderMetrics.stop(KafkaBinderMetrics.java:321)
	at org.springframework.context.support.DefaultLifecycleProcessor.doStop(DefaultLifecycleProcessor.java:476)
	...
	at org.springframework.cloud.stream.binder.DefaultBinderFactory.stop(DefaultBinderFactory.java:163)
	...
	at org.springframework.boot.SpringApplicationShutdownHook.run(SpringApplicationShutdownHook.java:116)
"pool-15-thread-2":
	at java.util.concurrent.locks.ReentrantLock.lock(ReentrantLock.java:323)
	at org.springframework.cloud.stream.binder.kafka.KafkaBinderMetrics.createConsumerFactory(KafkaBinderMetrics.java:247)
	at org.springframework.cloud.stream.binder.kafka.KafkaBinderMetrics.lambda$findTotalTopicGroupLag$4(KafkaBinderMetrics.java:211)
	at java.util.concurrent.ConcurrentHashMap.computeIfAbsent(ConcurrentHashMap.java:1724)
	- locked <0x00000000e4e193d0> (a java.util.concurrent.ConcurrentHashMap$ReservationNode)
	at org.springframework.cloud.stream.binder.kafka.KafkaBinderMetrics.findTotalTopicGroupLag(KafkaBinderMetrics.java:209)
	at org.springframework.cloud.stream.binder.kafka.KafkaBinderMetrics.computeUnconsumedMessages(KafkaBinderMetrics.java:198)
	at org.springframework.cloud.stream.binder.kafka.KafkaBinderMetrics.lambda$bindTo$0(KafkaBinderMetrics.java:163)
	...

(The CI line numbers are from an earlier 4.x/5.0.x release; the code path is the same.)

To Reproduce

Minimal Maven project (pom.xml + a single JUnit test, no broker needed) that fails deterministically on 5.0.4: https://gist.github.com/rafallezanko/1ba280839b3a229f1afb6c06289070ee

Put the test file under src/test/java/org/springframework/cloud/stream/binder/kafka/ (it is in that package because getTopicsInUse() is package-private), then run:

mvn test

The only test hook is a SimpleMeterRegistry whose remove() waits until the lag task is queued on consumerFactoryLock. close() calls remove() while holding that lock, so the hook just widens the race window. The test fails because stop() does not return within 15 s, and ThreadMXBean.findDeadlockedThreads() reports the same two threads as above.

Note: the scheduler threads are non-daemon, so the forked test JVM also does not exit after the failure. That is the same symptom we see in CI.

Expected behavior

close() / stop() always completes, and the JVM can exit.

Possible fix (suggestion)

• Do not call metadataConsumers.clear() (or close the consumers) while holding consumerFactoryLock. Or release the lock before awaitTermination.
• And/or do not create the consumer factory inside the computeIfAbsent mapping function. Call createConsumerFactory() first, then computeIfAbsent(group, g -> factory.createConsumer(...)).

Workaround

Deny the gauge so the lag task is never scheduled:
management.metrics.enable.spring.cloud.stream.binder.kafka.offset=false (or a MeterFilter.deny for spring.cloud.stream.binder.kafka.offset).