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

Abierto
  • ¿Algo no está bien?
#3,332 1 comentario 0 reacciones 0 asignados Ver en GitHub

Los mantenedores suelen responder en 1 día

@caiocv18 ya está trabajando en esto.

Desde el 6/11/2026.

  • #3334 de @caiocv18 — abierto

Evaluación

Dificultad
4/5
Tiempo estimado
3-5 días
Aptitud para principiantes
55/100
Tipo de issue
Error
Claridad
Bien especificado
Estado de actividad
Activo
Stack tecnológico
java, kafka, spring
Área
backend, performance

Línea de trabajo

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.

Escrito por el modelo de indexación a partir del texto del issue.

Descripción

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

Lenguaje dominante
Java
Estrellas
1.1k
Forks
653
Merge medio
2 d 15 h
PR fusionados (30 d)
8

Preparar el entorno

Este proyecto no incluye contenedor de desarrollo, Dockerfile ni guía de contribución, así que la configuración corre por tu cuenta: empieza por su README y consulta nuestra guía para la primera contribución para los pasos generales.

Primeros pasos

  1. Lee el issue completo y luego la guía de contribución del proyecto.
  2. Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
  3. Haz un fork del repositorio y trabaja en una rama.
  4. Abre un pull request que haga referencia al número del issue.

Más de spring-cloud/spring-cloud-stream

Todos los issues de spring-cloud/spring-cloud-stream

Issues similares

Más issues de Java

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.