Subscription: implement IoTConsensus-based subscription by DanielWang2035 · Pull Request #17238 · apache/iotdb

DanielWang2035 · 2026-03-02T02:29:23Z

[Draft] Subscription: implement IoTConsensus-based subscription

Description

This PR introduces a new subscription topic mode, mode=consensus, for clusters whose data_region_consensus_protocol_class is IoTConsensus.

For consensus-mode topics, the subscription runtime no longer creates a Pipe. Instead, it reads directly from IoTConsensus WAL, combines realtime in-memory handoff with WAL replay, converts WAL entries into subscription record batches, and serves them through the existing subscription client APIs. Commit progress is tracked per region/writer to support recovery and replay. Existing live and snapshot topics keep their current Pipe-based behavior unchanged.

This PR now includes the full end-to-end path:

DataNode runtime for realtime delivery, WAL replay, prefetching, commit tracking, and WAL-retention coordination
ConfigNode-side topic validation, commit-progress persistence/sync, and leader-change handling
Client-side consensus progress and seek support: seekToBeginning, seekToEnd, and seekAfter(TopicProgress)
Watermark events and recovery-aware polling/commit flow
Both tree-model and table-model topics, including database/table filtering and table column filtering
Consumer-group ownership and auto-binding when a subscription is created before matching DataRegions exist

Current Scope

Only available for mode=consensus
Only supported when data_region_consensus_protocol_class=IoTConsensus
Only supports record delivery (format=SubscriptionRecordHandler) for now; TsFile delivery still uses the existing Pipe path
Supports consensus-topic WAL retention controls via retention.bytes and retention.ms

Tests

Added UT/IT coverage for:

basic realtime delivery and subscribe-before-region scenarios
primitive, aligned, tree-model, and table-model data conversion
consumer-group fan-out and no-duplicate delivery within one group
checkpoint recovery and explicit seek replay
table database/table filtering and table column filtering
WAL reader, iterator, metadata, commit-state, and converter behavior

Copilot

Pull request overview

Implements a new consensus-based subscription delivery path for live Tablet-format topics by reading directly from IoTConsensus WAL (bypassing the Pipe framework), and routes subscription polling/commit through either the existing Pipe broker or the new consensus broker based on topic mode/format.

Changes:

Add consensus subscription components (setup handler, prefetching queue, WAL→Tablet converter, commit/progress tracking, consensus broker).
Update subscription agents/brokers to route poll/commit to pipe vs. consensus queues and manage consensus queue lifecycle.
Extend IoTConsensus write path + WAL retention logic to feed subscription queues and attempt to pin WAL via subscription progress.

Reviewed changes

Copilot reviewed 20 out of 20 changed files in this pull request and generated 7 comments.

Show a summary per file

File	Description
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMeta.java	Adds helpers to detect newly subscribed topics and expose subscribed topic names.
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/config/SubscriptionConfig.java	Flips subscription enabled flag to always-on.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/SubscriptionEvent.java	Exposes nack counter for recycling logic.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/SubscriptionConsensusProgress.java	New POJO for persisted consensus subscription progress.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java	Binds/unbinds consensus queues per topic/region and registers new-region callback.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java	New per-(group,topic,region) commit/progress persistence.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java	Core hybrid pending-queue + WAL-reader prefetch loop producing SubscriptionEvents.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java	Converts InsertNode variants into Tablet payloads with pattern filtering.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionBroker.java	Implements new ISubscriptionBroker interface and adds queue helpers.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/ISubscriptionBroker.java	New broker abstraction shared by pipe and consensus brokers.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/ConsensusSubscriptionBroker.java	New broker coordinating multiple region queues per topic.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgent.java	Detects newly subscribed topics and triggers consensus setup/teardown.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionBrokerAgent.java	Routes poll/commit/prefetch across pipe and consensus brokers.
iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java	Adds subscription queue registration + WAL safe-delete adjustment + write-path offering.
iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java	Adds static callback hook for new peer creation to auto-bind subscription queues.
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedure.java	Skips pipe removal for consensus-based topics.
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/CreateSubscriptionProcedure.java	Skips pipe creation for consensus-based topics and serializes that topic set.
example/session/src/main/java/org/apache/iotdb/ConsensusSubscriptionTest.java	Adds manual (non-IT) tests for tree-model consensus subscription.
example/session/src/main/java/org/apache/iotdb/ConsensusSubscriptionTableTest.java	Adds manual (non-IT) tests for table-model consensus subscription.
example/session/pom.xml	Overrides compiler level for the session example module.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2026-03-02T02:34:59Z

