From 225b25a5d91fff00b70ecb0dab75a6a2becf898c Mon Sep 17 00:00:00 2001 From: Ahmed Abualsaud Date: Fri, 2 Oct 2026 22:28:41 -0700 Subject: [PATCH] fix messaging --- .../io/iceberg/cdc/sink/OrderedCommitFn.java | 13 ++++++------ .../sdk/io/iceberg/cdc/sink/TableSetup.java | 21 +++++++++---------- .../sdk/io/iceberg/cdc/sink/WriteCdcRows.java | 3 +-- .../sdk/io/iceberg/cdc/sink/package-info.java | 4 +--- .../io/iceberg/cdc/sink/TableSetupTest.java | 2 +- 5 files changed, 19 insertions(+), 24 deletions(-) diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/OrderedCommitFn.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/OrderedCommitFn.java index 69bf48aa431d..970910a91201 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/OrderedCommitFn.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/OrderedCommitFn.java @@ -379,17 +379,16 @@ private CommitSummary commitOneWindow(Table table, WindowedCommit wc, @Nullable if (specId != currentSpecId) { metrics.specMismatchedWindows.inc(); LOG.warn( - "CDC sink '{}' committed window-end {} ms for table '{}' with equality deletes in " - + "a window mixing partition specs (pinned spec id {}, saw spec id {}): the " - + "partition spec evolved mid-run. An equality delete applies only to data " - + "files of its own (spec id, partition), so deletes may not reach rows " - + "written under the other spec. Run rewrite_data_files so that data is correctly " - + "repartitioned, then drain and restart the pipeline to converge on one spec.", + "CDC sink '{}' committed window-end {} ms for table '{}' with files from partition " + + "specs {} and {}, so some equality deletes may miss rows. Cancel the pipeline, " + + "run rewrite_data_files on the table, then restart from the last committed " + + "change (snapshot summary key {}).", sinkId, wc.getWindowEndMs(), dest, currentSpecId, - specId); + specId, + CommitToken.MAX_COMMITTED_SEQ_PREFIX + sinkId); break; } } diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/TableSetup.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/TableSetup.java index a0775c0bb438..c17458afea62 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/TableSetup.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/TableSetup.java @@ -121,7 +121,7 @@ protected boolean removeEldestEntry(Map.Entry eldest) { } @Nullable Dest existing = memo.get(destString); if (existing != null) { - requireResolvedPartitionSpec(destString, existing); + requireResolvedPartitionSpec(destString, existing, config.getSinkId()); return existing; } Dest dest; @@ -264,7 +264,7 @@ private PartitionSpec runSpec(Table table) { * Freshness is best-effort: the compared spec is the process-cached table's, refreshed only when * something in the process refreshes it. */ - private static void requireResolvedPartitionSpec(String destString, Dest dest) { + private static void requireResolvedPartitionSpec(String destString, Dest dest, String sinkId) { if (dest.partitionShardPlan() == null) { return; } @@ -273,17 +273,16 @@ private static void requireResolvedPartitionSpec(String destString, Dest dest) { throw new TableConfigException( "Table '" + destString - + "' changed its partition spec while the CDC sink was running with a " - + "shards_per_partition cap (spec id " + + "' changed its partition spec (spec id " + dest.spec().specId() - + " when the sink resolved the table, spec id " + + ", now spec id " + currentSpecId - + " now). Partition-block sharding derives each record's shard from the partition " - + "tuple under the resolved spec, so workers resolving different specs would split " - + "one primary key's window across shards and silently duplicate rows within a " - + "commit. " - + "Drain the pipeline before evolving the partition spec, and restart it " - + "afterwards."); + + ") while the CDC sink was running with shards_per_partition set. Cancel the " + + "pipeline, run rewrite_data_files on the table, then restart from the last " + + "committed change (snapshot summary key " + + CommitToken.MAX_COMMITTED_SEQ_PREFIX + + sinkId + + ")."); } } diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/WriteCdcRows.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/WriteCdcRows.java index e32ad2f8d5ab..a9ca1db6bc22 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/WriteCdcRows.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/WriteCdcRows.java @@ -305,8 +305,7 @@ public WriteCdcRows withUpsert(boolean upsert) { * token-bearing snapshot stays recent and is less likely to be lost to {@code expire_snapshots} * before the sink resumes. Disabled by default, and ignored for bounded (batch) input. * - *

With heartbeat enabled prefer cancel-and-resubmit over drain: the self-re-arming - * processing-time timer can keep a drain from completing. + *

With heartbeat enabled, stop the pipeline with cancel, not drain. */ public WriteCdcRows withTokenHeartbeat(Duration interval) { Preconditions.checkArgument( diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/package-info.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/package-info.java index c25496648f98..a31833a615ed 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/package-info.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/package-info.java @@ -57,9 +57,7 @@ * the table has too many concurrent writers, and the sink will not make progress past the * failing window (commits are strictly ordered). *

  • heartbeatCommits: empty token-refresh commits emitted while a destination is idle - * (only if {@code withTokenHeartbeat} is configured). Each one is a real snapshot. With - * heartbeat enabled prefer cancel-and-resubmit over drain: the self-re-arming processing-time - * timer can keep a drain from completing. + * (only if {@code withTokenHeartbeat} is configured). Each one is a real snapshot. * * *

    Safety tripwires

    diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/TableSetupTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/TableSetupTest.java index 63133acc32bd..9c35d0b1e8d7 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/TableSetupTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/TableSetupTest.java @@ -518,7 +518,7 @@ public void blockShardingStillRefusesSpecDrift() { assertThat(error.getMessage(), containsString(id.toString())); assertThat(error.getMessage(), containsString("spec id " + resolvedSpecId)); assertThat(error.getMessage(), containsString("spec id " + evolvedSpecId)); - assertThat(error.getMessage(), containsString("Drain the pipeline")); + assertThat(error.getMessage(), containsString("Cancel the pipeline")); } // -------------------------------------------------------------------------------------------