[BUG] ManagedRequestPipeline.close() leaves in-flight request futures pending forever

Abierto
  • ¿Algo no está bien?
#305 0 comentarios 0 reacciones 0 asignados Ver en GitHub

@ImDanXie ya está trabajando en esto.

Desde el 25/9/2026.

  • #306 de @ImDanXie — abierto

Evaluación

Dificultad
3/5
Tiempo estimado
1-2 días
Aptitud para principiantes
68/100
Tipo de issue
Error
Claridad
Bien especificado
Estado de actividad
Activo
Stack tecnológico
java
Área
backend, distributed-systems

Línea de trabajo

Start with base-rpc/base-rpc-client/src/main/java/org/apache/bifromq/baserpc/client/ManagedRequestPipeline.java and compare close() with onStreamError(), especially how preflightTaskQueue and inflightTaskQueue are cancelled. Verify that closing a pipeline completes both unsent and already-sent request futures, using the full client/server RPC stack or the in-process transport reproduction described in the issue.

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

Descripción

Environment
  • BiFroMQ 4.0.0 / main @ eef5e3af
  • Module: base-rpc/base-rpc-client
  • File: base-rpc/base-rpc-client/src/main/java/org/apache/bifromq/baserpc/client/ManagedRequestPipeline.java (close())
Summary

close() cancels only preflightTaskQueue (requests not yet sent). It never touches inflightTaskQueue (sent, awaiting a response). After super.close() disposes the stream, responses can never arrive and onStreamError() — the only path that calls cancelInflightTasks() — will not fire again, so the futures of already-sent requests never complete. Callers on the direct (non-batched) query/execute path hang forever; the batched path is only saved by Batcher's orTimeout(maxBurstLatency) fallback.

Pipelines are replaced via old.close() on store→server route changes, so one range migration with a query in flight triggers this.

Suggested fix

Abort in-flight tasks on close, mirroring onStreamError:

if (isClosed.compareAndSet(false, true)) {
    super.close();
    meter.recordCount(RPCMetric.ReqPipelineCompleteCount);
    cancelPreflightTasks(new RequestAbortException("Pipeline has closed"));
    cancelInflightTasks(new RequestAbortException("Pipeline has closed"));
}
Verification note

A regression test requires a full RPC stack (client + server + stream disposal); we verified the path statically and in a rolling multi-node deployment (see below). A reproduction using the in-process transport can be provided if useful.

Production validation

A 6-node BifroMQ 4.0.0 cluster (standalone.sh, systemd, mTLS clientAuth=REQUIRE), fronted by a TCP load balancer with 6 backends. Live clients: 3 bridge replicas (MQTT5, $oshare subscribers) and 10 backend service instances (persistent sessions). RocksDB engine, default configuration otherwise.

The fix (plus other changes) was rolled across all six nodes one node at a time (wholesale lib/ replacement, then restart), with a per-node gate (service active, all four ports listening, cluster mesh re-established). Six of six nodes succeeded, zero rollbacks.

Post-deploy checks (all measured after the rollout):

Check Result
Cluster mesh (inter-node 8898/8899 connections, per node) 408–417 established on every node
Live client connections (3 bridge replicas + 10 backend services over mTLS) 13/13 reconnected and stable
Cross-node delivery end-to-end publish on one node → bridge (connected elsewhere) consumed it → forwarded (counter 0 → 1)
Client-visible protocol smoke connect / subscribe / publish with QoS 1 all CONNACK 0 / PUBACK 0

No non-batched caller hangs were observed in the rollout window; the fix removes the latent path.

Lenguaje dominante
Java
Estrellas
806
Forks
98
Merge medio
12 h 35 min
PR fusionados (30 d)
3

Preparar el entorno

  • Incluye un Dockerfile o un archivo de Docker Compose
  • Sin plantilla de pull request
  • Sin guía de contribución

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 apache/bifromq

Todos los issues de apache/bifromq

Issues similares

Más issues de Java

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.