+    }
+    try (final FileInputStream fis = new FileInputStream(file)) {
+      final byte[] bytes = new byte[(int) file.length()];
+      fis.read(bytes);


tryRecover reads the progress file with a single fis.read(bytes) call, which is not guaranteed to fill the buffer. This can lead to partial reads and corrupted/failed deserialization. Use a readFully loop / DataInputStream.readFully, and validate that the expected number of bytes was read before deserializing.

Suggested change

fis.read(bytes);

int offset = 0;

while (offset < bytes.length) {

final int bytesRead = fis.read(bytes, offset, bytes.length - offset);

if (bytesRead < 0) {

break;

}

offset += bytesRead;

}

if (offset != bytes.length) {

LOGGER.warn(

"Failed to fully read consensus subscription progress from {}: expected {} bytes, read {} bytes",

file,

bytes.length,

offset);

return null;

}

Copilot · 2026-03-02T02:34:59Z

+      final long startSearchIndex = serverImpl.getSearchIndex() + 1;
+


startSearchIndex is always initialized as serverImpl.getSearchIndex() + 1, which ignores any persisted subscription progress in ConsensusSubscriptionCommitManager. After a DataNode restart (or queue rebind), this will skip unconsumed WAL entries and break the at-least-once / recovery semantics described in the PR. The start index should be derived from the recovered committed search index when state exists (e.g., max(committed+1, subscribeStart) depending on desired semantics).

Suggested change

final long startSearchIndex = serverImpl.getSearchIndex() + 1;

long startSearchIndex = serverImpl.getSearchIndex() + 1;

// If we have recovered a committed search index for this group, advance the

// startSearchIndex to at least committed + 1 to honor at-least-once semantics.

if (commitManager != null) {

try {

final long committedSearchIndex = commitManager.getCommittedSearchIndex(groupId);

if (committedSearchIndex >= 0) {

startSearchIndex = Math.max(startSearchIndex, committedSearchIndex + 1);

}

} catch (Exception e) {

LOGGER.warn(

"Failed to load committed search index for consensus group {} when binding "

+ "subscription prefetching queue, fallback to startSearchIndex={}",

groupId,

startSearchIndex,

e);

}

}

Copilot · 2026-03-02T02:34:59Z

        consumerGroupId,
        (id, broker) -> {
          if (Objects.isNull(broker)) {
+            dropped.set(true);


dropBroker sets dropped=true when the pipe broker entry is already null. This causes dropBroker to return true even when nothing was removed, and can also mask failure to drop a non-empty consensus broker (pipe broker null => dropped=true, consensus broker kept). Only mark dropped true when an existing broker entry is actually removed.

Suggested change

dropped.set(true);

// No existing broker to drop

Copilot · 2026-03-02T02:35:00Z

+        eventsToPoll.add(event);
+        totalSize += currentSize;
+
+        if (totalSize + currentSize > maxBytes) {


The maxBytes limiting logic is incorrect: the event is added and totalSize is incremented, then the code checks if (totalSize + currentSize > maxBytes) which double-counts currentSize and can still return a batch exceeding maxBytes. Check the size before adding the event (or adjust the condition) and decide whether to stop/push back the event when it would exceed the limit.

Suggested change

if (totalSize + currentSize > maxBytes) {

if (totalSize > maxBytes) {

Copilot · 2026-03-02T02:35:00Z

+    // Compute the minimum search index that subscription consumers still need.
+    // WAL entries at or after this index must be preserved.
+    long minSubscriptionIndex = Long.MAX_VALUE;
+    for (final LongSupplier supplier : subscriptionSyncIndexSuppliers) {
+      minSubscriptionIndex = Math.min(minSubscriptionIndex, supplier.getAsLong());
+    }
+
+    if (configuration.size() == 1 && subscriptionSyncIndexSuppliers.isEmpty()) {
+      // Single replica, no subscription consumers => delete all WAL freely
      consensusReqReader.setSafelyDeletedSearchIndex(Long.MAX_VALUE);
    } else {
-      consensusReqReader.setSafelyDeletedSearchIndex(getMinFlushedSyncIndex());
+      // min(replication progress, subscription progress) — preserve WAL for both
+      final long replicationIndex =
+          configuration.size() > 1 ? getMinFlushedSyncIndex() : Long.MAX_VALUE;
+      consensusReqReader.setSafelyDeletedSearchIndex(
+          Math.min(replicationIndex, minSubscriptionIndex));


checkAndUpdateSafeDeletedSearchIndex() now computes a subscription-aware safe delete index, but other code paths (e.g., LogDispatcher periodically calling reader.setSafelyDeletedSearchIndex(impl.getMinFlushedSyncIndex())) can overwrite it with a replication-only value. That can allow WAL deletion past the slowest subscription consumer, causing data loss. Consider centralizing all updates to safelyDeletedSearchIndex through a single subscription-aware method (or make getMinFlushedSyncIndex() incorporate subscription progress).

Copilot · 2026-03-02T02:35:00Z

+    <!-- TODO: remove below -->
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <source>11</source>
+                    <target>11</target>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>


This module overrides the example parent’s compiler level (example/pom.xml sets source/target=8) to Java 11. That can break building examples under the project’s expected Java version and makes the module inconsistent with other examples. Prefer inheriting the parent compiler settings (or updating the parent if the whole examples tree should move to 11) instead of overriding in this module.

Suggested change



<build>

<plugins>

<plugin>

<groupId>org.apache.maven.plugins</groupId>

<artifactId>maven-compiler-plugin</artifactId>

<configuration>

<source>11</source>

<target>11</target>

</configuration>

</plugin>

</plugins>

</build>

Caideyipi · 2026-03-12T02:30:27Z

  private long subscriptionMetaSyncerInitialSyncDelayMinutes = 3;
  private long subscriptionMetaSyncerSyncIntervalMinutes = 3;

+  private int subscriptionConsensusBatchMaxDelayInMs = 50;


May any of them support hot-reloading?

All configs here support hot-reloading, but only affect newly created topics and won't affect existing topics.

Caideyipi · 2026-03-12T02:50:06Z

+  private int subscriptionConsensusBatchMaxTabletCount = 64;
+  private int subscriptionConsensusBatchMaxWalEntries = 128;
+
+  private long subscriptionConsensusWalRetentionSizeInBytes = 512 * MB;


May consider "throttleThreshold" as default value or directly use that?

The scope of throttleThreshold is all WALNodes of one DataNode, but the scope of retention size is per region. As the number of region changes, we can't directly use (throttleThreshold / region count) for it. The default value of retention policy can be discussed later.

Caideyipi · 2026-03-12T04:29:18Z

+        final long totalWalSize = consensusReqReader.getTotalSize();
+        if (totalWalSize <= retentionSizeLimit) {
+          // WAL size is within retention limit — preserve all WAL for subscribers
+          subscriptionRetentionBound = ConsensusReqReader.DEFAULT_SAFELY_DELETED_SEARCH_INDEX;


Seemingly a little bit coarse?

The wal clean is either "all" or "not", does not have a bound like "only reserve the retention size amount of data"

The reserving logic does not consider the commit indexes of subscriptions

Fixed. 2. This is intentional: similar to Kafka, once data falls outside the configured retention window we do not keep it just because a consumer has not committed it yet.

Caideyipi · 2026-03-12T06:09:44Z

+              .getStringOrDefault(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_DEFAULT_VALUE);
+      final boolean isConsensusBasedTopic =
+          TopicConstant.MODE_LIVE_VALUE.equalsIgnoreCase(topicMode)
+              && !TopicConstant.FORMAT_TS_FILE_HANDLER_VALUE.equalsIgnoreCase(topicFormat);


Shall MODE_LIVE + TS_FILE_HANDLER throw an exception here?

Now the new mode explicitly uses parameter TopicConstant.MODE_KEY (MODE_SNAPSHOT_VALUE / MODE_LIVE_VALUE / MODE_CONSENSUS_VALUE). The new consensus mode will throw an exception if format is not RECORD.

…jacent tree-model rows with the same emitted device/schema into one Tablet

Resolve consensus subscription conflicts and address review comments.

VGalaxies · 2026-06-04T05:24:41Z

CI note: the two failed Simple (17) jobs are failing during Set up job before checkout/tests. After rerun they still stop while downloading actions/checkout@v5 on self-hosted runners (iotdb-1 / iotdb-5) with tar: command not found, so this looks like runner environment infrastructure rather than a code failure.

* Subscription: implement IoTConsensus-based subscription * fix some issues * support seek and WAL retention * refactor * refactor * fix * fix * fix part1 * fix part2 * fix part3.1 * fix part3.2 * fix part4.1 * fix seek * fix test * refactor thread model * add time based retention and column pattern * clean old epoch * add config for consensus subscription * fix comments * remove some tests * add check for non-IoTConsensus * refactor IT * fix review * fix UT & IT * fix comments * fix comments * remove writerEpoch field & unify the order of fields * fix test * spotless * fix IT * remove special logic of CONTINUOUS_SAME_SEARCH_INDEX_SEPARATOR_NODE * make partial commit responses explicit * keep processor-buffered messages pinned until release * make manual close safe with processor buffers * tolerate non-owning providers during consensus seek * fix UT * fix IT * refactor progress persist * spotless * fix some issues * fix some issues * extract some logic into broker interface * fix ColumnAlignProcessor * simplify some concepts * fix some naming and comments * fix some issues * fix some issues * rename * Avoided persisting writer.meta on every successful write & Grouped adjacent tree-model rows with the same emitted device/schema into one Tablet * rename * fix * fix * Refine subscription broker routing * Fix consensus subscription replay dedup --------- Co-authored-by: VGalaxies <vgalaxies@apache.org>

Subscription: implement IoTConsensus-based subscription

6252d13

DanielWang2035 requested a review from Copilot March 2, 2026 02:29

Copilot started reviewing on behalf of DanielWang2035 March 2, 2026 02:29 View session

Copilot AI reviewed Mar 2, 2026

View reviewed changes

DanielWang2035 added 2 commits March 3, 2026 18:59

fix some issues

36e3491

support seek and WAL retention

8d2ba7d

Caideyipi reviewed Mar 12, 2026

View reviewed changes

Comment thread iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java Outdated

Comment thread ...e/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionBrokerAgent.java Outdated

DanielWang2035 added 19 commits March 16, 2026 10:48

refactor

0e3b768

Merge branch 'master' into consensus-subscription

b3c39ff

refactor

932617e

fix

8ba8165

fix

658f0a6

fix part1

ae9d722

fix part2

e3eedcf

fix part3.1

a861944

fix part3.2

5a4f7da

fix part4.1

8e53fc6

merge master

32e6d11

fix seek

2794d31

fix test

6f98bd6

refactor thread model

0fbc551

add time based retention and column pattern

774427b

clean old epoch

db92a35

add config for consensus subscription

6e86c0b

fix comments

8b275b1

merge master

c29f262

DanielWang2035 added 21 commits May 15, 2026 13:42

tolerate non-owning providers during consensus seek

43cd5be

Merge branch 'master' into consensus-subscription

ada74c9

fix UT

739e308

fix IT

3d2dbf7

refactor progress persist

de3d167

merge master

ad1a609

spotless

94faf00

fix some issues

38fcc6b

fix some issues

f393fa2

extract some logic into broker interface

c616e59

fix ColumnAlignProcessor

f42f19e

simplify some concepts

f361474

fix some naming and comments

9da9880

fix some issues

b6efcb1

fix some issues

4968e1d

Merge branch 'master' into consensus-subscription

ef45738

rename

3103685

Avoided persisting writer.meta on every successful write & Grouped ad…

410aa89

…jacent tree-model rows with the same emitted device/schema into one Tablet

rename

2cda962

fix

becf7f3

fix

566ea99

jt2594838 reviewed Jun 3, 2026

View reviewed changes

VGalaxies added 2 commits June 3, 2026 06:42

Merge upstream master into consensus-subscription

5ea4280

Resolve consensus subscription conflicts and address review comments.

Refine subscription broker routing

aa6810e

jt2594838 reviewed Jun 4, 2026

View reviewed changes

Fix consensus subscription replay dedup

75bacb1

jt2594838 approved these changes Jun 8, 2026

View reviewed changes

jt2594838 merged commit 0987f6a into apache:master Jun 9, 2026
40 of 46 checks passed

-      fis.read(bytes);
+      int offset = 0;
+      while (offset < bytes.length) {
+        final int bytesRead = fis.read(bytes, offset, bytes.length - offset);
+        if (bytesRead < 0) {
+          break;
+        }
+        offset += bytesRead;
+      }
+      if (offset != bytes.length) {
+        LOGGER.warn(
+            "Failed to fully read consensus subscription progress from {}: expected {} bytes, read {} bytes",
+            file,
+            bytes.length,
+            offset);
+        return null;
+      }

		final long startSearchIndex = serverImpl.getSearchIndex() + 1;

-      final long startSearchIndex = serverImpl.getSearchIndex() + 1;
+      long startSearchIndex = serverImpl.getSearchIndex() + 1;
+      // If we have recovered a committed search index for this group, advance the
+      // startSearchIndex to at least committed + 1 to honor at-least-once semantics.
+      if (commitManager != null) {
+        try {
+          final long committedSearchIndex = commitManager.getCommittedSearchIndex(groupId);
+          if (committedSearchIndex >= 0) {
+            startSearchIndex = Math.max(startSearchIndex, committedSearchIndex + 1);
+          }
+        } catch (Exception e) {
+          LOGGER.warn(
+              "Failed to load committed search index for consensus group {} when binding "
+                  + "subscription prefetching queue, fallback to startSearchIndex={}",
+              groupId,
+              startSearchIndex,
+              e);
+        }
+      }

	if (totalSize + currentSize > maxBytes) {
	if (totalSize > maxBytes) {

Uh oh!

Conversation

DanielWang2035 commented Mar 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

[Draft] Subscription: implement IoTConsensus-based subscription

Description

Current Scope

Tests

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI Mar 2, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Mar 2, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Mar 2, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Mar 2, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Mar 2, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Mar 2, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

VGalaxies commented Jun 4, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

DanielWang2035 commented Mar 2, 2026 •

edited

Loading