This is an automated email from the ASF dual-hosted git repository.
ahmedabu98 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new b3bda9ac42a fix messaging (#40392)
b3bda9ac42a is described below
commit b3bda9ac42af52ef1254014be7373bca86c9319d
Author: Ahmed Abualsaud <[email protected]>
AuthorDate: Sun Oct 11 21:38:44 2026 -0700
fix messaging (#40392)
---
.../sdk/io/iceberg/cdc/sink/OrderedCommitFn.java | 13 ++++++-------
.../beam/sdk/io/iceberg/cdc/sink/TableSetup.java | 21 ++++++++++-----------
.../beam/sdk/io/iceberg/cdc/sink/WriteCdcRows.java | 3 +--
.../beam/sdk/io/iceberg/cdc/sink/package-info.java | 4 +---
.../sdk/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 69bf48aa431..970910a9120 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 @@ class OrderedCommitFn extends DoFn<KV<String,
WindowedCommit>, KV<String, Snapsh
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 a0775c0bb43..c17458afea6 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 @@ final class TableSetup implements Serializable {
}
@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 @@ final class TableSetup implements Serializable {
* 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 @@ final class TableSetup implements Serializable {
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 e32ad2f8d5a..a9ca1db6bc2 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 abstract class WriteCdcRows extends
PTransform<PCollection<Row>, IcebergW
* 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.
*
- * <p>With heartbeat enabled prefer cancel-and-resubmit over drain: the
self-re-arming
- * processing-time timer can keep a drain from completing.
+ * <p>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 c25496648f9..a31833a615e 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).
* <li><b>heartbeatCommits</b>: 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.
* </ul>
*
* <h3>Safety tripwires</h3>
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 63133acc32b..9c35d0b1e8d 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 class TableSetupTest {
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"));
}
//
-------------------------------------------------------------------------------------------
