You are viewing a plain text version of this content. The canonical link for it is here.

Posted to commits@rocketmq.apache.org by "qianye1001 (via GitHub)" <gi...@apache.org> on 2026/06/12 06:19:23 UTC

[I] [Enhancement] Support BatchChangeInvisibleTime to reduce network round-trips and RocksDB flushes [rocketmq]

qianye1001 opened a new issue, #10489:
URL: https://github.com/apache/rocketmq/issues/10489

   ### Before Creating the Enhancement Request
   
   - [x] I have confirmed that this should be classified as an enhancement rather than a bug/feature.
   
   ### Summary
   
   Add `BatchChangeInvisibleTime` support in the Proxy and Broker, following the existing `BatchAck` pattern. Currently, when the Proxy needs to change invisible time for multiple messages (client disconnect, write-back failure, periodic renewal, message filtering), it sends individual `CHANGE_MESSAGE_INVISIBLETIME` requests to the Broker one by one. This causes excessive network round-trips and, on the PopConsumerKVService (RocksDB) path, excessive RocksDB flushes (2N `db.write()` calls for N messages).
   
   ### Motivation
   
   Multiple scenarios in the Proxy require batch `changeInvisibleTime`:
   
   1. **Client disconnect** (`DefaultReceiptHandleManager.returnHandleGroup`): When a client goes offline, the Proxy scans all unacked receipt handles for that client and calls `changeInvisibleTime` for each one individually. A consumer with many in-flight messages triggers N separate network round-trips.
   
   2. **ReceiveMessage write-back failure** (`ReceiveMessageResponseStreamWriter`): When writing messages back to the gRPC client fails mid-stream, the Proxy nacks all remaining messages one by one.
   
   3. **Periodic renewal** (`DefaultReceiptHandleManager.scheduleRenewTask`): The Proxy periodically renews receipt handles that are about to expire. Each handle is renewed individually via a separate `changeInvisibleTime` call.
   
   4. **Pop message filtering** (`ConsumerProcessor.filterPopResult`): Messages filtered as `TO_RETURN` are changed invisible time individually.
   
   On the Broker side with `PopConsumerKVServiceEnable=true`, each individual `changeInvisibleTime` call in `PopConsumerService.changeInvisibilityDuration()` does:
   - `popConsumerStore.writeRecords(Collections.singletonList(ckRecord))` — 1 RocksDB WriteBatch with 1 entry
   - `popConsumerStore.deleteRecords(Collections.singletonList(ackRecord))` — 1 RocksDB WriteBatch with 1 entry
   
   For N messages, this results in 2N `db.write()` calls. Since `PopConsumerRocksdbStore` uses `sync=true` on WriteOptions, each write forces a WAL flush, making this a significant bottleneck.
   
   ### Describe the Solution You'd Like
   
   Add a new `BATCH_CHANGE_MESSAGE_INVISIBLETIME` request code, following the existing `BatchAck` (`BATCH_ACK_MESSAGE`) pattern:
   
   **Request format** — Each entry carries the same parameters as the single `ChangeInvisibleTimeRequestHeader`:
   - `consumerGroup`, `topic`, `queueId`, `extraInfo`, `offset`, `invisibleTime`, `liteTopic`, `suspend`
   
   **Response format** — Per-entry results with new receipt handle info (same as `ChangeInvisibleTimeResponseHeader`):
   - `code` (SUCCESS or error), `popTime`, `invisibleTime`, `reviveQid`
   
   **Layer-by-layer changes:**
   
   1. **Remoting**: Add `RequestCode.BATCH_CHANGE_MESSAGE_INVISIBLETIME`, `BatchChangeInvisibleTimeRequestBody` (with `List<ChangeInvisibleTimeRequestEntry>`), `BatchChangeInvisibleTimeResponseBody` (with per-entry results)
   
   2. **Broker** — True batch processing:
      - `PopConsumerService.batchChangeInvisibilityDuration()`: Collect all CK records and ack records, then call `writeRecords(allCkRecords)` once and `deleteRecords(allAckRecords)` once. This reduces N messages from 2N `db.write()` to just 2 `db.write()` calls.
      - `ChangeInvisibleTimeProcessor`: Handle `BATCH_CHANGE_MESSAGE_INVISIBLETIME` — on KV path, delegate to the batch method above; on traditional revive path, iterate and call `appendCheckPointThenAckOrigin()` per entry.
      - `BrokerController`: Register the new request code to `changeInvisibleTimeProcessor`.
   
   3. **Client**: Add `batchChangeInvisibleTimeAsync()` to `MQClientAPIImpl` and `MQClientAPIExt`.
   
   4. **Proxy Service**: Add `batchChangeInvisibleTime()` to `MessageService`, `ClusterMessageService`, and `LocalMessageService`.
   
   5. **Proxy Processor**: Add `batchChangeInvisibleTime()` to `ConsumerProcessor` (group by broker, filter expired handles), `MessagingProcessor` interface, and `DefaultMessagingProcessor`.
   
   6. **Caller integration**:
      - `DefaultReceiptHandleManager.returnHandleGroup()`: Collect all handles, batch send
      - `DefaultReceiptHandleManager.scheduleRenewTask()`: Collect handles needing renewal per group key, batch send, update each handle with per-entry result
      - `ReceiveMessageResponseStreamWriter`: Collect all nack messages, batch send
      - `ConsumerProcessor.filterPopResult()`: Collect all TO_RETURN messages, batch send
   
   7. **Config**: Add `enableBatchChangeInvisibleTime` flag (default false) for gradual rollout.
   
   ### Describe Alternatives You've Considered
   
   1. **Proxy-level batching only (no Broker changes)**: Group entries by broker at the Proxy, but still send individual `CHANGE_MESSAGE_INVISIBLETIME` requests. This reduces Proxy-side processing overhead but not the number of network round-trips or RocksDB flushes. Rejected because the Broker-side RocksDB flush reduction is the main performance win.
   
   2. **Reuse `BATCH_ACK_MESSAGE` with BitSet merging**: `BatchAck` merges multiple offsets sharing the same CK into a BitSet. This doesn't work for `ChangeInvisibleTime` because each entry requires its own new checkpoint write + old checkpoint ack, and each entry may have a different `invisibleTime`. A list-based request body is necessary.
   
   3. **Async fire-and-forget (no per-entry response)**: For the `returnHandleGroup` scenario, per-entry results aren't strictly needed. However, the renewal scenario (`startRenewMessage`) requires per-entry new receipt handles to update the cached handle via `messageReceiptHandle.updateReceiptHandle()`. Supporting per-entry responses makes the batch API general-purpose across all scenarios.
   
   ### Additional Context
   
   **RocksDB batch optimization detail:**
   
   Current (N messages = 2N `db.write()`):
   ```java
   // PopConsumerService.changeInvisibilityDuration() — called N times
   this.popConsumerStore.writeRecords(Collections.singletonList(ckRecord));   // 1 WriteBatch, 1 entry
   this.popConsumerStore.deleteRecords(Collections.singletonList(ackRecord)); // 1 WriteBatch, 1 entry
   ```
   
   Proposed (N messages = 2 `db.write()`):
   ```java
   // PopConsumerService.batchChangeInvisibilityDuration() — called once
   this.popConsumerStore.writeRecords(allCkRecords);     // 1 WriteBatch, N entries
   this.popConsumerStore.deleteRecords(allAckRecords);    // 1 WriteBatch, N entries
   ```
   
   `PopConsumerRocksdbStore.writeRecords/deleteRecords` already accept `List<PopConsumerRecord>` and use `WriteBatch` internally, so the infrastructure for batch RocksDB writes is already in place — we just need to stop passing `Collections.singletonList()`.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [I] [Enhancement] Support BatchChangeInvisibleTime to reduce network round-trips and RocksDB flushes [rocketmq]

