Skip to content

[ISSUE #11489] Improve POP KV ACK and ChangeInvisibleTime performance with group commit #11490

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
qianye1001 wants to merge 6 commits into from
Closed

[ISSUE #11489] Improve POP KV ACK and ChangeInvisibleTime performance with group commit #11490

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
c522ab8
[ISSUE #11489] Improve POP KV ACK and ChangeInvisibleTime performance…
qianye1001 Oct 23, 2026
099d986
[ISSUE #11489] Run POP KV completion callbacks concurrently
qianye1001 Oct 23, 2026
c181cfa
[ISSUE #11489] Exclude local POP KV design document
qianye1001 Oct 23, 2026
9d6bace
[ISSUE #11489] Use ServiceThread and an unbounded POP KV write queue
qianye1001 Oct 23, 2026
5123b87
[ISSUE #11489] Serialize POP cache transfer and checkpoint state changes
qianye1001 Oct 23, 2026
e2bc827
[ISSUE #11489] Simplify POP KV cache operations into explicit writer …
qianye1001 Oct 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
160 changes: 117 additions & 43 deletions broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerCache.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,9 +17,14 @@
package org.apache.rocketmq.broker.pop;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ConcurrentSkipListMap;
Expand All @@ -41,14 +46,14 @@ public class PopConsumerCache extends ServiceThread {
private static final long OFFSET_NOT_EXIST = -1L;

private final BrokerController brokerController;
private final PopConsumerKVStore consumerRecordStore;
private final PopConsumerRocksdbStore consumerRecordStore;
private final PopConsumerLockService consumerLockService;
private final Consumer<PopConsumerRecord> reviveConsumer;

private final AtomicInteger estimateCacheSize;
private final ConcurrentMap<String, ConsumerRecords> consumerRecordTable;

public PopConsumerCache(BrokerController brokerController, PopConsumerKVStore consumerRecordStore,
public PopConsumerCache(BrokerController brokerController, PopConsumerRocksdbStore consumerRecordStore,
PopConsumerLockService popConsumerLockService, Consumer<PopConsumerRecord> reviveConsumer) {

this.reviveConsumer = reviveConsumer;
Expand Down Expand Up @@ -89,6 +94,47 @@ public long getPopInFlightMessageCount(String groupId, String topicId, int queue
return consumerRecords != null ? consumerRecords.getInFlightRecordCount() : 0L;
}

private static List<PopConsumerRecord> snapshot(List<PopConsumerRecord> records) {
List<PopConsumerRecord> copy = new ArrayList<>(records.size());
records.forEach(record -> copy.add(PopConsumerRecord.decode(record.getValueBytes())));
return copy;
}

public CompletableFuture<Void> writeRecordsAsync(List<PopConsumerRecord> records) {
return consumerRecordStore.submitCacheRequest(new PopConsumerWriteBatchService.Request(
PopConsumerWriteBatchService.Type.POP, this, snapshot(records)));
}

public CompletableFuture<Void> ackAsync(PopConsumerRecord record) {
return consumerRecordStore.submitCacheRequest(new PopConsumerWriteBatchService.Request(
PopConsumerWriteBatchService.Type.ACK, this, snapshot(Collections.singletonList(record))));
}

public CompletableFuture<Void> changeAsync(PopConsumerRecord old, List<PopConsumerRecord> puts,
List<PopConsumerRecord> deletes) {
PopConsumerWriteBatchService.Request request = new PopConsumerWriteBatchService.Request(
PopConsumerWriteBatchService.Type.CHANGE, this, snapshot(Collections.singletonList(old)));
for (PopConsumerRecord record : puts) {
request.mutations.add(new PopConsumerWriteBatchService.Mutation(record.getKeyBytes(), record.getValueBytes()));
}
// Also delete a disk key left by a preceding transfer, even if its write reported failure.
for (PopConsumerRecord record : deletes) {
request.mutations.add(new PopConsumerWriteBatchService.Mutation(record.getKeyBytes(), null));
}
return consumerRecordStore.submitCacheRequest(request);
}

boolean contains(PopConsumerRecord record) {
ConsumerRecords records = consumerRecordTable.get(getKey(record));
return records != null && sameCheckpoint(records.recordTreeMap.get(record.getOffset()), record);
}

static boolean sameCheckpoint(PopConsumerRecord current, PopConsumerRecord expected) {
return current != null && current.getPopTime() == expected.getPopTime()
&& current.getInvisibleTime() == expected.getInvisibleTime();
}

// Cache mutations are published on the KV writer after the corresponding batch commits.
public void writeRecords(List<PopConsumerRecord> consumerRecordList) {
this.estimateCacheSize.addAndGet(consumerRecordList.size());
consumerRecordList.forEach(consumerRecord -> {
Expand Down Expand Up @@ -117,53 +163,56 @@ public List<PopConsumerRecord> deleteRecords(List<PopConsumerRecord> consumerRec
}

public int cleanupRecords(Consumer<PopConsumerRecord> consumer) {
int remain = 0;
Iterator<Map.Entry<String, ConsumerRecords>> iterator = consumerRecordTable.entrySet().iterator();
while (iterator.hasNext()) {
// revive or write record to store
ConsumerRecords records = iterator.next().getValue();
boolean timeout = consumerLockService.isLockTimeout(
records.getGroupId(), records.getTopicId());

if (timeout) {
records.stageExpiredRecords(Long.MAX_VALUE);
List<PopConsumerRecord> writeConsumerRecords =
new ArrayList<>(records.getRemoveTreeMap().values());
if (!writeConsumerRecords.isEmpty()) {
consumerRecordStore.writeRecords(writeConsumerRecords);
}
records.clearStagedRecords();
log.info("PopConsumerOffline, so clean expire records, groupId={}, topic={}, queueId={}, records={}",
records.getGroupId(), records.getTopicId(), records.getQueueId(), records.getInFlightRecordCount());
iterator.remove();
continue;
return cleanupRecordsAsync(consumer).join();
}

public CompletableFuture<Integer> cleanupRecordsAsync(Consumer<PopConsumerRecord> consumer) {
PopConsumerWriteBatchService.Request request = new PopConsumerWriteBatchService.Request(
PopConsumerWriteBatchService.Type.TRANSFER, this, Collections.emptyList());
return consumerRecordStore.submitCacheRequest(request).thenApply(ignored -> {
// Revive may read messages or send retries; never execute it on the KV writer.
if (!request.expired.isEmpty()) {
request.expired.forEach(consumer);
}
return getCacheSize();
});
}

long currentTime = System.currentTimeMillis();
records.stageExpiredRecords(currentTime);
List<PopConsumerRecord> writeConsumerRecords = new ArrayList<>();
records.getRemoveTreeMap().values().forEach(record -> {
if (record.getVisibilityTimeout() <= currentTime) {
consumer.accept(record);
List<PopConsumerRecord> prepareTransfer(List<PopConsumerRecord> expired, List<String> offlineQueues) {
List<PopConsumerRecord> writes = new ArrayList<>();
long now = System.currentTimeMillis();
for (Map.Entry<String, ConsumerRecords> entry : consumerRecordTable.entrySet()) {
ConsumerRecords records = entry.getValue();
boolean offline = consumerLockService.isLockTimeout(records.getGroupId(), records.getTopicId());
if (offline) {
offlineQueues.add(entry.getKey());
}
records.stageExpiredRecords(offline ? Long.MAX_VALUE : now);
for (PopConsumerRecord record : records.removeTreeMap.values()) {
if (!offline && !records.pendingDisk.contains(record) && record.getVisibilityTimeout() <= now) {
expired.add(record);
} else {
writeConsumerRecords.add(record);
// Keep this marker on failure: the write may have reached disk.
records.pendingDisk.add(record);
writes.add(record);
}
});
}
}
return writes;
}

// write to store and handle it later
consumerRecordStore.writeRecords(writeConsumerRecords);
void completeTransfer(List<String> offlineQueues) {
// The writer commits the entire transfer before processing any later cache operation.
offlineQueues.forEach(consumerRecordTable::remove);
for (ConsumerRecords records : consumerRecordTable.values()) {
records.clearStagedRecords();

// commit min offset in buffer to offset store
long offset = records.getMinOffsetInBuffer();
if (offset > OFFSET_NOT_EXIST) {
this.commitOffset("PopConsumerCache",
records.getGroupId(), records.getTopicId(), records.getQueueId(), offset);
commitOffset("PopConsumerCache", records.getGroupId(), records.getTopicId(), records.getQueueId(), offset);
}

remain += records.getInFlightRecordCount();
}
return remain;
estimateCacheSize.set(consumerRecordTable.values().stream()
.mapToInt(ConsumerRecords::getInFlightRecordCount).sum());
}

public void commitOffset(String clientHost, String groupId, String topicId, int queueId, long offset) {
Expand All @@ -184,7 +233,23 @@ public void commitOffset(String clientHost, String groupId, String topicId, int
}

public void removeRecords(String groupId, String topicId, int queueId) {
this.consumerRecordTable.remove(this.getKey(groupId, topicId, queueId));
removeRecords(getKey(groupId, topicId, queueId));
}

void removeRecords(String key) {
ConsumerRecords removed = this.consumerRecordTable.remove(key);
if (removed != null) {
estimateCacheSize.addAndGet(-removed.getInFlightRecordCount());
}
}

public void clearRecords(String groupId, String topicId, int queueId) {
PopConsumerWriteBatchService.Request request = new PopConsumerWriteBatchService.Request(
PopConsumerWriteBatchService.Type.CLEAR, this, Collections.emptyList());
request.cacheKey = getKey(groupId, topicId, queueId);
// Private future: completion workers may block here without waiting on themselves.
request.asyncCompletion = false;
consumerRecordStore.submitCacheRequest(request).join();
}

@Override
Expand All @@ -197,8 +262,7 @@ public void run() {
while (!this.isStopped()) {
try {
this.waitForRunning(TimeUnit.SECONDS.toMillis(1));
int cacheSize = this.cleanupRecords(reviveConsumer);
this.estimateCacheSize.set(cacheSize);
this.cleanupRecords(reviveConsumer);
} catch (Exception e) {
log.error("PopConsumerCacheService revive error", e);
}
Expand All @@ -213,6 +277,7 @@ protected static class ConsumerRecords {
private final BrokerConfig brokerConfig;
private final ConcurrentSkipListMap<Long /* offset */, PopConsumerRecord> removeTreeMap;
private final ConcurrentSkipListMap<Long /* offset */, PopConsumerRecord> recordTreeMap;
private final Set<PopConsumerRecord> pendingDisk = new HashSet<>();

public ConsumerRecords(BrokerConfig brokerConfig, String groupId, String topicId, int queueId) {
this.groupId = groupId;
Expand All @@ -228,7 +293,15 @@ public void write(PopConsumerRecord record) {
}

public boolean delete(PopConsumerRecord record) {
return recordTreeMap.remove(record.getOffset()) != null;
boolean removed = false;
for (ConcurrentSkipListMap<Long, PopConsumerRecord> records : Arrays.asList(recordTreeMap, removeTreeMap)) {
PopConsumerRecord current = records.get(record.getOffset());
if (sameCheckpoint(current, record) && records.remove(record.getOffset(), current)) {
pendingDisk.remove(current);
removed = true;
}
}
return removed;
}

public long getMinOffsetInBuffer() {
Expand Down Expand Up @@ -261,6 +334,7 @@ public void stageExpiredRecords(long currentTime) {

public void clearStagedRecords() {
removeTreeMap.clear();
pendingDisk.clear();
}

public ConcurrentSkipListMap<Long, PopConsumerRecord> getRemoveTreeMap() {
Expand Down
15 changes: 15 additions & 0 deletions broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerKVStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,9 @@
*/
package org.apache.rocketmq.broker.pop;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CompletableFuture;

public interface PopConsumerKVStore {

Expand Down Expand Up @@ -48,6 +50,19 @@ public interface PopConsumerKVStore {
*/
void deleteRecords(List<PopConsumerRecord> consumerRecordList);

default CompletableFuture<Void> writeRecordsAsync(List<PopConsumerRecord> records) {
return updateRecordsAsync(records, Collections.emptyList());
}

default CompletableFuture<Void> deleteRecordsAsync(List<PopConsumerRecord> records) {
return updateRecordsAsync(Collections.emptyList(), records);
}

/**
* Persist puts followed by deletes atomically. Complete only after the WAL is synced.
*/
CompletableFuture<Void> updateRecordsAsync(List<PopConsumerRecord> puts, List<PopConsumerRecord> deletes);

/**
* Scans and returns a list of expired consumer records within the specified time range.
* @param lowerTime The start time (inclusive) of the time range to search, in milliseconds.
Expand Down
Loading
Loading