apache · arimu1 · Sep 13, 2026 · Sep 13, 2026
diff --git a/CHANGES.md b/CHANGES.md
@@ -98,6 +98,7 @@
 * (Prism) Self-checkpointing splittable DoFns now resume after their requested delay instead of immediately, so polling SDFs no longer busy-spin ([#39848](https://github.com/apache/beam/issues/39848)).
 * (Java) MongoDbIO read splitting now preserves non-ObjectId `_id` types (e.g. string ids) instead of failing to parse the generated range filters ([#39900](https://github.com/apache/beam/issues/39900)).
 * (Go) Fixed GCS glob matching silently dropping objects when the glob pattern contains multi-byte characters ([#39969](https://github.com/apache/beam/issues/39969)).
+* (Java) BigtableIO.readChangeStream() with RESUME_OR_NEW no longer starts duplicate consumers when InitializeDoFn is re-executed within the same pipeline run ([#39970](https://github.com/apache/beam/issues/39970)).
 
 ## Security Fixes
 

diff --git a/...o/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java b/...o/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java
@@ -2553,8 +2553,9 @@ public PCollection<KV<ByteString, ChangeStreamMutation>> expand(PBegin input) {
         daoFactory.close();
       }
 
+      String pipelineRunId = UniqueIdGenerator.generateRowKeyPrefix();
       InitializeDoFn initializeDoFn =
-          new InitializeDoFn(daoFactory, startTime, existingPipelineOptions);
+          new InitializeDoFn(daoFactory, startTime, existingPipelineOptions, pipelineRunId);
       DetectNewPartitionsDoFn detectNewPartitionsDoFn =
           new DetectNewPartitionsDoFn(getEndTime(), actionFactory, daoFactory, metrics);
       ReadChangeStreamPartitionDoFn readChangeStreamPartitionDoFn =

diff --git a/...ain/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dao/MetadataTableAdminDao.java b/...ain/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dao/MetadataTableAdminDao.java
@@ -52,6 +52,7 @@ public class MetadataTableAdminDao {
   public static final String CF_VERSION = "version";
   public static final String CF_SHOULD_DELETE = "should_delete";
   public static final String QUALIFIER_DEFAULT = "latest";
+  public static final String QUALIFIER_PIPELINE_RUN_ID = "pipeline_run_id";
   public static final ImmutableList<String> COLUMN_FAMILIES =
       ImmutableList.of(
           CF_INITIAL_TOKEN,

diff --git a/...src/main/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dao/MetadataTableDao.java b/...src/main/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dao/MetadataTableDao.java
@@ -207,6 +207,45 @@ public ByteString convertPartitionToNewPartitionRowKey(ByteStringRange partition
     return new DetectNewPartitionsState(watermark, timestamp);
   }
 
+  /**
+   * Read the pipeline run id recorded when {@link
+   * org.apache.beam.sdk.io.gcp.bigtable.changestreams.dofn.InitializeDoFn} last completed for an
+   * active change stream pipeline.
+   */
+  public @Nullable String readPipelineRunId() {
+    Filter pipelineRunIdFilter =
+        FILTERS
+            .chain()
+            .filter(FILTERS.family().exactMatch(MetadataTableAdminDao.CF_VERSION))
+            .filter(FILTERS.qualifier().exactMatch(MetadataTableAdminDao.QUALIFIER_PIPELINE_RUN_ID))
+            .filter(FILTERS.limit().cellsPerColumn(1));
+    Row row = dataClient.readRow(tableId, getFullDetectNewPartition(), pipelineRunIdFilter);
+    if (row == null
+        || row.getCells(
+                MetadataTableAdminDao.CF_VERSION, MetadataTableAdminDao.QUALIFIER_PIPELINE_RUN_ID)
+            .isEmpty()) {
+      return null;
+    }
+    return row.getCells(
+            MetadataTableAdminDao.CF_VERSION, MetadataTableAdminDao.QUALIFIER_PIPELINE_RUN_ID)
+        .get(0)
+        .getValue()
+        .toStringUtf8();
+  }
+
+  /** Record the pipeline run id for the current change stream pipeline execution. */
+  public void writePipelineRunId(String pipelineRunId) {
+    long nowMicros = Instant.now().getMillis() * 1000L;
+    RowMutation rowMutation =
+        RowMutation.create(tableId, getFullDetectNewPartition())
+            .setCell(
+                MetadataTableAdminDao.CF_VERSION,
+                MetadataTableAdminDao.QUALIFIER_PIPELINE_RUN_ID,
+                nowMicros,
+                pipelineRunId);
+    mutateRowWithHardTimeout(rowMutation);
+  }
+
   /**
    * Returns all the new partitions resulting from splits and merges waiting to be streamed
    * including ones marked for deletion.

diff --git a/.../src/main/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dofn/InitializeDoFn.java b/.../src/main/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dofn/InitializeDoFn.java
@@ -40,12 +40,17 @@ public class InitializeDoFn extends DoFn<byte[], InitialPipelineState> implement
   private final DaoFactory daoFactory;
   private Instant startTime;
   private final ExistingPipelineOptions existingPipelineOptions;
+  private final String pipelineRunId;
 
   public InitializeDoFn(
-      DaoFactory daoFactory, Instant startTime, ExistingPipelineOptions existingPipelineOptions) {
+      DaoFactory daoFactory,
+      Instant startTime,
+      ExistingPipelineOptions existingPipelineOptions,
+      String pipelineRunId) {
     this.daoFactory = daoFactory;
     this.startTime = startTime;
     this.existingPipelineOptions = existingPipelineOptions;
+    this.pipelineRunId = pipelineRunId;
   }
 
   @ProcessElement
@@ -54,6 +59,14 @@ public void processElement(OutputReceiver<InitialPipelineState> receiver) throws
     LOG.info("{}", daoFactory.getMetadataTableDebugString());
     LOG.info("ChangeStreamName: {}", daoFactory.getChangeStreamName());
 
+    String storedPipelineRunId = daoFactory.getMetadataTableDao().readPipelineRunId();
+    if (storedPipelineRunId != null && storedPipelineRunId.equals(pipelineRunId)) {
+      LOG.info(
+          "Initialize already completed for pipeline run {}, skipping duplicate initialization",
+          pipelineRunId);
+      return;
+    }
+
     boolean resume = false;
     DetectNewPartitionsState detectNewPartitionsState =
         daoFactory.getMetadataTableDao().readDetectNewPartitionsState();
@@ -101,6 +114,7 @@ public void processElement(OutputReceiver<InitialPipelineState> receiver) throws
         // terminate pipeline
         return;
     }
+    daoFactory.getMetadataTableDao().writePipelineRunId(pipelineRunId);
     daoFactory.getMetadataTableDao().writeDetectNewPartitionVersion();
     receiver.output(new InitialPipelineState(startTime, resume));
   }

diff --git a/.../test/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dofn/InitializeDoFnTest.java b/.../test/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dofn/InitializeDoFnTest.java
@@ -34,6 +34,7 @@
 import com.google.cloud.bigtable.emulator.v2.BigtableEmulatorRule;
 import com.google.protobuf.ByteString;
 import java.io.IOException;
+import java.util.UUID;
 import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO;
 import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO.ExistingPipelineOptions;
 import org.apache.beam.sdk.io.gcp.bigtable.changestreams.dao.DaoFactory;
@@ -61,6 +62,7 @@ public class InitializeDoFnTest {
   private transient MetadataTableDao metadataTableDao;
   @Mock private DoFn.OutputReceiver<InitialPipelineState> outputReceiver;
   private final String tableId = "table";
+  private static final String PIPELINE_RUN_ID = "test-pipeline-run";
 
   private static BigtableDataClient dataClient;
   private static BigtableTableAdminClient adminClient;
@@ -83,7 +85,7 @@ public static void beforeClass() throws IOException {
 
   @Before
   public void setUp() throws IOException {
-    String changeStreamName = "changeStreamName";
+    String changeStreamName = "changeStreamName-" + UUID.randomUUID();
     metadataTableAdminDao =
         spy(new MetadataTableAdminDao(adminClient, null, changeStreamName, tableId));
     metadataTableAdminDao.createMetadataTable();
@@ -99,7 +101,10 @@ public void testInitializeDefault() throws IOException {
     Instant startTime = Instant.now();
     InitializeDoFn initializeDoFn =
         new InitializeDoFn(
-            daoFactory, startTime, BigtableIO.ExistingPipelineOptions.FAIL_IF_EXISTS);
+            daoFactory,
+            startTime,
+            BigtableIO.ExistingPipelineOptions.FAIL_IF_EXISTS,
+            PIPELINE_RUN_ID);
     initializeDoFn.processElement(outputReceiver);
     verify(outputReceiver, times(1)).output(new InitialPipelineState(startTime, false));
   }
@@ -110,7 +115,10 @@ public void testInitializeStopWithExistingPipeline() throws IOException {
     Instant startTime = Instant.now();
     InitializeDoFn initializeDoFn =
         new InitializeDoFn(
-            daoFactory, startTime, BigtableIO.ExistingPipelineOptions.FAIL_IF_EXISTS);
+            daoFactory,
+            startTime,
+            BigtableIO.ExistingPipelineOptions.FAIL_IF_EXISTS,
+            PIPELINE_RUN_ID);
     initializeDoFn.processElement(outputReceiver);
     verify(outputReceiver, never()).output(any());
   }
@@ -134,7 +142,10 @@ public void testInitializeStopWithoutDNP() throws IOException {
     Instant startTime = Instant.now();
     InitializeDoFn initializeDoFn =
         new InitializeDoFn(
-            daoFactory, startTime, BigtableIO.ExistingPipelineOptions.FAIL_IF_EXISTS);
+            daoFactory,
+            startTime,
+            BigtableIO.ExistingPipelineOptions.FAIL_IF_EXISTS,
+            PIPELINE_RUN_ID);
     initializeDoFn.processElement(outputReceiver);
     verify(outputReceiver, times(1)).output(new InitialPipelineState(startTime, false));
     assertNull(dataClient.readRow(tableId, metadataTableAdminDao.getChangeStreamNamePrefix()));
@@ -156,7 +167,11 @@ public void testInitializeResumeWithoutDNP() throws IOException {
                 123L));
     Instant startTime = Instant.now();
     InitializeDoFn initializeDoFn =
-        new InitializeDoFn(daoFactory, startTime, BigtableIO.ExistingPipelineOptions.RESUME_OR_NEW);
+        new InitializeDoFn(
+            daoFactory,
+            startTime,
+            BigtableIO.ExistingPipelineOptions.RESUME_OR_NEW,
+            PIPELINE_RUN_ID);
     initializeDoFn.processElement(outputReceiver);
     // We want to resume but there's no DNP row, so we resume from the startTime provided.
     verify(outputReceiver, times(1)).output(new InitialPipelineState(startTime, false));
@@ -166,6 +181,7 @@ public void testInitializeResumeWithoutDNP() throws IOException {
   public void testInitializeResumeWithDNP() throws IOException {
     Instant resumeTime = Instant.now().minus(Duration.standardSeconds(10000));
     metadataTableDao.updateDetectNewPartitionWatermark(resumeTime);
+    metadataTableDao.writePipelineRunId("previous-pipeline-run");
     long nowMicros = Instant.now().getMillis() * 1000L;
     dataClient.mutateRow(
         RowMutation.create(
@@ -180,7 +196,11 @@ public void testInitializeResumeWithDNP() throws IOException {
                 123L));
     Instant startTime = Instant.now();
     InitializeDoFn initializeDoFn =
-        new InitializeDoFn(daoFactory, startTime, BigtableIO.ExistingPipelineOptions.RESUME_OR_NEW);
+        new InitializeDoFn(
+            daoFactory,
+            startTime,
+            BigtableIO.ExistingPipelineOptions.RESUME_OR_NEW,
+            PIPELINE_RUN_ID);
     initializeDoFn.processElement(outputReceiver);
     verify(outputReceiver, times(1)).output(new InitialPipelineState(resumeTime, true));
     assertNull(dataClient.readRow(tableId, metadataTableAdminDao.getChangeStreamNamePrefix()));
@@ -202,7 +222,8 @@ public void testInitializeSkipCleanupWithoutDNP() throws IOException {
                 123L));
     Instant startTime = Instant.now();
     InitializeDoFn initializeDoFn =
-        new InitializeDoFn(daoFactory, startTime, ExistingPipelineOptions.SKIP_CLEANUP);
+        new InitializeDoFn(
+            daoFactory, startTime, ExistingPipelineOptions.SKIP_CLEANUP, PIPELINE_RUN_ID);
     initializeDoFn.processElement(outputReceiver);
     // Skip cleanup will always resume from startTime
     verify(outputReceiver, times(1)).output(new InitialPipelineState(startTime, false));
@@ -228,11 +249,28 @@ public void testInitializeSkipCleanupWithDNP() throws IOException {
                 123L));
     Instant startTime = Instant.now();
     InitializeDoFn initializeDoFn =
-        new InitializeDoFn(daoFactory, startTime, ExistingPipelineOptions.SKIP_CLEANUP);
+        new InitializeDoFn(
+            daoFactory, startTime, ExistingPipelineOptions.SKIP_CLEANUP, PIPELINE_RUN_ID);
     initializeDoFn.processElement(outputReceiver);
     // We don't want the pipeline to resume to avoid duplicates
     verify(outputReceiver, never()).output(any());
     // Existing metadata shouldn't be cleaned up
     assertNotNull(dataClient.readRow(tableId, metadataRowKey));
   }
+
+  @Test
+  public void testInitializeSkipsDuplicatePipelineRun() throws IOException {
+    Instant resumeTime = Instant.now().minus(Duration.standardSeconds(10000));
+    metadataTableDao.updateDetectNewPartitionWatermark(resumeTime);
+    metadataTableDao.writePipelineRunId(PIPELINE_RUN_ID);
+    Instant startTime = Instant.now();
+    InitializeDoFn initializeDoFn =
+        new InitializeDoFn(
+            daoFactory,
+            startTime,
+            BigtableIO.ExistingPipelineOptions.RESUME_OR_NEW,
+            PIPELINE_RUN_ID);
+    initializeDoFn.processElement(outputReceiver);
+    verify(outputReceiver, never()).output(any());
+  }
 }