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

Aperta
  • Qualcosa non va?
#3,332 0 commenti 0 reazioni 0 assegnatari Vedi su GitHub

I maintainer di solito rispondono entro 1 giorno

Nessuno ha ancora preso questa issue.

Valutazione

Difficoltà
4/5
Tempo stimato
3-5 giorni
Idoneità per principianti
55/100
Tipo di issue
Bug
Chiarezza
Specificata chiaramente
Stato di attività
Attiva
Stack tecnologico
java, kafka, spring
Ambito
backend, performance

Direzione di ricerca

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.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Descrizione

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).

Lingua principale
Java
Stelle
1.1k
Fork
651
Merge medio
2g 15h
PR unite (30g)
8

Preparare l'ambiente

Questo progetto non fornisce container di sviluppo, Dockerfile né guida per i contributori, quindi l'ambiente è a tuo carico: parti dal suo README e consulta la nostra guida al primo contributo per i passaggi generali.

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di spring-cloud/spring-cloud-stream

Tutte le issue di spring-cloud/spring-cloud-stream

Issue simili

Altre issue su Java

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.