This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 378196eaf6c [Subscription] Preserve trailing consensus progress
broadcasts (#19057)
378196eaf6c is described below
commit 378196eaf6cf9203c972ab9d561c4285ad455773
Author: Caideyipi <[email protected]>
AuthorDate: Fri Oct 30 12:50:20 2026 +0800
[Subscription] Preserve trailing consensus progress broadcasts (#19057)
* Fix trailing consensus subscription progress broadcasts
* Fix subscription broadcast state lifecycle
* Strengthen subscription broadcast lifecycle test
* Strengthen rejected broadcast schedule assertion
---
.../ConsensusSubscriptionCommitManager.java | 141 ++++++++--
.../ConsensusSubscriptionCommitBroadcastTest.java | 288 +++++++++++++++++++++
2 files changed, 409 insertions(+), 20 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java
index f9a27372d5e..b183ca745f9 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java
@@ -79,7 +79,9 @@ import java.util.Set;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.LongSupplier;
/**
* Manages committed progress for consensus-based subscriptions.
@@ -128,10 +130,12 @@ public class ConsensusSubscriptionCommitManager {
/** Rate-limiting: last broadcast timestamp per key. */
private final Map<String, Long> lastBroadcastTime = new
ConcurrentHashMap<>();
- /** Single-threaded executor for fire-and-forget broadcasts. */
- private final ExecutorService broadcastExecutor =
- IoTDBThreadPoolFactory.newSingleThreadExecutor(
- ThreadName.SUBSCRIPTION_CONSENSUS_PROGRESS_BROADCASTER.getName());
+ /** At most one scheduled broadcast per writer and commit-state generation.
*/
+ private final Map<String, ConsensusSubscriptionCommitState>
pendingBroadcasts =
+ new ConcurrentHashMap<>();
+
+ private final ScheduledExecutorService broadcastExecutor;
+ private final LongSupplier broadcastClock;
/** Key: versioned, encoded (consumerGroupId, topicName, regionId) ->
progress tracking state. */
private final Map<String, ConsensusSubscriptionCommitState> commitStates =
@@ -174,6 +178,19 @@ public class ConsensusSubscriptionCommitManager {
}
ConsensusSubscriptionCommitManager(final ConfigNodeProgressFetcher
configNodeProgressFetcher) {
+ this(
+ configNodeProgressFetcher,
+ IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor(
+ ThreadName.SUBSCRIPTION_CONSENSUS_PROGRESS_BROADCASTER.getName()),
+ System::currentTimeMillis);
+ }
+
+ ConsensusSubscriptionCommitManager(
+ final ConfigNodeProgressFetcher configNodeProgressFetcher,
+ final ScheduledExecutorService broadcastExecutor,
+ final LongSupplier broadcastClock) {
+ this.broadcastExecutor = broadcastExecutor;
+ this.broadcastClock = broadcastClock;
this.configNodeProgressFetcher =
Objects.nonNull(configNodeProgressFetcher)
? configNodeProgressFetcher
@@ -307,7 +324,7 @@ public class ConsensusSubscriptionCommitManager {
final CommitStateKey stateKey =
getCommitStateKey(setupSnapshot.consumerGroupId, topicName,
entry.getKey());
final StateSetupSnapshot stateSnapshot = entry.getValue();
- commitStates.put(
+ replaceCommitState(
stateKey.encodedStateKey,
new ConsensusSubscriptionCommitState(
stateKey.regionIdStr,
@@ -363,7 +380,7 @@ public class ConsensusSubscriptionCommitManager {
|| preservedRegionIds.contains(entry.getValue().regionIdStr)) {
continue;
}
- commitStates.remove(entry.getKey());
+ removeCommitState(entry.getKey());
pendingInitialProgressKeys.remove(entry.getKey());
pendingNewRegionProgressKeys.remove(entry.getKey());
configNodeProgressAvailableKeys.remove(entry.getKey());
@@ -709,7 +726,7 @@ public class ConsensusSubscriptionCommitManager {
final CommitStateKey stateKey =
getCommitStateKey(consumerGroupId, topicName, regionId.toString());
recoverTopicStatesIfNeeded(stateKey);
- commitStates.remove(stateKey.encodedStateKey);
+ removeCommitState(stateKey.encodedStateKey);
commitStateKeys.remove(stateKey.encodedStateKey);
pendingInitialProgressKeys.remove(stateKey.encodedStateKey);
pendingNewRegionProgressKeys.remove(stateKey.encodedStateKey);
@@ -732,7 +749,7 @@ public class ConsensusSubscriptionCommitManager {
while (keyIterator.hasNext()) {
final Map.Entry<String, CommitStateKey> entry = keyIterator.next();
if (entry.getValue().sameTopic(consumerGroupId, topicName)) {
- commitStates.remove(entry.getKey());
+ removeCommitState(entry.getKey());
pendingInitialProgressKeys.remove(entry.getKey());
pendingNewRegionProgressKeys.remove(entry.getKey());
configNodeProgressAvailableKeys.remove(entry.getKey());
@@ -854,8 +871,8 @@ public class ConsensusSubscriptionCommitManager {
// ======================== Progress Broadcast (Leader → Follower)
========================
/**
- * Broadcasts committed progress to followers if enough time has elapsed
since the last broadcast
- * for this key. The broadcast is async and fire-and-forget.
+ * Coalesces committed progress while throttling broadcasts. The latest ACK
is sent even when no
+ * further writes or ACKs arrive after the throttle window.
*/
private void maybeBroadcast(
final String key,
@@ -868,23 +885,105 @@ public class ConsensusSubscriptionCommitManager {
return;
}
final String broadcastKey = buildBroadcastKey(key, committedWriterId);
- final long now = System.currentTimeMillis();
- final Long last = lastBroadcastTime.get(broadcastKey);
- if (last != null && now - last < MIN_BROADCAST_INTERVAL_MS) {
+ final ConsensusSubscriptionCommitState state = commitStates.get(key);
+ if (Objects.isNull(state)) {
return;
}
- lastBroadcastTime.put(broadcastKey, now);
- broadcastExecutor.submit(
- () ->
- doBroadcast(
- consumerGroupId, topicName, regionId, committedWriterProgress,
committedWriterId));
+ synchronized (lastBroadcastTime) {
+ if (commitStates.get(key) != state) {
+ return;
+ }
+ if (pendingBroadcasts.get(broadcastKey) == state) {
+ return;
+ }
+ final Long last = lastBroadcastTime.get(broadcastKey);
+ final long delay =
+ Objects.isNull(last)
+ ? 0L
+ : Math.max(0L, MIN_BROADCAST_INTERVAL_MS -
(broadcastClock.getAsLong() - last));
+ pendingBroadcasts.put(broadcastKey, state);
+ try {
+ broadcastExecutor.schedule(
+ () -> {
+ synchronized (state.broadcastLifecycleLock) {
+ final WriterProgress latestProgress;
+ synchronized (lastBroadcastTime) {
+ if (!pendingBroadcasts.remove(broadcastKey, state)
+ || commitStates.get(key) != state) {
+ return;
+ }
+ lastBroadcastTime.put(broadcastKey,
broadcastClock.getAsLong());
+ latestProgress =
+ state
+ .getCommittedRegionProgress()
+ .getWriterPositions()
+ .get(committedWriterId);
+ }
+ if (Objects.nonNull(latestProgress)) {
+ doBroadcast(
+ consumerGroupId, topicName, regionId, latestProgress,
committedWriterId);
+ }
+ }
+ },
+ delay,
+ TimeUnit.MILLISECONDS);
+ } catch (final RuntimeException e) {
+ pendingBroadcasts.remove(broadcastKey, state);
+ throw e;
+ }
+ }
+ }
+
+ private void removeCommitState(final String stateKey) {
+ while (true) {
+ final ConsensusSubscriptionCommitState state =
commitStates.get(stateKey);
+ if (Objects.isNull(state)) {
+ return;
+ }
+ synchronized (state.broadcastLifecycleLock) {
+ synchronized (lastBroadcastTime) {
+ if (commitStates.remove(stateKey, state)) {
+ clearBroadcastStateUnderLock(stateKey);
+ return;
+ }
+ }
+ }
+ }
+ }
+
+ private void replaceCommitState(
+ final String stateKey, final ConsensusSubscriptionCommitState
replacement) {
+ while (true) {
+ final ConsensusSubscriptionCommitState state =
commitStates.get(stateKey);
+ if (Objects.isNull(state)) {
+ if (Objects.isNull(commitStates.putIfAbsent(stateKey, replacement))) {
+ return;
+ }
+ continue;
+ }
+ synchronized (state.broadcastLifecycleLock) {
+ synchronized (lastBroadcastTime) {
+ if (commitStates.replace(stateKey, state, replacement)) {
+ clearBroadcastStateUnderLock(stateKey);
+ return;
+ }
+ }
+ }
+ }
+ }
+
+ /** Clears scheduled work and throttle history for one subscription state
generation. */
+ private void clearBroadcastStateUnderLock(final String stateKey) {
+ final String broadcastKeyPrefix = stateKey + KEY_SEPARATOR;
+ pendingBroadcasts.keySet().removeIf(key ->
key.startsWith(broadcastKeyPrefix));
+ lastBroadcastTime.keySet().removeIf(key ->
key.startsWith(broadcastKeyPrefix));
}
/**
* Sends committed progress to all follower replicas of the given region.
Uses the partition cache
* to discover replica endpoints and skips the local DataNode.
*/
- private void doBroadcast(
+ void doBroadcast(
final String consumerGroupId,
final String topicName,
final ConsensusGroupId regionId,
@@ -1695,6 +1794,8 @@ public class ConsensusSubscriptionCommitManager {
*/
public static class ConsensusSubscriptionCommitState {
+ private final Object broadcastLifecycleLock = new Object();
+
private final String regionId;
private final SubscriptionConsensusProgress progress;
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitBroadcastTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitBroadcastTest.java
new file mode 100644
index 00000000000..a138fea012d
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitBroadcastTest.java
@@ -0,0 +1,288 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.subscription.broker.consensus;
+
+import org.apache.iotdb.commons.consensus.DataRegionId;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.rpc.subscription.payload.poll.RegionProgress;
+import org.apache.iotdb.rpc.subscription.payload.poll.WriterId;
+import org.apache.iotdb.rpc.subscription.payload.poll.WriterProgress;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.mockito.ArgumentCaptor;
+
+import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+public class ConsensusSubscriptionCommitBroadcastTest {
+
+ @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ private static final DataRegionId REGION = new DataRegionId(31);
+ private static final WriterId WRITER = new WriterId(REGION.toString(), 7);
+
+ private final ScheduledExecutorService executor =
mock(ScheduledExecutorService.class);
+ private final AtomicLong now = new AtomicLong(10_000L);
+ private ConsensusSubscriptionCommitManager manager;
+ private String originalSystemDir;
+
+ @Before
+ public void setUp() throws Exception {
+ originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .setSystemDir(temporaryFolder.newFolder().getAbsolutePath());
+ manager =
+ spy(
+ new ConsensusSubscriptionCommitManager(
+ (consumerGroupId, topicName, regionId) ->
+
ConsensusSubscriptionCommitManager.ConfigNodeProgressQueryResult.available(
+ new RegionProgress(Collections.emptyMap())),
+ executor,
+ now::get));
+ doNothing().when(manager).doBroadcast(eq("group"), eq("topic"),
eq(REGION), any(), any());
+ }
+
+ @After
+ public void tearDown() {
+ IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
+ }
+
+ @Test
+ public void testThrottledFinalAckIsBroadcastWithoutAnotherCommit() {
+ commit(WRITER, 1L);
+ final ArgumentCaptor<Runnable> tasks =
ArgumentCaptor.forClass(Runnable.class);
+ verify(executor).schedule(tasks.capture(), eq(0L),
eq(TimeUnit.MILLISECONDS));
+ tasks.getValue().run();
+ verify(manager).doBroadcast("group", "topic", REGION, progress(1L),
WRITER);
+
+ now.addAndGet(100L);
+ commit(WRITER, 2L);
+ commit(WRITER, 3L);
+ verify(executor).schedule(tasks.capture(), eq(4900L),
eq(TimeUnit.MILLISECONDS));
+ verify(executor, times(2)).schedule(any(Runnable.class), anyLong(),
eq(TimeUnit.MILLISECONDS));
+
+ now.addAndGet(4900L);
+ tasks.getValue().run();
+ verify(manager).doBroadcast("group", "topic", REGION, progress(3L),
WRITER);
+ verify(manager, never()).doBroadcast("group", "topic", REGION,
progress(2L), WRITER);
+ }
+
+ @Test
+ public void testWritersHaveIndependentBroadcasts() {
+ final WriterId otherWriter = new WriterId(REGION.toString(), 8);
+ commit(WRITER, 1L);
+ commit(otherWriter, 1L);
+ final ArgumentCaptor<Runnable> tasks =
ArgumentCaptor.forClass(Runnable.class);
+ verify(executor, times(2)).schedule(tasks.capture(), eq(0L),
eq(TimeUnit.MILLISECONDS));
+ tasks.getAllValues().forEach(Runnable::run);
+
+ verify(manager).doBroadcast("group", "topic", REGION, progress(1L),
WRITER);
+ verify(manager).doBroadcast("group", "topic", REGION, progress(1L),
otherWriter);
+ }
+
+ @Test
+ public void testRemovedStateCannotBroadcastIntoResubscribedState() {
+ commit(WRITER, 1L);
+ final ArgumentCaptor<Runnable> tasks =
ArgumentCaptor.forClass(Runnable.class);
+ verify(executor).schedule(tasks.capture(), eq(0L),
eq(TimeUnit.MILLISECONDS));
+ final Runnable oldTask = tasks.getValue();
+
+ manager.removeState("group", "topic", REGION);
+ commit(WRITER, 2L);
+ verify(executor, times(2)).schedule(tasks.capture(), eq(0L),
eq(TimeUnit.MILLISECONDS));
+ oldTask.run();
+ verify(manager, never()).doBroadcast("group", "topic", REGION,
progress(1L), WRITER);
+ tasks.getValue().run();
+ verify(manager).doBroadcast("group", "topic", REGION, progress(2L),
WRITER);
+ }
+
+ @Test
+ public void testResubscribedStateDoesNotInheritBroadcastThrottle() {
+ commit(WRITER, 1L);
+ final ArgumentCaptor<Runnable> tasks =
ArgumentCaptor.forClass(Runnable.class);
+ verify(executor).schedule(tasks.capture(), eq(0L),
eq(TimeUnit.MILLISECONDS));
+ tasks.getValue().run();
+
+ now.addAndGet(100L);
+ manager.removeState("group", "topic", REGION);
+ commit(WRITER, 2L);
+
+ verify(executor, times(2)).schedule(tasks.capture(), eq(0L),
eq(TimeUnit.MILLISECONDS));
+ tasks.getValue().run();
+ verify(manager).doBroadcast("group", "topic", REGION, progress(2L),
WRITER);
+ }
+
+ @Test
+ public void testRemovingStateWaitsForBroadcastAlreadyInProgress() throws
Exception {
+ final CountDownLatch broadcastStarted = new CountDownLatch(1);
+ final CountDownLatch releaseBroadcast = new CountDownLatch(1);
+ final CountDownLatch removeStarted = new CountDownLatch(1);
+ final AtomicReference<Throwable> broadcastFailure = new
AtomicReference<>();
+ final AtomicReference<Throwable> removeFailure = new AtomicReference<>();
+ doAnswer(
+ invocation -> {
+ broadcastStarted.countDown();
+ assertTrue(releaseBroadcast.await(30L, TimeUnit.SECONDS));
+ return null;
+ })
+ .when(manager)
+ .doBroadcast(eq("group"), eq("topic"), eq(REGION), any(), eq(WRITER));
+ doAnswer(
+ invocation -> {
+ removeStarted.countDown();
+ return invocation.callRealMethod();
+ })
+ .when(manager)
+ .removeState("group", "topic", REGION);
+
+ commit(WRITER, 1L);
+ final ArgumentCaptor<Runnable> tasks =
ArgumentCaptor.forClass(Runnable.class);
+ verify(executor).schedule(tasks.capture(), eq(0L),
eq(TimeUnit.MILLISECONDS));
+ final Thread broadcastThread =
+ new Thread(
+ () -> {
+ try {
+ tasks.getValue().run();
+ } catch (final Throwable t) {
+ broadcastFailure.set(t);
+ }
+ },
+ "broadcast-test");
+ final CountDownLatch removeFinished = new CountDownLatch(1);
+ final Thread removeThread =
+ new Thread(
+ () -> {
+ try {
+ manager.removeState("group", "topic", REGION);
+ } catch (final Throwable t) {
+ removeFailure.set(t);
+ } finally {
+ removeFinished.countDown();
+ }
+ },
+ "remove-state-test");
+ try {
+ broadcastThread.start();
+ assertTrue(broadcastStarted.await(5L, TimeUnit.SECONDS));
+ removeThread.start();
+ assertTrue(removeStarted.await(5L, TimeUnit.SECONDS));
+ assertTrue(
+ waitForThreadBlockedInMethod(
+ removeThread,
+ ConsensusSubscriptionCommitManager.class.getName(),
+ "removeCommitState",
+ 5L,
+ TimeUnit.SECONDS));
+ } finally {
+ releaseBroadcast.countDown();
+ }
+ assertTrue(removeFinished.await(5L, TimeUnit.SECONDS));
+ broadcastThread.join(5_000L);
+ removeThread.join(5_000L);
+ assertNull(broadcastFailure.get());
+ assertNull(removeFailure.get());
+ verify(manager).doBroadcast("group", "topic", REGION, progress(1L),
WRITER);
+ }
+
+ private static boolean waitForThreadBlockedInMethod(
+ final Thread thread,
+ final String className,
+ final String methodName,
+ final long timeout,
+ final TimeUnit unit)
+ throws InterruptedException {
+ final long deadline = System.nanoTime() + unit.toNanos(timeout);
+ while (System.nanoTime() < deadline) {
+ if (thread.getState() == Thread.State.BLOCKED) {
+ for (final StackTraceElement frame : thread.getStackTrace()) {
+ if (className.equals(frame.getClassName()) &&
methodName.equals(frame.getMethodName())) {
+ return true;
+ }
+ }
+ }
+ Thread.sleep(1L);
+ }
+ return false;
+ }
+
+ @Test
+ public void testRejectedScheduleDoesNotSuppressLaterBroadcast() {
+ final AtomicBoolean rejectFirstSchedule = new AtomicBoolean(true);
+ final AtomicReference<Runnable> scheduledTask = new AtomicReference<>();
+ doAnswer(
+ invocation -> {
+ if (rejectFirstSchedule.getAndSet(false)) {
+ throw new RejectedExecutionException();
+ }
+ scheduledTask.set(invocation.getArgument(0));
+ return null;
+ })
+ .when(executor)
+ .schedule(any(Runnable.class), anyLong(), eq(TimeUnit.MILLISECONDS));
+
+ try {
+ commit(WRITER, 1L);
+ throw new AssertionError("Expected the first scheduled broadcast to be
rejected");
+ } catch (final RejectedExecutionException expected) {
+ // The executor can reject during shutdown; later attempts must not
remain marked pending.
+ }
+
+ commit(WRITER, 2L);
+ assertNotNull(scheduledTask.get());
+ scheduledTask.get().run();
+
+ verify(executor, times(2)).schedule(any(Runnable.class), eq(0L),
eq(TimeUnit.MILLISECONDS));
+ verify(manager).doBroadcast("group", "topic", REGION, progress(2L),
WRITER);
+ }
+
+ private void commit(final WriterId writerId, final long sequence) {
+ manager.recordMapping("group", "topic", REGION, writerId,
progress(sequence));
+ assertTrue(manager.commit("group", "topic", REGION, writerId,
progress(sequence)));
+ }
+
+ private static WriterProgress progress(final long sequence) {
+ return new WriterProgress(100L + sequence, sequence);
+ }
+}
