[iceberg] Restore flush to clear factory cache for schema reload

flush(boolean) was a no-op to avoid splitting same-PK updates into
multiple batches within a checkpoint. But without clearing the writer
factory cache, new writes after a checkpoint could use a stale factory
that doesn't reflect schema changes applied by IcebergMetadataApplier.

Change flush to clear writerFactoryMap only. Writers themselves keep
running so same-PK updates within a checkpoint go through the same
writer and use position deletes (exact file+offset), which work
correctly within a single snapshot without needing a separate seq_num.

Schema-change flushes still go through flushTableWriter(tableId), which
drains the writer and bumps batchIndex so equality-deletes land in a
snapshot with a strictly higher sequence number than the data they target.

Author: spoorthibasu

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/v2/IcebergWriter.java

@@ -190,10 +190,13 @@ public void write(Event event, Context context) throws IOException {
     }
 
     @Override
-    public void flush(boolean flush) {
-        // Flush may be called many times during one checkpoint by non-data events.
-        // Avoid rotating all task writers here, which can split same-PK updates into multiple
-        // batches within one checkpoint and break dedup semantics in downstream reads.
+    public void flush(boolean flush) throws IOException {
+        // Drop cached writer factories so the next write reloads them from the current catalog
+        // schema. Writers themselves keep running — same-writer updates use position deletes
+        // (exact file+offset), which work correctly within a single snapshot. Schema-change
+        // splits are handled by flushTableWriter, which completes the writer and advances
+        // batchIndex so equality-deletes land in a snapshot with a higher seq_num.
+        writerFactoryMap.clear();
     }
 
     private void flushTableWriter(TableId tableId) throws IOException {

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/test/java/org/apache/flink/cdc/connectors/iceberg/sink/v2/IcebergWriterTest.java

@@ -649,12 +649,25 @@ public void testNoDuplicateWhenFlushSplitsSamePkUpdatesWithinCheckpoint() throws
                 generator.generate(new Object[] {1L, BinaryStringData.fromString("b")});
 
         icebergWriter.write(DataChangeEvent.insertEvent(tableId, recordA), null);
-        // flush(false) is a no-op; both events reach the same writer so the position-delete
-        // mechanism within the writer handles dedup correctly.
+        // flush only clears the factory cache; the active writer keeps running. The update
+        // therefore goes to the same writer as the insert, which uses a position delete
+        // (exact file+offset) rather than an equality delete — no separate snapshot needed.
         icebergWriter.flush(false);
         icebergWriter.write(DataChangeEvent.updateEvent(tableId, recordA, recordB), null);
 
         Collection<WriteResultWrapper> writeResults = icebergWriter.prepareCommit();
+
+        // flush only clears the factory cache — the writer keeps running. Both the insert and
+        // the update go through the same writer (batchIndex 0), which uses position deletes
+        // for the update. Position deletes reference the exact file+row offset, so they work
+        // correctly within a single snapshot without needing a separate seq_num.
+        List<Integer> batchIndexes =
+                writeResults.stream()
+                        .map(WriteResultWrapper::getBatchIndex)
+                        .sorted()
+                        .collect(Collectors.toList());
+        Assertions.assertThat(batchIndexes).containsExactly(0);
+
         IcebergCommitter icebergCommitter = new IcebergCommitter(catalogOptions, new HashMap<>());
         Collection<Committer.CommitRequest<WriteResultWrapper>> collection =
                 writeResults.stream().map(MockCommitRequestImpl::new).collect(Collectors.toList());
@@ -665,16 +678,11 @@ public void testNoDuplicateWhenFlushSplitsSamePkUpdatesWithinCheckpoint() throws
     }
 
     /**
-     * Verifies that same-PK updates split across a schema-change-triggered writer flush within one
-     * checkpoint do not produce duplicate records.
-     *
-     * <p>Root cause: when a schema-change event forces {@code flushTableWriter}, the pre-change
-     * writes land in one {@code WriteResultWrapper} (batch 0) and the post-change writes land in
-     * another (batch 1). Committing both in a single Iceberg {@code RowDelta} would give all files
-     * the same sequence number N, so the equality-delete file from batch 1 (seq=N) cannot delete
-     * the data file from batch 0 (seq=N) — Iceberg equality-delete semantics require strictly lower
-     * sequence numbers. The fix commits each batch as a separate sequential snapshot, giving batch
-     * 1 a higher sequence number so its equality-delete correctly supersedes batch 0's data.
+     * A schema change mid-checkpoint splits writes into two batches for the same table. Both
+     * batches must not land in the same Iceberg snapshot: files in one snapshot share the same
+     * seq_num, so the equality-delete from batch 1 would silently miss the insert from batch 0.
+     * Each batch gets its own snapshot so the delete's seq_num is strictly higher than the data it
+     * targets.
      */
     @Test
     public void testNoDuplicateWhenSchemaChangeFlushSplitsSamePkUpdates() throws Exception {