Posted by "RockteMQ-AI (via GitHub)" <gi...@apache.org>.

RockteMQ-AI commented on issue #10489:
URL: https://github.com/apache/rocketmq/issues/10489#issuecomment-4709877351

**Issue Evaluation**

Category: `type/enhancement` | Status: **Evaluated**

**Feasibility:** Feasible, but complex scope
**Scope:** Proxy layer (gRPC handler), Broker layer (new request code + processor), potentially `rocketmq-clients`
**Compatibility:** New request code — backward compatible. Older brokers will reject the batch request; Proxy should fall back to individual calls.

This is a meaningful optimization for scenarios with bulk invisible time changes (client disconnect, write-back failure, periodic renewal). The proposal to follow the existing `BatchAck` pattern is sound.

Key considerations:
- **Batch size limit** — Need a configurable upper bound to prevent oversized requests.
- **Partial failure semantics** — If some messages in the batch fail, should the whole batch rollback or return per-message results?
- **RocksDB flush reduction** — The batch write should use a single `WriteBatch` to amortize flush cost.
- **Client SDK coordination** — If the Java/Go/C++ clients need to call this directly, protocol changes in `rocketmq-clients` are needed.

This is a larger feature that benefits from a design discussion before implementation. A RIP (RocketMQ Improvement Proposal) or detailed design doc would help align on the approach.

---
*Automated evaluation by RockteMQ-AI*

--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [I] [Enhancement] Support BatchChangeInvisibleTime to reduce network round-trips and RocksDB flushes [rocketmq]

Posted by "RockteMQ-AI (via GitHub)" <gi...@apache.org>.

RockteMQ-AI commented on issue #10489:
URL: https://github.com/apache/rocketmq/issues/10489#issuecomment-4689086976

   **Issue Evaluation**
   
   Category: `type/enhancement` | Status: **Evaluated**
   
   Thank you for this enhancement suggestion. This has been noted for community discussion.
   
   If you'd like to contribute a fix, feel free to submit a PR. The maintainers will review and provide feedback.
   
   ---
   *Automated evaluation by RockteMQ-AI*
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [I] [Enhancement] Support BatchChangeInvisibleTime to reduce network round-trips and RocksDB flushes [rocketmq]

Posted by "RockteMQ-AI (via GitHub)" <gi...@apache.org>.

RockteMQ-AI commented on issue #10489:
URL: https://github.com/apache/rocketmq/issues/10489#issuecomment-4711387847

**Issue Evaluation**

Category: `type/enhancement` | Status: **Evaluated**

Proposing `BatchChangeInvisibleTime` to batch multiple `ChangeInvisibleTime` operations into a single RPC, reducing network round-trips and RocksDB flushes.

**Assessment:** This is a significant feature enhancement touching the remoting protocol, broker processor, and potentially client SDKs. The motivation (reducing RocksDB write amplification from per-message invisible time changes) is valid.

**Feasibility:** Feasible but complex. Requires:
- New request code in remoting protocol
- Batch processor in broker
- Client SDK support (potentially coordinated with rocketmq-clients)
- Backward compatibility with existing single-message ChangeInvisibleTime

**Scope:** remoting, broker, store, potentially client SDKs
**Compatibility:** Additive change (new request code), but needs careful protocol design.
**Recommendation:** This is a larger effort — consider discussing the API design with committers before implementation.

---
*Automated evaluation by RockteMQ-AI*

--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org