[Enhancement] Improve ACK and ChangeInvisibleTime performance with POP KV group commit

Aperta
  • Qualcosa non va?
#11,489 0 commenti 0 reazioni 0 assegnatari Vedi su GitHub

I maintainer di solito rispondono entro 1 giorno

Nessuno ha ancora preso questa issue.

  • #11490 di @qianye1001 — chiusa senza merge

Valutazione

Difficoltà
5/5
Tempo stimato
Più di una settimana
Idoneità per principianti
30/100
Tipo di issue
Funzionalità
Chiarezza
Specificata chiaramente
Stato di attività
Attiva
Stack tecnologico
java
Ambito
backend, databases

Direzione di ricerca

Start with PopConsumerRocksdbStore.deleteRecords(), changeInvisibilityDuration(), and the POP KV cache flush and maintenance paths described in the issue. Trace how these writes share the database and how CHANGE orders cache transfer, persistence, and removal; then identify the existing ServiceThread and response Future patterns. Done means ACK and CHANGE responses follow persistence success or failure, batching preserves ordering and durability, and shutdown drains accepted work.

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

Descrizione

Before Creating the Enhancement Request
  • I have confirmed that this should be classified as an enhancement rather than a bug/feature.
Summary

With popConsumerKVServiceEnable=true, ACK and ChangeInvisibleTime can spend substantial time waiting inside synchronous RocksDB writes. Both request types share the Broker ACK executor, so bursts of invisible-time changes can occupy its workers and delay otherwise unrelated ACK requests.

Improve ACK/CHANGE throughput and tail latency by batching POP KV mutations across requests and completing responses asynchronously after persistence, while keeping synchronous WAL durability.

Motivation

The current POP KV path has the following behavior:

  • An ACK that misses the POP checkpoint cache calls PopConsumerRocksdbStore.deleteRecords() directly. Although the service method is named ackAsync(), the RocksDB write happens before the method returns.
  • changeInvisibilityDuration() writes the replacement checkpoint synchronously and, when necessary, deletes the old checkpoint in another synchronous write. Even when the old checkpoint is cached, the replacement checkpoint still requires a storage write.
  • Both write paths use WriteOptions.sync=true and WAL enabled. The request worker remains occupied until db.write() returns; WAL synchronization and RocksDB write stalls can therefore consume the shared ACK executor's capacity.
  • POP checkpoint writes, cache flushing and expired-checkpoint/revive maintenance also write to the same database. Cache transfer must be ordered with CHANGE across selection, persistence and cache removal: otherwise a delayed flush can restore an old checkpoint after CHANGE has persisted its replacement and deleted the old key.

Bursty renewals, client-disconnection cleanup or other batches of invisible-time changes can consequently increase ACK queueing latency and reduce consumption throughput. Increasing the request thread count does not remove the synchronous storage work or aggregate mutations at the Broker layer. RocksDB can already group concurrent writers internally, but callers still wait in its synchronous write API.

A workload for evaluating this problem is a small ACK executor serving a steady ACK stream together with periodic CHANGE bursts. Use enablePopBufferMerge=false to exercise storage ACKs directly, and repeat with the cache enabled to cover renewal writes and cache misses. Measure ACK/CHANGE latency, executor queueing, completed requests and WAL synchronization activity. This issue does not claim a measured end-to-end throughput improvement.

Describe the Solution You'd Like
  • Introduce a shared, unbounded POP KV mutation queue and one batch writer based on RocketMQ ServiceThread. Drain currently available requests into a WriteBatch without deliberately delaying low-load requests to collect a batch.
  • Return real persistence Futures to ACK and CHANGE callers. Send success only after the required write completes; propagate write failures and submissions rejected during shutdown to the response.
  • Make the POP KV CHANGE response asynchronous independently of the legacy appendCkAsync / appendAckAsync settings.
  • Route checkpoint creation and background checkpoint maintenance through the same writer. Serialize cache transfer and CHANGE state transitions on the writer, including cache inspection and post-commit removal, so either ordering completes without restoring the old checkpoint. Preserve operation order, cached-checkpoint safety, same-key replacement behavior and scan-cursor advancement after successful persistence.
  • Preserve sync=true and WAL enabled, limit individual batch size, and drain accepted writes and completion callbacks before closing the database.
Describe Alternatives You've Considered
  • More ACK workers: allows more synchronous callers to wait, but does not address request-thread occupancy or Broker-level batching.
  • One background write per request: releases request workers, but serializing individual small writes without batching can reduce throughput.
  • sync=false: changes the durability guarantee and still does not provide a callback-based nonblocking write API.
  • An ACK worker elected as batch leader: can bound blocked ACK workers to one; a dedicated writer instead isolates the request pool from storage stalls and also serves background producers.
Additional Context

Inspected on develop at 8ab8191478115d5e2f5344127fba704267fdb659 (project version 5.5.1, RocksDB JNI 8.4.4).

Related: #10489 proposes a new batch CHANGE protocol across Proxy/Broker. This issue concerns Broker-side aggregation and asynchronous persistence for existing ACK/CHANGE requests, without requiring a new client or wire protocol.

Lingua principale
Java
Stelle
22.7k
Fork
12k
Merge medio
4g 20h
PR unite (30g)
11

Preparare l'ambiente

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

Tutte le issue di apache/rocketmq

Issue simili

Altre issue su Java

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.