(hudi) branch master updated: test(flink): improve coverage for utility classes (#19546)

[danny0405]

This is an automated email from the ASF dual-hosted git repository.

danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git


The following commit(s) were added to refs/heads/master by this push:
     new 92f9640509b0 test(flink): improve coverage for utility classes (#19546)
92f9640509b0 is described below

commit 92f9640509b0c4043a6daca3c1127170cced672a
Author: Shuo Cheng <njucs...@gmail.com>
AuthorDate: Tue Aug 11 10:43:20 2026 +0800

    test(flink): improve coverage for utility classes (#19546)
    
    * test(flink): improve coverage for utility classes
---
 .../schema/TestSchemaProviderCompatibility.java    |  79 +++++++++++
 .../aggregate/TestBootstrapAggregation.java        |  58 ++++++++
 .../sink/buffer/TestBufferAccountingUtilities.java | 103 ++++++++++++++
 .../sink/clustering/TestClusteringEventModels.java |  63 +++++++++
 .../hudi/sink/common/TestWriteOperatorFactory.java |  43 ++++++
 .../sink/compact/TestCompactionEventModels.java    |  64 +++++++++
 .../sink/event/TestCorrespondentEventModels.java   |  59 ++++++++
 .../TestMemoryPagesExhaustedException.java         |  40 ++++++
 .../muttley/TestFlinkHudiMuttleyExceptions.java    |  58 ++++++++
 .../sink/overwrite/TestPartitionOverwriteMode.java |  35 +++++
 .../sink/partitioner/index/TestIndexBackends.java  |  56 ++++++++
 .../sink/partitioner/index/TestIndexRowUtils.java  | 115 +++++++++++++++
 .../sink/transform/TestTransformUtilities.java     | 106 ++++++++++++++
 .../apache/hudi/sink/utils/TestSinkUtilities.java  | 132 +++++++++++++++++
 .../enumerator/TestHoodieEnumeratorPosition.java   |  54 +++++++
 .../hudi/source/prune/TestPrimaryKeyPruners.java   |  75 ++++++++++
 .../source/rebalance/TestStreamReadRebalance.java  |  96 +++++++++++++
 .../source/split/TestSplitRequestEventModel.java   |  47 +++++++
 .../hudi/source/stats/TestColumnStatsModels.java   |  59 ++++++++
 .../format/TestFormatIteratorsAndIOFactory.java    | 105 ++++++++++++++
 .../format/mor/TestMergeOnReadTableState.java      |  68 +++++++++
 .../hudi/table/lookup/TestLookupUtilities.java     |  67 +++++++++
 .../org/apache/hudi/util/TestFlinkUtilities.java   | 156 +++++++++++++++++++++
 23 files changed, 1738 insertions(+)

diff --git 
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/schema/TestSchemaProviderCompatibility.java
 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/schema/TestSchemaProviderCompatibility.java
new file mode 100644
index 000000000000..ac010b2440ee
--- /dev/null
+++ 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/schema/TestSchemaProviderCompatibility.java
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.schema;
+
+import org.apache.hudi.common.schema.HoodieSchema;
+
+import org.apache.avro.Schema;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+class TestSchemaProviderCompatibility {
+
+  private static final Schema AVRO_SCHEMA = new Schema.Parser().parse(
+      
"{\"type\":\"record\",\"name\":\"record\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"}]}");
+
+  @Test
+  @SuppressWarnings("deprecation")
+  void testLegacyProviderConvertsSourceAndTargetSchemas() {
+    SchemaProvider provider = new SchemaProvider() {
+      @Override
+      public Schema getSourceSchema() {
+        return AVRO_SCHEMA;
+      }
+    };
+
+    assertEquals(HoodieSchema.fromAvroSchema(AVRO_SCHEMA), 
provider.getSourceHoodieSchema());
+    assertEquals(HoodieSchema.fromAvroSchema(AVRO_SCHEMA), 
provider.getTargetHoodieSchema());
+    assertSame(AVRO_SCHEMA, provider.getTargetSchema());
+  }
+
+  @Test
+  void testModernProviderFallsBackToSourceHoodieSchemaForTarget() {
+    HoodieSchema schema = HoodieSchema.fromAvroSchema(AVRO_SCHEMA);
+    SchemaProvider provider = new SchemaProvider() {
+      @Override
+      public HoodieSchema getSourceHoodieSchema() {
+        return schema;
+      }
+    };
+
+    assertSame(schema, provider.getTargetHoodieSchema());
+  }
+
+  @Test
+  @SuppressWarnings("deprecation")
+  void testNullLegacySchemaAndUnsupportedDefault() {
+    SchemaProvider nullProvider = new SchemaProvider() {
+      @Override
+      public Schema getSourceSchema() {
+        return null;
+      }
+    };
+    SchemaProvider defaultProvider = new SchemaProvider() { };
+
+    assertNull(nullProvider.getSourceHoodieSchema());
+    assertNull(nullProvider.getTargetHoodieSchema());
+    assertThrows(UnsupportedOperationException.class, 
defaultProvider::getSourceSchema);
+  }
+}
diff --git 
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/bootstrap/aggregate/TestBootstrapAggregation.java
 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/bootstrap/aggregate/TestBootstrapAggregation.java
new file mode 100644
index 000000000000..8a4caa473111
--- /dev/null
+++ 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/bootstrap/aggregate/TestBootstrapAggregation.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.sink.bootstrap.aggregate;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+class TestBootstrapAggregation {
+
+  @Test
+  void testAccumulatorCountsDistinctTasksAndMerges() {
+    BootstrapAccumulator first = new BootstrapAccumulator();
+    first.update(0);
+    first.update(0);
+    first.update(1);
+    assertEquals(2, first.readyTaskNum());
+    assertSame(first, first.merge(null));
+
+    BootstrapAccumulator second = new BootstrapAccumulator();
+    second.update(1);
+    second.update(2);
+    assertSame(first, first.merge(second));
+    assertEquals(3, first.readyTaskNum());
+  }
+
+  @Test
+  void testAggregateFunctionLifecycle() {
+    BootstrapAggFunction function = new BootstrapAggFunction();
+    BootstrapAccumulator first = function.createAccumulator();
+    BootstrapAccumulator second = function.createAccumulator();
+
+    assertSame(first, function.add(3, first));
+    function.add(3, first);
+    function.add(4, second);
+    assertEquals(1, function.getResult(first));
+    assertSame(first, function.merge(first, second));
+    assertEquals(2, function.getResult(first));
+    assertEquals("BootstrapAggFunction", BootstrapAggFunction.NAME);
+  }
+}
diff --git 
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/buffer/TestBufferAccountingUtilities.java
 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/buffer/TestBufferAccountingUtilities.java
new file mode 100644
index 000000000000..cb3b36d5df92
--- /dev/null
+++ 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/buffer/TestBufferAccountingUtilities.java
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.sink.buffer;
+
+import org.apache.hudi.configuration.FlinkOptions;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.memory.MemorySegmentFactory;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TestBufferAccountingUtilities {
+
+  @Test
+  void testTotalSizeTracerTracksCountdownAndReset() {
+    Configuration conf = new Configuration();
+    conf.set(FlinkOptions.WRITE_TASK_MAX_SIZE, 103D);
+    conf.set(FlinkOptions.WRITE_MERGE_MAX_MEMORY, 1);
+    TotalSizeTracer tracer = new TotalSizeTracer(conf);
+    long oneMb = 1024L * 1024L;
+
+    assertEquals(2 * oneMb, tracer.maxBufferSize);
+    assertFalse(tracer.trace(oneMb));
+    assertFalse(tracer.trace(oneMb));
+    assertTrue(tracer.trace(1));
+    tracer.countDown(oneMb + 1);
+    assertEquals(oneMb, tracer.bufferSize);
+    tracer.reset();
+    assertEquals(0, tracer.bufferSize);
+  }
+
+  @Test
+  void testTotalSizeTracerRejectsInsufficientTaskMemory() {
+    Configuration conf = new Configuration();
+    conf.set(FlinkOptions.WRITE_TASK_MAX_SIZE, 101D);
+    conf.set(FlinkOptions.WRITE_MERGE_MAX_MEMORY, 1);
+
+    IllegalStateException exception = 
assertThrows(IllegalStateException.class, () -> new TotalSizeTracer(conf));
+    
assertTrue(exception.getMessage().contains(FlinkOptions.WRITE_TASK_MAX_SIZE.key()));
+  }
+
+  @Test
+  void testBufferSizeDetectorUsesBinaryRowSizeAndResets() {
+    BufferSizeDetector detector = new BufferSizeDetector(0.00001);
+    BinaryRowData smallRow = new BinaryRowData(0);
+    smallRow.pointTo(MemorySegmentFactory.wrap(new byte[8]), 0, 8);
+
+    assertFalse(detector.detect(smallRow));
+    assertEquals(8, detector.getLastRecordSize());
+    assertTrue(detector.detect(smallRow));
+    assertTrue(detector.isFull());
+
+    detector.reset();
+    assertEquals(-1, detector.getLastRecordSize());
+    assertEquals(0, detector.totalSize);
+    assertFalse(detector.isFull());
+  }
+
+  @Test
+  void testBufferSizeDetectorReusesLastSampleForRegularObjects() {
+    BufferSizeDetector detector = new BufferSizeDetector(10);
+    assertFalse(detector.detect("first value"));
+    long sampledSize = detector.getLastRecordSize();
+
+    assertTrue(sampledSize > 0);
+    assertFalse(detector.detect("second value"));
+    assertEquals(sampledSize * 2, detector.totalSize);
+
+    // The seeded random makes sampling deterministic and covers both outcomes.
+    boolean sampled = false;
+    boolean skipped = false;
+    for (int i = 0; i < 500 && !(sampled && skipped); i++) {
+      if (detector.sampling()) {
+        sampled = true;
+      } else {
+        skipped = true;
+      }
+    }
+    assertTrue(sampled);
+    assertTrue(skipped);
+  }
+}
diff --git 
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/clustering/TestClusteringEventModels.java
 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/clustering/TestClusteringEventModels.java
new file mode 100644
index 000000000000..62d0209da75f
--- /dev/null
+++ 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/clustering/TestClusteringEventModels.java
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.sink.clustering;
+
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.model.ClusteringGroupInfo;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+
+class TestClusteringEventModels {
+
+  @Test
+  void testPlanEventAccessors() {
+    ClusteringGroupInfo group = mock(ClusteringGroupInfo.class);
+    Map<String, String> params = Collections.singletonMap("strategy", "sort");
+    ClusteringPlanEvent event = new ClusteringPlanEvent("001", group, params);
+    event.setIndex(7);
+
+    assertEquals("001", event.getClusteringInstantTime());
+    assertSame(group, event.getClusteringGroupInfo());
+    assertSame(params, event.getStrategyParams());
+    assertEquals(7, event.getIndex());
+  }
+
+  @Test
+  void testCommitEventDistinguishesSuccessAndFailure() {
+    assertTrue(new ClusteringCommitEvent("001", "file-1", 1).isFailed());
+
+    WriteStatus status = mock(WriteStatus.class);
+    ClusteringCommitEvent success = new ClusteringCommitEvent(
+        "002", "file-2", Collections.singletonList(status), 2);
+    assertFalse(success.isFailed());
+    assertEquals("002", success.getInstant());
+    assertEquals("file-2", success.getFileIds());
+    assertSame(status, success.getWriteStatuses().get(0));
+    assertEquals(2, success.getTaskID());
+  }
+}
diff --git 
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/common/TestWriteOperatorFactory.java
 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/common/TestWriteOperatorFactory.java
new file mode 100644
index 000000000000..635393c00143
--- /dev/null
+++ 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/common/TestWriteOperatorFactory.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.sink.common;
+
+import org.apache.hudi.sink.StreamWriteOperatorCoordinator;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.mockito.Mockito.mock;
+
+class TestWriteOperatorFactory {
+
+  @Test
+  @SuppressWarnings("unchecked")
+  void testFactoryAndCoordinatorProvider() {
+    WriteOperatorFactory<String> factory = WriteOperatorFactory.instance(
+        new Configuration(), mock(AbstractWriteOperator.class));
+
+    assertNotNull(factory);
+    assertInstanceOf(StreamWriteOperatorCoordinator.Provider.class,
+        factory.getCoordinatorProvider("operator", new OperatorID()));
+  }
+}
diff --git 
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/TestCompactionEventModels.java
 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/TestCompactionEventModels.java
new file mode 100644
index 000000000000..87149b39d90b
--- /dev/null
+++ 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/TestCompactionEventModels.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.sink.compact;
+
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.model.CompactionOperation;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+
+class TestCompactionEventModels {
+
+  @Test
+  void testPlanEventConstructorsAndAccessors() {
+    CompactionOperation operation = mock(CompactionOperation.class);
+    CompactionPlanEvent event = new CompactionPlanEvent("001", operation, 3, 
true, true);
+
+    assertEquals("001", event.getCompactionInstantTime());
+    assertSame(operation, event.getOperation());
+    assertEquals(3, event.getIndex());
+    assertTrue(event.isMetadataTable());
+    assertTrue(event.isLogCompaction());
+  }
+
+  @Test
+  void testCommitEventDistinguishesSuccessAndFailure() {
+    CompactionCommitEvent failed = new CompactionCommitEvent("001", "file-1", 
2, false, true);
+    assertTrue(failed.isFailed());
+    assertTrue(failed.isLogCompaction());
+
+    WriteStatus status = mock(WriteStatus.class);
+    CompactionCommitEvent success = new CompactionCommitEvent(
+        "002", "file-2", Collections.singletonList(status), 4, true, false);
+    assertFalse(success.isFailed());
+    assertEquals("002", success.getInstant());
+    assertEquals("file-2", success.getFileId());
+    assertSame(status, success.getWriteStatuses().get(0));
+    assertEquals(4, success.getTaskID());
+    assertTrue(success.isMetadataTable());
+  }
+}
diff --git 
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/event/TestCorrespondentEventModels.java
 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/event/TestCorrespondentEventModels.java
new file mode 100644
index 000000000000..8dc7102aad25
--- /dev/null
+++ 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/event/TestCorrespondentEventModels.java
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.sink.event;
+
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.jobgraph.tasks.TaskOperatorEventGateway;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.mockito.Mockito.mock;
+
+class TestCorrespondentEventModels {
+
+  @Test
+  void testCorrespondentFactoryAndRequestResponseModels() {
+    OperatorID operatorId = new OperatorID();
+    TaskOperatorEventGateway gateway = mock(TaskOperatorEventGateway.class);
+    Correspondent correspondent = Correspondent.getInstance(operatorId, 
gateway);
+    assertSame(operatorId, correspondent.getOperatorID());
+    assertSame(gateway, correspondent.getGateway());
+
+    assertEquals(9L, 
Correspondent.InstantTimeRequest.getInstance(9L).getCheckpointId());
+    assertEquals("001", 
Correspondent.InstantTimeResponse.getInstance("001").getInstant());
+    assertNotNull(Correspondent.InflightInstantsRequest.getInstance());
+
+    HashMap<Long, String> instants = new HashMap<>();
+    instants.put(9L, "001");
+    assertSame(instants,
+        
Correspondent.InflightInstantsResponse.getInstance(instants).getInflightInstants());
+  }
+
+  @Test
+  void testCommitAckFactoryAndAccessors() {
+    CommitAckEvent event = CommitAckEvent.getInstance(11L);
+    assertEquals(11L, event.getCheckpointId());
+    event.setCheckpointId(12L);
+    assertEquals(12L, event.getCheckpointId());
+  }
+}
diff --git 
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/exception/TestMemoryPagesExhaustedException.java
 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/exception/TestMemoryPagesExhaustedException.java
new file mode 100644
index 000000000000..a0776d8396e7
--- /dev/null
+++ 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/exception/TestMemoryPagesExhaustedException.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.sink.exception;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+class TestMemoryPagesExhaustedException {
+
+  @Test
+  void testConstructorsPreserveMessageAndCause() {
+    RuntimeException cause = new RuntimeException("root cause");
+    MemoryPagesExhaustedException withCause =
+        new MemoryPagesExhaustedException("no pages", cause);
+    MemoryPagesExhaustedException withoutCause =
+        new MemoryPagesExhaustedException("still no pages");
+
+    assertEquals("no pages", withCause.getMessage());
+    assertSame(cause, withCause.getCause());
+    assertEquals("still no pages", withoutCause.getMessage());
+  }
+}
diff --git 
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/muttley/TestFlinkHudiMuttleyExceptions.java
 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/muttley/TestFlinkHudiMuttleyExceptions.java
new file mode 100644
index 000000000000..0fbfbdcc6b69
--- /dev/null
+++ 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/muttley/TestFlinkHudiMuttleyExceptions.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.sink.muttley;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+class TestFlinkHudiMuttleyExceptions {
+
+  @Test
+  void testBaseExceptionConstructorsAndValidation() {
+    FlinkHudiMuttleyException withoutMessage = new 
FlinkHudiMuttleyException(418);
+    FlinkHudiMuttleyException withMessage = new 
FlinkHudiMuttleyException("teapot", 418);
+
+    assertNull(withoutMessage.getMessage());
+    assertEquals(418, withoutMessage.getStatusCode());
+    assertEquals("teapot", withMessage.getMessage());
+    assertEquals(418, FlinkHudiMuttleyException.validate(418, code -> code >= 
400));
+    assertThrows(IllegalArgumentException.class,
+        () -> FlinkHudiMuttleyException.validate(399, code -> code >= 400));
+  }
+
+  @Test
+  void testClientAndServerExceptionConstructors() {
+    FlinkHudiMuttleyClientException client = new 
FlinkHudiMuttleyClientException(404);
+    FlinkHudiMuttleyClientException customClient =
+        new FlinkHudiMuttleyClientException("missing", 404);
+    FlinkHudiMuttleyServerException server = new 
FlinkHudiMuttleyServerException(503);
+    FlinkHudiMuttleyServerException customServer =
+        new FlinkHudiMuttleyServerException("unavailable", 503);
+
+    assertEquals("Muttley client error with status code: 404", 
client.getMessage());
+    assertEquals("missing", customClient.getMessage());
+    assertEquals(404, customClient.getStatusCode());
+    assertEquals("Muttley server error with status code: 503", 
server.getMessage());
+    assertEquals("unavailable", customServer.getMessage());
+    assertEquals(503, customServer.getStatusCode());
+  }
+}
diff --git 
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/overwrite/TestPartitionOverwriteMode.java
 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/overwrite/TestPartitionOverwriteMode.java
new file mode 100644
index 000000000000..c09a623ea144
--- /dev/null
+++ 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/overwrite/TestPartitionOverwriteMode.java
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.sink.overwrite;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class TestPartitionOverwriteMode {
+
+  @Test
+  void testEnumValues() {
+    assertArrayEquals(
+        new PartitionOverwriteMode[] {PartitionOverwriteMode.STATIC, 
PartitionOverwriteMode.DYNAMIC},
+        PartitionOverwriteMode.values());
+    assertEquals(PartitionOverwriteMode.DYNAMIC, 
PartitionOverwriteMode.valueOf("DYNAMIC"));
+  }
+}
diff --git 
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/index/TestIndexBackends.java
 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/index/TestIndexBackends.java
new file mode 100644
index 000000000000..85d6feed642c
--- /dev/null
+++ 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/index/TestIndexBackends.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.sink.partitioner.index;
+
+import org.apache.hudi.common.model.HoodieRecordGlobalLocation;
+
+import org.apache.flink.api.common.state.ValueState;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class TestIndexBackends {
+
+  @Test
+  @SuppressWarnings("unchecked")
+  void testFlinkStateBackendDelegatesGetAndUpdate() throws Exception {
+    ValueState<HoodieRecordGlobalLocation> state = mock(ValueState.class);
+    HoodieRecordGlobalLocation location = 
mock(HoodieRecordGlobalLocation.class);
+    when(state.value()).thenReturn(location);
+    FlinkStateIndexBackend backend = new FlinkStateIndexBackend(state);
+
+    assertSame(location, backend.get("ignored-key"));
+    backend.update("ignored-key", location);
+    verify(state).update(location);
+    backend.close();
+  }
+
+  @Test
+  void testDummyPartitionedBackendIsNoOp() throws Exception {
+    DummyPartitionedIndexBackend backend = new DummyPartitionedIndexBackend();
+
+    assertNull(backend.get("partition", "key"));
+    backend.update("partition", "key", "file-id");
+    backend.close();
+  }
+}
diff --git 
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/index/TestIndexRowUtils.java
 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/index/TestIndexRowUtils.java
new file mode 100644
index 000000000000..a3452e0560e7
--- /dev/null
+++ 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/index/TestIndexRowUtils.java
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.sink.partitioner.index;
+
+import org.apache.hudi.client.model.HoodieFlinkInternalRow;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
+
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.types.RowKind;
+import org.junit.jupiter.api.Test;
+
+import java.util.UUID;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class TestIndexRowUtils {
+
+  @Test
+  void testCreateRecordIndexRowsForInsertAndDelete() {
+    String fileId = UUID.randomUUID().toString();
+    HoodieFlinkInternalRow insert = internalRow("I", fileId);
+    HoodieFlinkInternalRow delete = internalRow("D", fileId);
+
+    RowData insertIndexRow = IndexRowUtils.createRecordIndexRow(insert);
+    RowData deleteIndexRow = IndexRowUtils.createRecordIndexRow(delete);
+
+    assertEquals(RowKind.INSERT, insertIndexRow.getRowKind());
+    assertEquals(RowKind.DELETE, deleteIndexRow.getRowKind());
+    assertEquals(IndexRowUtils.RLI_TYPE, insertIndexRow.getByte(0));
+    assertEquals("key-1", IndexRowUtils.getRecordKey(insertIndexRow));
+    assertEquals("partition-1", IndexRowUtils.getPartition(insertIndexRow));
+    assertEquals(new HoodieKey("key-1", "partition-1"), 
IndexRowUtils.getHoodieKey(insertIndexRow));
+  }
+
+  @Test
+  void testCreateRecordIndexRowRejectsUnexpectedOperation() {
+    HoodieException exception = assertThrows(HoodieException.class,
+        () -> IndexRowUtils.createRecordIndexRow(internalRow("U", 
UUID.randomUUID().toString())));
+    assertEquals("Unexpected operation type: U", exception.getMessage());
+  }
+
+  @Test
+  void testConvertInsertAndDeleteIndexRowsToHoodieRecords() {
+    String fileId = UUID.randomUUID().toString();
+    RowData insertIndexRow = 
IndexRowUtils.createRecordIndexRow(internalRow("I", fileId));
+    RowData deleteIndexRow = 
IndexRowUtils.createRecordIndexRow(internalRow("D", fileId));
+    HoodieWriteConfig writeConfig = mock(HoodieWriteConfig.class);
+    when(writeConfig.getWritesFileIdEncoding()).thenReturn(0);
+    when(writeConfig.isRecordLevelIndexEnabled()).thenReturn(true);
+
+    HoodieRecord insertRecord = IndexRowUtils.convertToHoodieRecord(1L, 
insertIndexRow, writeConfig);
+    HoodieRecord deleteRecord = IndexRowUtils.convertToHoodieRecord(1L, 
deleteIndexRow, writeConfig);
+
+    assertNotNull(insertRecord);
+    assertNotNull(deleteRecord);
+    assertEquals("key-1", insertRecord.getRecordKey());
+    assertEquals("key-1", deleteRecord.getRecordKey());
+  }
+
+  @Test
+  void testConvertRejectsUnsupportedRowKindAndIndexType() {
+    HoodieWriteConfig writeConfig = mock(HoodieWriteConfig.class);
+    GenericRowData unsupportedKind = indexRow(IndexRowUtils.RLI_TYPE, 
RowKind.UPDATE_AFTER);
+    GenericRowData unsupportedType = indexRow((byte) 1, RowKind.INSERT);
+
+    HoodieException kindException = assertThrows(HoodieException.class,
+        () -> IndexRowUtils.convertToHoodieRecord(1L, unsupportedKind, 
writeConfig));
+    assertEquals("Unsupported operation type for index row: UPDATE_AFTER", 
kindException.getMessage());
+    HoodieException typeException = assertThrows(HoodieException.class,
+        () -> IndexRowUtils.convertToHoodieRecord(1L, unsupportedType, 
writeConfig));
+    assertEquals("Unsupported type for index row: 1", 
typeException.getMessage());
+  }
+
+  private static HoodieFlinkInternalRow internalRow(String operation, String 
fileId) {
+    HoodieFlinkInternalRow row =
+        new HoodieFlinkInternalRow("key-1", "partition-1", operation, new 
GenericRowData(0));
+    row.setFileId(fileId);
+    return row;
+  }
+
+  private static GenericRowData indexRow(byte indexType, RowKind rowKind) {
+    GenericRowData row = new 
GenericRowData(IndexRowUtils.INDEX_ROW_TYPE.getFieldCount());
+    row.setField(0, indexType);
+    row.setField(1, StringData.fromString("key-1"));
+    row.setField(2, StringData.fromString("partition-1"));
+    row.setField(3, StringData.fromString(UUID.randomUUID().toString()));
+    row.setRowKind(rowKind);
+    return row;
+  }
+}
diff --git 
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/transform/TestTransformUtilities.java
 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/transform/TestTransformUtilities.java
new file mode 100644
index 000000000000..177800590b09
--- /dev/null
+++ 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/transform/TestTransformUtilities.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.sink.transform;
+
+import org.apache.hudi.client.model.HoodieFlinkInternalRow;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.sink.bulk.RowDataKeyGen;
+import org.apache.hudi.table.action.commit.BucketInfo;
+import org.apache.hudi.table.action.commit.BucketType;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.types.logical.RowType;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class TestTransformUtilities {
+
+  @Test
+  void testRecordConverterBuildsFlinkRecordFromKeyAndBucket() {
+    RowDataKeyGen keyGen = mock(RowDataKeyGen.class);
+    RowData row = GenericRowData.of(1);
+    when(keyGen.getRecordKey(row)).thenReturn("record-key");
+    BucketInfo bucket = new BucketInfo(BucketType.INSERT, "file-id", 
"partition");
+
+    HoodieRecord record = RecordConverter.getInstance(keyGen).convert(row, 
bucket);
+
+    assertEquals("record-key", record.getRecordKey());
+    assertEquals("partition", record.getPartitionPath());
+    assertSame(row, record.getData());
+  }
+
+  @Test
+  void testRowDataToHoodieFunctionFactoryAndMapping() throws Exception {
+    RowType rowType = (RowType) DataTypes.ROW(
+        DataTypes.FIELD("id", DataTypes.STRING()),
+        DataTypes.FIELD("partition", DataTypes.STRING())).getLogicalType();
+    Configuration conf = new Configuration();
+    conf.set(FlinkOptions.RECORD_KEY_FIELD, "id");
+    conf.set(FlinkOptions.PARTITION_PATH_FIELD, "partition");
+
+    RowDataToHoodieFunction<RowData, HoodieFlinkInternalRow> regular =
+        RowDataToHoodieFunctions.create(rowType, conf);
+    assertEquals(RowDataToHoodieFunction.class, regular.getClass());
+
+    GenericRowData row = GenericRowData.of(
+        StringData.fromString("id-1"), StringData.fromString("p1"));
+    HoodieFlinkInternalRow result = regular.map(row);
+    assertEquals("id-1", result.getRecordKey());
+    assertEquals("p1", result.getPartitionPath());
+    assertSame(row, result.getRowData());
+
+    conf.set(FlinkOptions.WRITE_RATE_LIMIT, 10L);
+    assertInstanceOf(RowDataToHoodieFunctionWithRateLimit.class,
+        RowDataToHoodieFunctions.create(rowType, conf));
+  }
+
+  @Test
+  @SuppressWarnings("unchecked")
+  void testChainedTransformerAppliesInOrderAndReportsNames() {
+    Transformer first = mock(Transformer.class);
+    Transformer second = mock(Transformer.class);
+    DataStream<RowData> source = mock(DataStream.class);
+    DataStream<RowData> intermediate = mock(DataStream.class);
+    DataStream<RowData> result = mock(DataStream.class);
+    when(first.apply(source)).thenReturn(intermediate);
+    when(second.apply(intermediate)).thenReturn(result);
+    ChainedTransformer chained = new ChainedTransformer(Arrays.asList(first, 
second));
+
+    assertSame(result, chained.apply(source));
+    org.mockito.InOrder order = inOrder(first, second);
+    order.verify(first).apply(source);
+    order.verify(second).apply(intermediate);
+    assertEquals(Arrays.asList(first.getClass().getName(), 
second.getClass().getName()),
+        chained.getTransformersNames());
+  }
+}
diff --git 
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestSinkUtilities.java
 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestSinkUtilities.java
new file mode 100644
index 000000000000..2eaa59d0f0e2
--- /dev/null
+++ 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestSinkUtilities.java
@@ -0,0 +1,132 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.sink.utils;
+
+import org.apache.hudi.exception.HoodieException;
+
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.core.memory.MemorySegmentFactory;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.table.data.GenericRowData;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TestSinkUtilities {
+
+  @Test
+  void testNaturalOrderImplementationsUseConstantKeys() {
+    GenericRowData first = GenericRowData.of(1);
+    GenericRowData second = GenericRowData.of(2);
+    assertEquals(0, new NaturalOrderRecordComparator().compare(first, second));
+
+    NaturalOrderKeyComputer keyComputer = new NaturalOrderKeyComputer();
+    MemorySegment firstSegment = MemorySegmentFactory.wrap(new byte[] {42});
+    MemorySegment secondSegment = MemorySegmentFactory.wrap(new byte[] {24});
+    keyComputer.putKey(first, firstSegment, 0);
+
+    assertEquals(0, firstSegment.get(0));
+    assertEquals(0, keyComputer.compareKey(firstSegment, 0, secondSegment, 0));
+    keyComputer.swapKey(firstSegment, 0, secondSegment, 0);
+    assertEquals(0, firstSegment.get(0));
+    assertEquals(24, secondSegment.get(0));
+    assertEquals(1, keyComputer.getNumKeyBytes());
+    assertTrue(keyComputer.isKeyFullyDetermines());
+    assertFalse(keyComputer.invertKey());
+  }
+
+  @Test
+  void testOperatorIdGenerationIsDeterministicAndUidSpecific() {
+    OperatorID first = OperatorIDGenerator.fromUid("writer");
+
+    assertEquals(first, OperatorIDGenerator.fromUid("writer"));
+    assertNotEquals(first, OperatorIDGenerator.fromUid("committer"));
+  }
+
+  @Test
+  void testSamplingActionRunsAtEachConfiguredBoundary() {
+    SamplingActionExecutor executor = new SamplingActionExecutor(3);
+    AtomicInteger invocations = new AtomicInteger();
+
+    for (int i = 0; i < 7; i++) {
+      executor.runIfNecessary(invocations::incrementAndGet);
+    }
+
+    assertEquals(2, invocations.get());
+  }
+
+  @Test
+  void testExplicitClassloaderThreadFactoryConfiguresSingleThread() throws 
Exception {
+    ClassLoader classLoader = new ClassLoader() { };
+    AtomicReference<Throwable> failure = new AtomicReference<>();
+    Thread.UncaughtExceptionHandler handler = (thread, throwable) -> 
failure.set(throwable);
+    ExplicitClassloaderThreadFactory factory =
+        new ExplicitClassloaderThreadFactory("coordinator", classLoader, 
handler);
+    RuntimeException expected = new RuntimeException("expected");
+
+    Thread thread = factory.newThread(() -> {
+      throw expected;
+    });
+    assertEquals("coordinator", thread.getName());
+    assertSame(classLoader, thread.getContextClassLoader());
+    assertSame(handler, thread.getUncaughtExceptionHandler());
+    thread.start();
+    thread.join();
+
+    assertSame(expected, failure.get());
+    assertThrows(Error.class, () -> factory.newThread(() -> { }));
+  }
+
+  @Test
+  void testTimeWaitBuilderAndTimeout() {
+    assertThrows(NullPointerException.class, () -> TimeWait.builder().build());
+
+    TimeWait wait = TimeWait.builder()
+        .action("checkpoint acknowledgement")
+        .timeout(1)
+        .interval(1)
+        .build();
+    wait.waitFor();
+    wait.waitFor();
+
+    HoodieException timeout = assertThrows(HoodieException.class, 
wait::waitFor);
+    assertTrue(timeout.getMessage().contains("checkpoint acknowledgement"));
+  }
+
+  @Test
+  void testTimeWaitWrapsInterruptionAndPreservesCause() {
+    TimeWait wait = TimeWait.builder().action("interruptible 
action").interval(1).build();
+    Thread.currentThread().interrupt();
+    try {
+      HoodieException exception = assertThrows(HoodieException.class, 
wait::waitFor);
+      assertTrue(exception.getCause() instanceof InterruptedException);
+      assertTrue(exception.getMessage().contains("interruptible action"));
+    } finally {
+      assertFalse(Thread.currentThread().isInterrupted());
+    }
+  }
+}
diff --git 
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/enumerator/TestHoodieEnumeratorPosition.java
 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/enumerator/TestHoodieEnumeratorPosition.java
new file mode 100644
index 000000000000..5d84a49b7740
--- /dev/null
+++ 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/enumerator/TestHoodieEnumeratorPosition.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source.enumerator;
+
+import org.apache.hudi.common.util.Option;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TestHoodieEnumeratorPosition {
+
+  @Test
+  void testEmptyAndNullStringsProduceEmptyPosition() {
+    HoodieEnumeratorPosition empty = HoodieEnumeratorPosition.empty();
+
+    assertEquals(empty, HoodieEnumeratorPosition.of(null, ""));
+    assertFalse(empty.getIssuedInstant().isPresent());
+    assertFalse(empty.getIssuedOffset().isPresent());
+  }
+
+  @Test
+  void testStringAndOptionFactoriesPreservePosition() {
+    HoodieEnumeratorPosition fromStrings = HoodieEnumeratorPosition.of("001", 
"002");
+    HoodieEnumeratorPosition fromOptions =
+        HoodieEnumeratorPosition.of(Option.of("001"), Option.of("002"));
+
+    assertEquals(fromStrings, fromOptions);
+    assertEquals(fromStrings.hashCode(), fromOptions.hashCode());
+    assertEquals("001", fromStrings.getIssuedInstant().get());
+    assertEquals("002", fromStrings.getIssuedOffset().get());
+    assertTrue(fromStrings.toString().contains("001"));
+    assertNotEquals(fromStrings, HoodieEnumeratorPosition.empty());
+  }
+}
diff --git 
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/prune/TestPrimaryKeyPruners.java
 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/prune/TestPrimaryKeyPruners.java
new file mode 100644
index 000000000000..f96574aec723
--- /dev/null
+++ 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/prune/TestPrimaryKeyPruners.java
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source.prune;
+
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.index.bucket.BucketIdentifier;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.expressions.CallExpression;
+import org.apache.flink.table.expressions.FieldReferenceExpression;
+import org.apache.flink.table.expressions.ResolvedExpression;
+import org.apache.flink.table.expressions.ValueLiteralExpression;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.function.Function;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.mockito.Mockito.mock;
+
+class TestPrimaryKeyPruners {
+
+  @Test
+  void testBucketIdFollowsRecordKeyFieldOrderRegardlessOfFilterOrder() {
+    Configuration conf = new Configuration();
+    conf.set(FlinkOptions.RECORD_KEY_FIELD, "id,tenant");
+    ResolvedExpression tenantFilter = equality("tenant", "tenant-a", true);
+    ResolvedExpression idFilter = equality("id", "id-1", false);
+
+    Function<Integer, Integer> bucketId =
+        PrimaryKeyPruners.getBucketIdFunc(Arrays.asList(tenantFilter, 
idFilter), conf);
+
+    List<String> orderedValues = Arrays.asList("id-1", "tenant-a");
+    assertEquals(BucketIdentifier.getBucketId(orderedValues, 8), 
bucketId.apply(8));
+    assertEquals(BucketIdentifier.getBucketId(orderedValues, 16), 
bucketId.apply(16));
+  }
+
+  @Test
+  void testPartitionBucketIdFunctionIsDisabledWithoutBucketFunction() {
+    assertNull(PartitionBucketIdFunc.create(Option.empty(), 
mock(HoodieTableMetaClient.class), 8));
+  }
+
+  private static ResolvedExpression equality(String field, String value, 
boolean literalFirst) {
+    FieldReferenceExpression fieldReference =
+        new FieldReferenceExpression(field, DataTypes.STRING(), 0, 0);
+    ValueLiteralExpression literal =
+        new ValueLiteralExpression(value, DataTypes.STRING().notNull());
+    return CallExpression.permanent(
+        BuiltInFunctionDefinitions.EQUALS,
+        literalFirst ? Arrays.asList(literal, fieldReference) : 
Arrays.asList(fieldReference, literal),
+        DataTypes.BOOLEAN());
+  }
+}
diff --git 
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/rebalance/TestStreamReadRebalance.java
 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/rebalance/TestStreamReadRebalance.java
new file mode 100644
index 000000000000..9c3b1325db86
--- /dev/null
+++ 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/rebalance/TestStreamReadRebalance.java
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source.rebalance;
+
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.common.util.hash.BucketIndexUtil;
+import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.index.bucket.BucketIdentifier;
+import 
org.apache.hudi.source.rebalance.partitioner.StreamReadAppendPartitioner;
+import 
org.apache.hudi.source.rebalance.partitioner.StreamReadBucketIndexPartitioner;
+import org.apache.hudi.source.rebalance.selector.StreamReadAppendKeySelector;
+import 
org.apache.hudi.source.rebalance.selector.StreamReadBucketIndexKeySelector;
+import org.apache.hudi.table.format.mor.MergeOnReadInputSplit;
+
+import org.apache.flink.configuration.Configuration;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+
+class TestStreamReadRebalance {
+
+  @Test
+  void testAppendSelectorAndPartitionerUseSplitNumber() throws Exception {
+    MergeOnReadInputSplit split = newSplit(7, "partition", "00000003-file");
+
+    Integer key = new StreamReadAppendKeySelector().getKey(split);
+    assertEquals(7, key);
+    assertEquals(3, new StreamReadAppendPartitioner(4).partition(key, 128));
+  }
+
+  @Test
+  void testBucketSelectorAndPartitionerUsePartitionAndFileId() throws 
Exception {
+    String partition = "partition=par1";
+    String fileId = BucketIdentifier.newBucketFileIdPrefix(1);
+    MergeOnReadInputSplit split = newSplit(1, partition, fileId);
+    Pair<String, String> key = new 
StreamReadBucketIndexKeySelector().getKey(split);
+
+    assertEquals(Pair.of(partition, fileId), key);
+
+    Configuration conf = new Configuration();
+    conf.set(FlinkOptions.READ_TASKS, 4);
+    conf.set(FlinkOptions.BUCKET_INDEX_NUM_BUCKETS, 3);
+    StreamReadBucketIndexPartitioner partitioner = new 
StreamReadBucketIndexPartitioner(conf);
+    int actual = partitioner.partition(key, 128);
+    int expected = BucketIndexUtil.getPartitionIndexFunc(4).apply(3, 
partition, 1);
+    assertEquals(expected, actual);
+
+    String otherPartition = "partition=par2";
+    Pair<String, String> otherPartitionKey = Pair.of(otherPartition, fileId);
+    int otherPartitionResult = partitioner.partition(otherPartitionKey, 128);
+    assertEquals(
+        BucketIndexUtil.getPartitionIndexFunc(4).apply(3, otherPartition, 1),
+        otherPartitionResult);
+    assertNotEquals(actual, otherPartitionResult);
+
+    Pair<String, String> otherBucketKey =
+        Pair.of(partition, BucketIdentifier.newBucketFileIdPrefix(2));
+    int otherBucketResult = partitioner.partition(otherBucketKey, 128);
+    assertEquals(
+        BucketIndexUtil.getPartitionIndexFunc(4).apply(3, partition, 2),
+        otherBucketResult);
+    assertNotEquals(actual, otherBucketResult);
+  }
+
+  private static MergeOnReadInputSplit newSplit(int splitNumber, String 
partition, String fileId) {
+    return new MergeOnReadInputSplit(
+        splitNumber,
+        null,
+        Option.empty(),
+        "001",
+        "/tmp/table",
+        1024,
+        "payload_combine",
+        null,
+        fileId,
+        partition);
+  }
+}
diff --git 
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/split/TestSplitRequestEventModel.java
 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/split/TestSplitRequestEventModel.java
new file mode 100644
index 000000000000..5d1f15f22689
--- /dev/null
+++ 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/split/TestSplitRequestEventModel.java
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source.split;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TestSplitRequestEventModel {
+
+  @Test
+  void testAllConstructorsAndAccessors() {
+    SplitRequestEvent empty = new SplitRequestEvent();
+    assertTrue(empty.finishedSplitIds().isEmpty());
+    assertNull(empty.requesterHostname());
+
+    SplitRequestEvent withoutHost = new 
SplitRequestEvent(Collections.singletonList("split-1"));
+    assertEquals(Collections.singletonList("split-1"), 
withoutHost.finishedSplitIds());
+    assertNull(withoutHost.requesterHostname());
+
+    SplitRequestEvent withHost = new SplitRequestEvent(
+        Arrays.asList("split-1", "split-2"), "worker.example");
+    assertEquals(Arrays.asList("split-1", "split-2"), 
withHost.finishedSplitIds());
+    assertEquals("worker.example", withHost.requesterHostname());
+  }
+}
diff --git 
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/stats/TestColumnStatsModels.java
 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/stats/TestColumnStatsModels.java
new file mode 100644
index 000000000000..c4767db0d968
--- /dev/null
+++ 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/stats/TestColumnStatsModels.java
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source.stats;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+class TestColumnStatsModels {
+
+  @Test
+  void testColumnStatsValueSemantics() {
+    ColumnStats stats = new ColumnStats(1, 9, 2);
+
+    assertEquals(1, stats.getMinVal());
+    assertEquals(9, stats.getMaxVal());
+    assertEquals(2, stats.getNullCnt());
+    assertEquals(stats, new ColumnStats(1, 9, 2));
+    assertEquals(stats.hashCode(), new ColumnStats(1, 9, 2).hashCode());
+    assertNotEquals(stats, new ColumnStats(null, 9, 2));
+    assertNull(new ColumnStats(null, null, 0).getMinVal());
+  }
+
+  @Test
+  void testColumnStatsSchemaConstantsResolveExpectedFields() {
+    assertNotNull(ColumnStatsSchemas.METADATA_SCHEMA);
+    assertNotNull(ColumnStatsSchemas.METADATA_DATA_TYPE);
+    assertNotNull(ColumnStatsSchemas.COL_STATS_DATA_TYPE);
+    assertEquals(6, ColumnStatsSchemas.COL_STATS_TARGET_POS.length);
+    int[] expectedSourcePositions = new int[6];
+    expectedSourcePositions[ColumnStatsSchemas.ORD_FILE_NAME] = 0;
+    expectedSourcePositions[ColumnStatsSchemas.ORD_MIN_VAL] = 2;
+    expectedSourcePositions[ColumnStatsSchemas.ORD_MAX_VAL] = 3;
+    expectedSourcePositions[ColumnStatsSchemas.ORD_NULL_CNT] = 5;
+    expectedSourcePositions[ColumnStatsSchemas.ORD_VAL_CNT] = 4;
+    expectedSourcePositions[ColumnStatsSchemas.ORD_COL_NAME] = 1;
+    assertArrayEquals(expectedSourcePositions, 
ColumnStatsSchemas.COL_STATS_TARGET_POS);
+  }
+}
diff --git 
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/TestFormatIteratorsAndIOFactory.java
 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/TestFormatIteratorsAndIOFactory.java
new file mode 100644
index 000000000000..911daab8b7fc
--- /dev/null
+++ 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/TestFormatIteratorsAndIOFactory.java
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.table.format;
+
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.util.collection.ClosableIterator;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.io.storage.row.HoodieRowDataFileWriterFactory;
+import org.apache.hudi.storage.HoodieStorage;
+import 
org.apache.hudi.table.format.cow.vector.reader.ParquetColumnarRowSplitReader;
+import org.apache.hudi.util.RowDataProjection;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class TestFormatIteratorsAndIOFactory {
+
+  @Test
+  void testParquetSplitIteratorDelegatesLifecycle() throws Exception {
+    ParquetColumnarRowSplitReader reader = 
mock(ParquetColumnarRowSplitReader.class);
+    RowData row = GenericRowData.of(1);
+    when(reader.reachedEnd()).thenReturn(false, true);
+    when(reader.nextRecord()).thenReturn(row);
+    ParquetSplitRecordIterator iterator = new 
ParquetSplitRecordIterator(reader);
+
+    assertTrue(iterator.hasNext());
+    assertSame(row, iterator.next());
+    assertFalse(iterator.hasNext());
+    iterator.close();
+    verify(reader).close();
+  }
+
+  @Test
+  void testParquetSplitIteratorWrapsReaderIoFailures() throws Exception {
+    ParquetColumnarRowSplitReader reader = 
mock(ParquetColumnarRowSplitReader.class);
+    when(reader.reachedEnd()).thenThrow(new IOException("read failure"));
+    ParquetSplitRecordIterator iterator = new 
ParquetSplitRecordIterator(reader);
+
+    HoodieIOException readFailure = assertThrows(HoodieIOException.class, 
iterator::hasNext);
+    assertInstanceOf(IOException.class, readFailure.getCause());
+
+    org.mockito.Mockito.doThrow(new IOException("close 
failure")).when(reader).close();
+    HoodieIOException closeFailure = assertThrows(HoodieIOException.class, 
iterator::close);
+    assertInstanceOf(IOException.class, closeFailure.getCause());
+  }
+
+  @Test
+  @SuppressWarnings("unchecked")
+  void testSchemaEvolvedIteratorProjectsAndClosesNestedIterator() {
+    ClosableIterator<RowData> nested = mock(ClosableIterator.class);
+    GenericRowData input = GenericRowData.of(1, 2);
+    when(nested.hasNext()).thenReturn(true);
+    when(nested.next()).thenReturn(input);
+    LogicalType[] projectedTypes = {DataTypes.INT().getLogicalType()};
+    RowDataProjection projection = RowDataProjection.instance(projectedTypes, 
new int[] {1});
+    SchemaEvolvedRecordIterator iterator = new 
SchemaEvolvedRecordIterator(nested, projection);
+
+    assertTrue(iterator.hasNext());
+    assertEquals(2, iterator.next().getInt(0));
+    iterator.close();
+    verify(nested).close();
+  }
+
+  @Test
+  void testFlinkIoFactoryCreatesRowDataFactories() {
+    HoodieStorage storage = mock(HoodieStorage.class);
+    HoodieFlinkIOFactory factory = new HoodieFlinkIOFactory(storage);
+
+    assertInstanceOf(HoodieRowDataFileWriterFactory.class,
+        factory.getWriterFactory(HoodieRecord.HoodieRecordType.FLINK));
+    assertInstanceOf(HoodieRowDataFileReaderFactory.class,
+        factory.getReaderFactory(HoodieRecord.HoodieRecordType.FLINK));
+  }
+}
diff --git 
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/mor/TestMergeOnReadTableState.java
 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/mor/TestMergeOnReadTableState.java
new file mode 100644
index 000000000000..e7f64f1237ca
--- /dev/null
+++ 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/mor/TestMergeOnReadTableState.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.table.format.mor;
+
+import org.apache.hudi.common.model.HoodieRecord;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.types.logical.RowType;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+class TestMergeOnReadTableState {
+
+  @Test
+  void testStateExposesSchemasSplitsAndRequiredPositions() {
+    RowType rowType = (RowType) DataTypes.ROW(
+        DataTypes.FIELD("id", DataTypes.STRING()),
+        DataTypes.FIELD(HoodieRecord.OPERATION_METADATA_FIELD, 
DataTypes.STRING()),
+        DataTypes.FIELD("name", DataTypes.STRING())).getLogicalType();
+    RowType requiredRowType = (RowType) DataTypes.ROW(
+        DataTypes.FIELD("name", DataTypes.STRING()),
+        DataTypes.FIELD("id", DataTypes.STRING())).getLogicalType();
+    MergeOnReadTableState<String> state = new MergeOnReadTableState<>(
+        rowType, requiredRowType, "table-schema", "required-schema", 
Collections.singletonList("split"));
+
+    assertSame(rowType, state.getRowType());
+    assertSame(requiredRowType, state.getRequiredRowType());
+    assertEquals("table-schema", state.getTableSchema());
+    assertEquals("required-schema", state.getRequiredSchema());
+    assertEquals(Collections.singletonList("split"), state.getInputSplits());
+    assertEquals(1, state.getOperationPos());
+    assertArrayEquals(new int[] {2, 0}, state.getRequiredPositions());
+  }
+
+  @Test
+  void testMissingRequiredFieldIsReportedAsNegativePosition() {
+    RowType rowType = RowType.of(DataTypes.INT().getLogicalType(), 
DataTypes.STRING().getLogicalType());
+    RowType required = new RowType(Collections.singletonList(
+        new RowType.RowField("missing", DataTypes.STRING().getLogicalType())));
+    MergeOnReadTableState<Integer> state =
+        new MergeOnReadTableState<>(rowType, required, "schema", "required", 
Arrays.asList(1, 2));
+
+    assertEquals(-1, state.getOperationPos());
+    assertArrayEquals(new int[] {-1}, state.getRequiredPositions());
+  }
+}
diff --git 
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/lookup/TestLookupUtilities.java
 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/lookup/TestLookupUtilities.java
new file mode 100644
index 000000000000..1d714bfe4496
--- /dev/null
+++ 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/lookup/TestLookupUtilities.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.table.lookup;
+
+import 
org.apache.flink.table.connector.source.lookup.AsyncLookupFunctionProvider;
+import org.apache.flink.table.connector.source.lookup.LookupFunctionProvider;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.mockito.Mockito.mock;
+
+class TestLookupUtilities {
+
+  @Test
+  void testHeapLookupCacheStoresDuplicateKeysAndClears() throws Exception {
+    HeapLookupCache cache = new HeapLookupCache();
+    RowData key = GenericRowData.of(1);
+    RowData first = GenericRowData.of("first");
+    RowData second = GenericRowData.of("second");
+
+    assertNull(cache.getRows(key));
+    cache.addRow(key, first);
+    cache.addRow(key, second);
+    List<RowData> rows = cache.getRows(key);
+    assertEquals(2, rows.size());
+    assertEquals(first, rows.get(0));
+    assertEquals(second, rows.get(1));
+
+    cache.clear();
+    assertNull(cache.getRows(key));
+    cache.addRow(key, first);
+    cache.close();
+    assertNull(cache.getRows(key));
+  }
+
+  @Test
+  void testLookupRuntimeProviderFactorySelectsSyncAndAsyncProviders() {
+    HoodieLookupFunction function = mock(HoodieLookupFunction.class);
+
+    assertInstanceOf(LookupFunctionProvider.class,
+        LookupRuntimeProviderFactory.create(function, false, 1));
+    assertInstanceOf(AsyncLookupFunctionProvider.class,
+        LookupRuntimeProviderFactory.create(function, true, 3));
+  }
+}
diff --git 
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/util/TestFlinkUtilities.java
 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/util/TestFlinkUtilities.java
new file mode 100644
index 000000000000..c883076ed04d
--- /dev/null
+++ 
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/util/TestFlinkUtilities.java
@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.util;
+
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.table.format.CastMap;
+
+import org.apache.flink.api.common.io.InputFormat;
+import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
+import org.apache.flink.core.io.GenericInputSplit;
+import org.apache.flink.formats.json.JsonRowDataDeserializationSchema;
+import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.connector.sink.abilities.SupportsRowLevelDelete;
+import org.apache.flink.table.connector.sink.abilities.SupportsRowLevelUpdate;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.types.RowKind;
+import org.junit.jupiter.api.Test;
+
+import java.util.NoSuchElementException;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class TestFlinkUtilities {
+
+  @Test
+  void testStateBackendConverterHandlesSupportedAndUnknownValues() {
+    FlinkStateBackendConverter converter = new FlinkStateBackendConverter();
+
+    assertInstanceOf(HashMapStateBackend.class, converter.convert("hashmap"));
+    assertInstanceOf(EmbeddedRocksDBStateBackend.class, 
converter.convert("rocksdb"));
+    HoodieException exception = assertThrows(HoodieException.class, () -> 
converter.convert("memory"));
+    assertTrue(exception.getMessage().contains("memory"));
+  }
+
+  @Test
+  @SuppressWarnings("unchecked")
+  void testEmptyInputFormatContainsNoRecords() throws Exception {
+    InputFormat<RowData, GenericInputSplit> inputFormat =
+        (InputFormat<RowData, GenericInputSplit>) 
InputFormats.EMPTY_INPUT_FORMAT;
+    GenericInputSplit[] splits = inputFormat.createInputSplits(2);
+
+    assertEquals(1, splits.length);
+    inputFormat.open(splits[0]);
+    assertTrue(inputFormat.reachedEnd());
+    assertThrows(NoSuchElementException.class, () -> 
inputFormat.nextRecord(null));
+    inputFormat.close();
+  }
+
+  @Test
+  void testRowDataProjectionPreservesKindNullsAndSelectedOrder() {
+    RowType rowType = RowType.of(
+        DataTypes.INT().getLogicalType(),
+        DataTypes.STRING().getLogicalType());
+    GenericRowData input = GenericRowData.of(7, null);
+    input.setRowKind(RowKind.DELETE);
+
+    RowDataProjection projection = RowDataProjection.instanceV2(rowType, new 
int[] {1, 0});
+    RowData projected = projection.project(input);
+
+    assertEquals(RowKind.DELETE, projected.getRowKind());
+    assertTrue(projected.isNullAt(0));
+    assertEquals(7, projected.getInt(1));
+    assertArrayEquals(new Object[] {null, 7}, 
projection.projectAsValues(input));
+  }
+
+  @Test
+  void testRowDataProjectionFactoriesAndValidation() {
+    LogicalType[] types = {
+        DataTypes.INT().getLogicalType(),
+        DataTypes.STRING().getLogicalType()
+    };
+    RowType rowType = RowType.of(types);
+    GenericRowData input = GenericRowData.of(3, 
StringData.fromString("value"));
+
+    RowData projected = RowDataProjection.instance(rowType, new int[] {0, 
1}).project(input);
+    assertEquals(3, projected.getInt(0));
+    assertEquals("value", projected.getString(1).toString());
+    assertThrows(IllegalArgumentException.class,
+        () -> RowDataProjection.instance(types, new int[] {0}));
+  }
+
+  @Test
+  void testCastProjectionHandlesValuesAndNulls() {
+    LogicalType[] types = {
+        DataTypes.INT().getLogicalType(),
+        DataTypes.STRING().getLogicalType()
+    };
+    RowDataCastProjection projection = new RowDataCastProjection(types, new 
CastMap());
+    GenericRowData input = GenericRowData.of(11, null);
+
+    RowData projected = projection.project(input);
+    assertEquals(11, projected.getInt(0));
+    assertTrue(projected.isNullAt(1));
+  }
+
+  @Test
+  void testSharedModificationAndChangelogConstants() {
+    assertTrue(ChangelogModes.FULL.contains(RowKind.UPDATE_BEFORE));
+    assertFalse(ChangelogModes.UPSERT.contains(RowKind.UPDATE_BEFORE));
+    assertTrue(ChangelogModes.UPSERT.contains(RowKind.DELETE));
+    assertEquals(
+        SupportsRowLevelDelete.RowLevelDeleteMode.DELETED_ROWS,
+        DataModificationInfos.DEFAULT_DELETE_INFO.getRowLevelDeleteMode());
+    
assertTrue(DataModificationInfos.DEFAULT_DELETE_INFO.requiredColumns().isEmpty());
+    assertEquals(
+        SupportsRowLevelUpdate.RowLevelUpdateMode.UPDATED_ROWS,
+        DataModificationInfos.DEFAULT_UPDATE_INFO.getRowLevelUpdateMode());
+    
assertTrue(DataModificationInfos.DEFAULT_UPDATE_INFO.requiredColumns().isEmpty());
+  }
+
+  @Test
+  void testJsonDeserializationFunctionFactoriesAndLifecycle() throws Exception 
{
+    RowType rowType = RowType.of(DataTypes.STRING().getLogicalType());
+    assertInstanceOf(JsonDeserializationFunction.class, 
JsonDeserializationFunction.getInstance(rowType));
+
+    JsonRowDataDeserializationSchema schema = 
mock(JsonRowDataDeserializationSchema.class);
+    RowData expected = GenericRowData.of(StringData.fromString("value"));
+    when(schema.deserialize(any(byte[].class))).thenReturn(expected);
+    JsonDeserializationFunction function = new 
JsonDeserializationFunction(schema);
+
+    function.open(new org.apache.flink.configuration.Configuration());
+    assertSame(expected, function.map("{\"f0\":\"value\"}"));
+    verify(schema).open(null);
+  }
+}