diff --git a/CHANGES.md b/CHANGES.md index f0d5d06b9d9f..704a3241733a 100644 --- 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/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java index 02236469e325..9b84ec48fbd1 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java @@ -2553,8 +2553,9 @@ public PCollection> 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/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dao/MetadataTableAdminDao.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dao/MetadataTableAdminDao.java index 12e9ff29ea19..a5dffa412a8a 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dao/MetadataTableAdminDao.java +++ b/sdks/java/io/google-cloud-platform/src/main/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 COLUMN_FAMILIES = ImmutableList.of( CF_INITIAL_TOKEN, diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dao/MetadataTableDao.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dao/MetadataTableDao.java index 83115297764b..9480080f2b8c 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dao/MetadataTableDao.java +++ b/sdks/java/io/google-cloud-platform/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/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dofn/InitializeDoFn.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dofn/InitializeDoFn.java index 1bf2b5788592..6df746bdcd11 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dofn/InitializeDoFn.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dofn/InitializeDoFn.java @@ -40,12 +40,17 @@ public class InitializeDoFn extends DoFn 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 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 receiver) throws // terminate pipeline return; } + daoFactory.getMetadataTableDao().writePipelineRunId(pipelineRunId); daoFactory.getMetadataTableDao().writeDetectNewPartitionVersion(); receiver.output(new InitialPipelineState(startTime, resume)); } diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dofn/InitializeDoFnTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dofn/InitializeDoFnTest.java index 84a3b51f207e..13e0b0ab251d 100644 --- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dofn/InitializeDoFnTest.java +++ b/sdks/java/io/google-cloud-platform/src/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 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()); + } }