[iceberg] Add test for mixed data across subtasks with multiple schema changes

Adds testNoDuplicateWithMixedDataAcrossSubtasksAndMultipleSchemaChanges to
cover the scenario where subtask 0 has data only before SC1, subtask 1 has
data only between SC1 and SC2, and both have updates after SC2. Verifies
that equality-deletes in batch 2 correctly suppress stale data from batches
0 and 1 across all three snapshots.

Author: spoorthibasu

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

@@ -1294,6 +1294,143 @@ public void testSameBatchIndexFromTwoSubtasksMergedIntoOneSnapshot() throws Exce
         Assertions.assertThat(result).containsExactlyInAnyOrder("1, a", "2, b");
     }
 
+    /**
+     * Verifies no duplicates in the most complex parallel scenario: subtask 0 has data only before
+     * SC1, subtask 1 has data only between SC1 and SC2, and both have updates after SC2. This
+     * exercises all three batchIndex slots across two subtasks simultaneously and confirms that
+     * equality-deletes in batch 2 correctly suppress stale data from batches 0 and 1.
+     */
+    @Test
+    public void testNoDuplicateWithMixedDataAcrossSubtasksAndMultipleSchemaChanges()
+            throws Exception {
+        Map<String, String> catalogOptions = new HashMap<>();
+        catalogOptions.put("type", "hadoop");
+        catalogOptions.put(
+                "warehouse",
+                new File(temporaryFolder.toFile(), UUID.randomUUID().toString()).toString());
+        catalogOptions.put("cache-enabled", "false");
+        Catalog catalog =
+                CatalogUtil.buildIcebergCatalog(
+                        "cdc-iceberg-catalog", catalogOptions, new Configuration());
+        IcebergMetadataApplier icebergMetadataApplier = new IcebergMetadataApplier(catalogOptions);
+
+        String jobId = UUID.randomUUID().toString();
+        String operatorId = UUID.randomUUID().toString();
+
+        IcebergWriter writer0 =
+                new IcebergWriter(
+                        catalogOptions, 0, 1, ZoneId.systemDefault(), 0, jobId, operatorId);
+        IcebergWriter writer1 =
+                new IcebergWriter(
+                        catalogOptions, 1, 1, ZoneId.systemDefault(), 0, jobId, operatorId);
+
+        TableId tableId = TableId.parse("test.iceberg_table");
+        Schema schema0 =
+                Schema.newBuilder()
+                        .physicalColumn("id", DataTypes.BIGINT().notNull())
+                        .physicalColumn("name", DataTypes.VARCHAR(100))
+                        .primaryKey("id")
+                        .build();
+        CreateTableEvent createEvent = new CreateTableEvent(tableId, schema0);
+        icebergMetadataApplier.applySchemaChange(createEvent);
+        writer0.write(createEvent, null);
+        writer1.write(createEvent, null);
+
+        BinaryRecordDataGenerator gen0 =
+                new BinaryRecordDataGenerator(
+                        schema0.getColumnDataTypes().toArray(new DataType[0]));
+
+        // Batch 0: only subtask 0 has data before SC1.
+        writer0.write(
+                DataChangeEvent.insertEvent(
+                        tableId, gen0.generate(new Object[] {1L, BinaryStringData.fromString("a")})),
+                null);
+        // Subtask 1 has no data before SC1.
+
+        // SC1 broadcast to both subtasks.
+        AddColumnEvent sc1 =
+                new AddColumnEvent(
+                        tableId,
+                        Arrays.asList(
+                                AddColumnEvent.last(
+                                        new PhysicalColumn(
+                                                "extra1", DataTypes.STRING(), null, null))));
+        icebergMetadataApplier.applySchemaChange(sc1);
+        writer0.write(sc1, null); // has writer → flush batchIndex=0; counter → 1
+        writer1.write(sc1, null); // no writer  → counter must still advance to 1
+
+        Schema schema1 = SchemaUtils.applySchemaChangeEvent(schema0, sc1);
+        BinaryRecordDataGenerator gen1 =
+                new BinaryRecordDataGenerator(
+                        schema1.getColumnDataTypes().toArray(new DataType[0]));
+
+        // Batch 1: only subtask 1 has data between SC1 and SC2.
+        writer1.write(
+                DataChangeEvent.insertEvent(
+                        tableId,
+                        gen1.generate(new Object[] {2L, BinaryStringData.fromString("b"), null})),
+                null);
+        // Subtask 0 has no data between SC1 and SC2.
+
+        // SC2 broadcast to both subtasks.
+        AddColumnEvent sc2 =
+                new AddColumnEvent(
+                        tableId,
+                        Arrays.asList(
+                                AddColumnEvent.last(
+                                        new PhysicalColumn(
+                                                "extra2", DataTypes.STRING(), null, null))));
+        icebergMetadataApplier.applySchemaChange(sc2);
+        writer0.write(sc2, null); // no writer  → counter must still advance to 2
+        writer1.write(sc2, null); // has writer → flush batchIndex=1; counter → 2
+
+        Schema schema2 = SchemaUtils.applySchemaChangeEvent(schema1, sc2);
+        BinaryRecordDataGenerator gen2 =
+                new BinaryRecordDataGenerator(
+                        schema2.getColumnDataTypes().toArray(new DataType[0]));
+
+        // Batch 2: both subtasks update their respective rows after SC2.
+        // Subtask 0 updates id=1 "a" → "c"; subtask 1 updates id=2 "b" → "d".
+        writer0.write(
+                DataChangeEvent.updateEvent(
+                        tableId,
+                        gen2.generate(new Object[] {1L, BinaryStringData.fromString("a"), null, null}),
+                        gen2.generate(
+                                new Object[] {1L, BinaryStringData.fromString("c"), null, null})),
+                null);
+        writer1.write(
+                DataChangeEvent.updateEvent(
+                        tableId,
+                        gen2.generate(new Object[] {2L, BinaryStringData.fromString("b"), null, null}),
+                        gen2.generate(
+                                new Object[] {2L, BinaryStringData.fromString("d"), null, null})),
+                null);
+
+        List<WriteResultWrapper> allResults = new ArrayList<>();
+        allResults.addAll(writer0.prepareCommit());
+        allResults.addAll(writer1.prepareCommit());
+
+        // Expect 3 batches: {0: sub0}, {1: sub1}, {2: sub0+sub1}
+        long distinctBatchIndices =
+                allResults.stream()
+                        .mapToInt(WriteResultWrapper::getBatchIndex)
+                        .distinct()
+                        .count();
+        Assertions.assertThat(distinctBatchIndices).isEqualTo(3);
+
+        IcebergCommitter committer = new IcebergCommitter(catalogOptions);
+        committer.commit(
+                allResults.stream()
+                        .map(MockCommitRequestImpl::new)
+                        .collect(Collectors.toList()));
+
+        // Only the final values must survive. Equality-deletes in batch 2 (seq N+2) must suppress
+        // the stale inserts in batch 0 (seq N) and batch 1 (seq N+1).
+        List<String> result = fetchTableContent(catalog, tableId, null);
+        Assertions.assertThat(result)
+                .containsExactlyInAnyOrder("1, c, null, null", "2, d, null, null");
+    }
+
     private static long countSnapshots(Table table) {
         long count = 0;
         for (Snapshot ignored : table.snapshots()) {