Fix Iceberg CDC sink messaging

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;
       }
     }

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<String, Dest> 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
+              + ").");
     }
   }
 

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.
    *
-   * <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(

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>

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"));
   }
 
   // -------------------------------------------------------------------------------------------