This is an automated email from the ASF dual-hosted git repository. danny0405 pushed a commit to branch release-1.1.2 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 10162c55beda22b13ec98e51ef829f2b064b8855 Author: Shuo Cheng <[email protected]> AuthorDate: Thu Sep 24 10:10:35 2026 +0800 fix(flink): avoid blocking existing instant lookups on instant creation (#20031) (cherry picked from commit 20037d3c3f0e58de450c61b3fa4edd57e1a24935) --- .../hudi/sink/StreamWriteOperatorCoordinator.java | 6 ++---- .../sink/TestStreamWriteOperatorCoordinator.java | 25 ++++++++++++++++++++++ 2 files changed, 27 insertions(+), 4 deletions(-) diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java index 22ef43e40a35..251bbc94ea55 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java @@ -391,12 +391,10 @@ public class StreamWriteOperatorCoordinator } private CompletableFuture<CoordinationResponse> handleInstantRequest(Correspondent.InstantTimeRequest request) { - if (instantRequestExecutor.hasRunningTasks()) { - return CompletableFuture.completedFuture(CoordinationResponseSerDe.wrap(Correspondent.InstantTimeResponse.getInstance(null))); - } long checkpointId = request.getCheckpointId(); + // Existing instants must remain available while another checkpoint's creation is blocked. Pair<String, WriteMetadataEvent[]> instantTimeAndEventBuffer = this.eventBuffers.getInstantAndEventBuffer(checkpointId); - if (instantTimeAndEventBuffer == null) { + if (instantTimeAndEventBuffer == null && !instantRequestExecutor.hasRunningTasks()) { instantRequestExecutor.execute(() -> { if (this.eventBuffers.getInstantAndEventBuffer(checkpointId) == null) { // Wait until previous instants are committed. diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java index 9815878eff99..e74495348bad 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java @@ -722,6 +722,31 @@ public class TestStreamWriteOperatorCoordinator { assertTrue(inflights.containsInstant(instant)); } + @Test + void testReadyInstantRemainsAvailableWhileAnotherCreationIsBlocked() throws Exception { + String existingInstant = requestInstantTime(1L); + MockOperatorCoordinatorContext ctx = (MockOperatorCoordinatorContext) coordinator.getContext(); + CountDownLatch gate = new CountDownLatch(1); + NonThrownExecutor worker = new GatedInstantRequestExecutor( + Mockito.mock(Logger.class), + (errMsg, t) -> ctx.failJob(new HoodieException(errMsg, t)), gate); + coordinator.setInstantRequestExecutor(worker); + try { + Correspondent.InstantTimeResponse pending = CoordinationResponseSerDe.unwrap( + coordinator.handleCoordinationRequest(Correspondent.InstantTimeRequest.getInstance(2L)) + .get(1, TimeUnit.SECONDS)); + assertNull(pending.getInstant()); + assertTrue(worker.hasRunningTasks()); + Correspondent.InstantTimeResponse ready = CoordinationResponseSerDe.unwrap( + coordinator.handleCoordinationRequest(Correspondent.InstantTimeRequest.getInstance(1L)) + .get(1, TimeUnit.SECONDS)); + assertEquals(existingInstant, ready.getInstant(), + "An existing instant must remain available while another checkpoint's creation is blocked"); + } finally { + gate.countDown(); + } + } + // ------------------------------------------------------------------------- // Utilities // -------------------------------------------------------------------------
