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 b799fd41c1fdb7d13ff9b30d70a1b8573c11a73e Author: fhan <[email protected]> AuthorDate: Wed Sep 23 15:15:27 2026 +0800 fix(flink): serve instant-time requests asynchronously to avoid coordination RPC timeout (#19960) * fix(flink): serve instant-time requests asynchronously to avoid coordination RPC timeout * refactor(flink): simplify asynchronous instant polling --------- Co-authored-by: fhan <[email protected]> Co-authored-by: danny0405 <[email protected]> (cherry picked from commit 7430134280ae5684daee64ece7660a77577bf65f) --- .../hudi/sink/StreamWriteOperatorCoordinator.java | 41 ++++--- .../hudi/sink/bulk/BulkInsertWriteFunction.java | 3 +- .../sink/common/AbstractStreamWriteFunction.java | 3 +- .../org/apache/hudi/sink/event/Correspondent.java | 63 ++++++++++- .../apache/hudi/sink/utils/NonThrownExecutor.java | 24 +++- .../sink/TestStreamWriteOperatorCoordinator.java | 70 ++++++++++-- .../common/TestAbstractStreamWriteFunction.java | 15 ++- .../sink/event/TestCorrespondentEventModels.java | 126 +++++++++++++++++++++ .../utils/BucketStreamWriteFunctionWrapper.java | 1 + .../hudi/sink/utils/BulkInsertFunctionWrapper.java | 2 + .../hudi/sink/utils/InsertFunctionWrapper.java | 2 + .../apache/hudi/sink/utils/MockCorrespondent.java | 10 +- .../sink/utils/MockCorrespondentWithTimeout.java | 12 +- .../sink/utils/StreamWriteFunctionWrapper.java | 2 + .../hudi/sink/utils/TestNonThrownExecutor.java | 84 ++++++++++++++ 15 files changed, 402 insertions(+), 56 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 60ece5a1a2f1..22ef43e40a35 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 @@ -384,23 +384,30 @@ public class StreamWriteOperatorCoordinator @Override public CompletableFuture<CoordinationResponse> handleCoordinationRequest(CoordinationRequest request) { - CompletableFuture<CoordinationResponse> response = new CompletableFuture<>(); - instantRequestExecutor.execute(() -> { - Correspondent.InstantTimeRequest instantTimeRequest = (Correspondent.InstantTimeRequest) request; - long checkpointId = instantTimeRequest.getCheckpointId(); - Pair<String, WriteMetadataEvent[]> instantTimeAndEventBuffer = this.eventBuffers.getInstantAndEventBuffer(checkpointId); - final String instantTime; - if (instantTimeAndEventBuffer == null) { - // wait until previous instants are committed. - eventBuffers.awaitAllInstantsToCompleteIfNecessary(); - instantTime = startInstant(); - this.eventBuffers.initNewEventBuffer(checkpointId, instantTime, this.parallelism); - } else { - instantTime = instantTimeAndEventBuffer.getLeft(); - } - response.complete(CoordinationResponseSerDe.wrap(Correspondent.InstantTimeResponse.getInstance(instantTime))); - }, "request instant time"); - return response; + if (request instanceof Correspondent.InstantTimeRequest) { + return handleInstantRequest((Correspondent.InstantTimeRequest) request); + } + throw new HoodieException("Unexpected coordination request type: " + request.getClass().getSimpleName()); + } + + private CompletableFuture<CoordinationResponse> handleInstantRequest(Correspondent.InstantTimeRequest request) { + if (instantRequestExecutor.hasRunningTasks()) { + return CompletableFuture.completedFuture(CoordinationResponseSerDe.wrap(Correspondent.InstantTimeResponse.getInstance(null))); + } + long checkpointId = request.getCheckpointId(); + Pair<String, WriteMetadataEvent[]> instantTimeAndEventBuffer = this.eventBuffers.getInstantAndEventBuffer(checkpointId); + if (instantTimeAndEventBuffer == null) { + instantRequestExecutor.execute(() -> { + if (this.eventBuffers.getInstantAndEventBuffer(checkpointId) == null) { + // Wait until previous instants are committed. + eventBuffers.awaitAllInstantsToCompleteIfNecessary(); + this.eventBuffers.initNewEventBuffer(checkpointId, startInstant(), this.parallelism); + } + }, "request instant time"); + instantTimeAndEventBuffer = this.eventBuffers.getInstantAndEventBuffer(checkpointId); + } + String instantTime = instantTimeAndEventBuffer == null ? null : instantTimeAndEventBuffer.getLeft(); + return CompletableFuture.completedFuture(CoordinationResponseSerDe.wrap(Correspondent.InstantTimeResponse.getInstance(instantTime))); } // ------------------------------------------------------------------------- diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/BulkInsertWriteFunction.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/BulkInsertWriteFunction.java index 30f2aa7abd68..51ec432fa3aa 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/BulkInsertWriteFunction.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/BulkInsertWriteFunction.java @@ -21,6 +21,7 @@ package org.apache.hudi.sink.bulk; import org.apache.hudi.client.HoodieFlinkWriteClient; import org.apache.hudi.client.WriteStatus; import org.apache.hudi.common.model.WriteOperationType; +import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.sink.StreamWriteOperatorCoordinator; import org.apache.hudi.sink.common.AbstractWriteFunction; import org.apache.hudi.sink.event.Correspondent; @@ -173,6 +174,6 @@ public class BulkInsertWriteFunction<I> * Returns the instant to write. */ private String instantToWrite() { - return this.correspondent.requestInstantTime(-1L); + return this.correspondent.requestInstantTime(-1L, this.config.get(FlinkOptions.WRITE_COMMIT_ACK_TIMEOUT)); } } diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/common/AbstractStreamWriteFunction.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/common/AbstractStreamWriteFunction.java index 0884ea47517a..cc72da9bd12b 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/common/AbstractStreamWriteFunction.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/common/AbstractStreamWriteFunction.java @@ -23,6 +23,7 @@ import org.apache.hudi.client.WriteStatus; import org.apache.hudi.common.table.HoodieTableMetaClient; import org.apache.hudi.common.table.timeline.HoodieTimeline; import org.apache.hudi.common.util.ValidationUtils; +import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.sink.StreamWriteOperatorCoordinator; import org.apache.hudi.sink.event.CommitAckEvent; import org.apache.hudi.sink.event.Correspondent; @@ -242,6 +243,6 @@ public abstract class AbstractStreamWriteFunction<I> * @return The instant time */ protected String instantToWrite(boolean hasData) { - return this.correspondent.requestInstantTime(this.checkpointId); + return this.correspondent.requestInstantTime(this.checkpointId, this.config.get(FlinkOptions.WRITE_COMMIT_ACK_TIMEOUT)); } } diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/event/Correspondent.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/event/Correspondent.java index e4dd405bd40e..ce3b3758f725 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/event/Correspondent.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/event/Correspondent.java @@ -28,10 +28,23 @@ import org.apache.flink.runtime.operators.coordination.CoordinationRequest; import org.apache.flink.runtime.operators.coordination.CoordinationResponse; import org.apache.flink.util.SerializedValue; +import java.util.concurrent.TimeUnit; + /** * Correspondent between a write task with the coordinator. */ public class Correspondent { + + /** + * Initial backoff in milliseconds between two instant-time polls, allowing for instant creation latency. + */ + private static final long POLL_BASE_MS = 200L; + + /** + * Upper bound in milliseconds of the backoff between two instant-time polls. + */ + private static final long POLL_CAP_MS = 1000L; + private final OperatorID operatorID; private final TaskOperatorEventGateway gateway; @@ -59,16 +72,50 @@ public class Correspondent { } /** - * Sends a request to the coordinator to fetch the instant time. + * Requests the instant time for the given checkpoint from the coordinator. + * + * <p>Polls with capped exponential backoff until the instant is non-null or the timeout expires. + * Request failures are propagated immediately. + * + * @param checkpointId The checkpoint id (or -1 for bulk insert) + * @param pollBudgetMs The overall budget to wait for an instant, in milliseconds + * + * @return the instant time to write with */ - public String requestInstantTime(long checkpointId) { + public String requestInstantTime(long checkpointId, long pollBudgetMs) { + final long deadlineNanos = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(pollBudgetMs); + long backoffMs = POLL_BASE_MS; try { - InstantTimeResponse response = CoordinationResponseSerDe.unwrap(this.gateway.sendRequestToCoordinator(this.operatorID, - new SerializedValue<>(InstantTimeRequest.getInstance(checkpointId))).get()); - return response.getInstant(); + do { + String instant = fetchInstantTimeResponse(checkpointId).getInstant(); + if (instant != null) { + return instant; + } + long remainingNanos = deadlineNanos - System.nanoTime(); + if (remainingNanos <= 0) { + break; + } + TimeUnit.NANOSECONDS.sleep(Math.min(remainingNanos, TimeUnit.MILLISECONDS.toNanos(backoffMs))); + backoffMs = Math.min(backoffMs * 2, POLL_CAP_MS); + } while (System.nanoTime() < deadlineNanos); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new HoodieException("Interrupted while requesting the instant time from the coordinator", e); } catch (Exception e) { - throw new HoodieException("Error requesting the instant time from the coordinator", e); + throw new HoodieException( + "Error requesting the instant time from the coordinator for checkpoint " + checkpointId, e); } + throw new HoodieException("Timeout waiting for the instant time from the coordinator for checkpoint " + checkpointId); + } + + /** + * Sends a single instant-time request to the coordinator and returns its response. + * + * <p>Isolated so tests can stub the transport while reusing the poll loop in {@link #requestInstantTime}. + */ + protected InstantTimeResponse fetchInstantTimeResponse(long checkpointId) throws Exception { + return CoordinationResponseSerDe.unwrap(this.gateway.sendRequestToCoordinator(this.operatorID, + new SerializedValue<>(InstantTimeRequest.getInstance(checkpointId))).get()); } public OperatorID getOperatorID() { @@ -102,6 +149,10 @@ public class Correspondent { * A response with instant time. */ public static class InstantTimeResponse implements CoordinationResponse { + + /** + * The instant time, or null while the instant is still being created. + */ private final String instant; private InstantTimeResponse(String instant) { diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/NonThrownExecutor.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/NonThrownExecutor.java index 4da4a7718a67..82d3193e874f 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/NonThrownExecutor.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/NonThrownExecutor.java @@ -29,8 +29,10 @@ import java.util.Objects; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Supplier; /** @@ -47,6 +49,8 @@ public class NonThrownExecutor implements AutoCloseable { */ private final ExecutorService executor; + private final AtomicInteger pendingTasks = new AtomicInteger(); + /** * Exception hook for post-exception handling. */ @@ -88,7 +92,12 @@ public class NonThrownExecutor implements AutoCloseable { final ExceptionHook hook, final String actionName, final Object... actionParams) { - executor.execute(wrapAction(action, hook, actionName, actionParams)); + try { + executor.execute(wrapAction(action, hook, actionName, actionParams)); + } catch (RejectedExecutionException e) { + pendingTasks.decrementAndGet(); + handleException(e, hook, getActionString(actionName, actionParams)); + } } /** @@ -97,6 +106,9 @@ public class NonThrownExecutor implements AutoCloseable { public void executeSync(ThrowingRunnable<Throwable> action, String actionName, Object... actionParams) { try { executor.submit(wrapAction(action, this.exceptionHook, actionName, actionParams)).get(); + } catch (RejectedExecutionException e) { + pendingTasks.decrementAndGet(); + handleException(e, this.exceptionHook, getActionString(actionName, actionParams)); } catch (InterruptedException e) { handleException(e, this.exceptionHook, getActionString(actionName, actionParams)); } catch (ExecutionException e) { @@ -105,6 +117,13 @@ public class NonThrownExecutor implements AutoCloseable { } } + /** + * Returns whether any task is queued or running. + */ + public boolean hasRunningTasks() { + return pendingTasks.get() > 0; + } + @Override public void close() throws Exception { if (executor != null) { @@ -125,6 +144,7 @@ public class NonThrownExecutor implements AutoCloseable { final String actionName, final Object... actionParams) { + pendingTasks.incrementAndGet(); return () -> { final Supplier<String> actionString = getActionString(actionName, actionParams); try { @@ -132,6 +152,8 @@ public class NonThrownExecutor implements AutoCloseable { logger.info("Executor executes action [{}] success!", actionString.get()); } catch (Throwable t) { handleException(t, hook, actionString); + } finally { + pendingTasks.decrementAndGet(); } }; } 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 c2e7a0da6ef8..9815878eff99 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 @@ -41,6 +41,7 @@ import org.apache.hudi.sink.event.Correspondent; import org.apache.hudi.sink.event.WriteMetadataEvent; import org.apache.hudi.sink.utils.CoordinationResponseSerDe; import org.apache.hudi.sink.utils.MockCoordinatorExecutor; +import org.apache.hudi.sink.utils.MockCorrespondent; import org.apache.hudi.sink.utils.NonThrownExecutor; import org.apache.hudi.storage.HoodieStorage; import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration; @@ -54,6 +55,7 @@ import org.apache.flink.runtime.operators.coordination.MockOperatorCoordinatorCo import org.apache.flink.runtime.operators.coordination.OperatorCoordinator; import org.apache.flink.runtime.operators.coordination.OperatorEvent; import org.apache.flink.util.FileUtils; +import org.apache.flink.util.function.ThrowingRunnable; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.junit.jupiter.api.AfterEach; @@ -70,6 +72,7 @@ import java.io.File; import java.io.IOException; import java.util.Collections; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import static org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_GENERATOR; @@ -303,8 +306,8 @@ public class TestStreamWriteOperatorCoordinator { coordinator.getEventBuffer()[0].getWriteStatuses().size(), is(1)); long nextCkpId = 1; - coordinator.handleCoordinationRequest(Correspondent.InstantTimeRequest.getInstance(nextCkpId)); - OperatorEvent event4 = createOperatorEvent(0, nextCkpId, "002", "par1", false, false, 0.1); + String instant2 = requestInstantTime(nextCkpId); + OperatorEvent event4 = createOperatorEvent(0, nextCkpId, instant2, "par1", false, false, 0.1); coordinator.handleEventFromOperator(0, event4); assertThat("First instant is not committed yet, new event should not override the old event", coordinator.getEventBuffer(0)[0].getWriteStatuses().size(), is(1)); @@ -685,6 +688,40 @@ public class TestStreamWriteOperatorCoordinator { } } + @Test + void testInstantRequestPollsWhileCreationBlockedThenSucceeds() throws Exception { + MockOperatorCoordinatorContext ctx = (MockOperatorCoordinatorContext) coordinator.getContext(); + CountDownLatch lockHeld = new CountDownLatch(1); + NonThrownExecutor gatedWorker = Mockito.spy(new GatedInstantRequestExecutor( + Mockito.mock(Logger.class), + (errMsg, t) -> ctx.failJob(new HoodieException(errMsg, t)), + lockHeld)); + coordinator.setInstantRequestExecutor(gatedWorker); + + try { + // The first request starts creation; subsequent requests must not queue more work while it is blocked. + for (long checkpointId : new long[] {1L, 1L, 2L}) { + Correspondent.InstantTimeResponse pending = CoordinationResponseSerDe.unwrap( + coordinator.handleCoordinationRequest(Correspondent.InstantTimeRequest.getInstance(checkpointId)) + .get(1, TimeUnit.SECONDS)); + assertNull(pending.getInstant()); + } + assertTrue(gatedWorker.hasRunningTasks()); + Mockito.verify(gatedWorker, Mockito.times(1)).execute(Mockito.any(), Mockito.eq("request instant time")); + } finally { + lockHeld.countDown(); + } + + String instant = requestInstantTime(1L); + assertNotNull(instant); + assertEquals(instant, requestInstantTime(1L), "Repeated requests must reuse the instant"); + assertFalse(ctx.isJobFailed()); + HoodieTimeline inflights = StreamerUtil.createMetaClient(TestConfigurations.getDefaultConf(tempFile.getAbsolutePath())) + .reloadActiveTimeline().filterInflights(); + assertEquals(1, inflights.countInstants()); + assertTrue(inflights.containsInstant(instant)); + } + // ------------------------------------------------------------------------- // Utilities // ------------------------------------------------------------------------- @@ -694,12 +731,8 @@ public class TestStreamWriteOperatorCoordinator { } private String requestInstantTime(StreamWriteOperatorCoordinator coordinator, long checkpointId) { - try { - Correspondent.InstantTimeResponse response = CoordinationResponseSerDe.unwrap(coordinator.handleCoordinationRequest(Correspondent.InstantTimeRequest.getInstance(checkpointId)).get()); - return response.getInstant(); - } catch (Exception e) { - throw new HoodieException("Error requesting the instant time from the coordinator", e); - } + return new MockCorrespondent(coordinator) + .requestInstantTime(checkpointId, TimeUnit.SECONDS.toMillis(10)); } private void resetToMergeOnRead(Configuration conf) throws Exception { @@ -727,6 +760,27 @@ public class TestStreamWriteOperatorCoordinator { return new StreamWriteOperatorCoordinator(conf, coordinatorContext); } + /** + * A real single-thread instant-request worker whose submitted creation task blocks on a latch before + * running, to deterministically delay instant creation (simulating a held table lock). + */ + private static final class GatedInstantRequestExecutor extends NonThrownExecutor { + private final CountDownLatch gate; + + private GatedInstantRequestExecutor(Logger logger, ExceptionHook exceptionHook, CountDownLatch gate) { + super(logger, null, exceptionHook, true); + this.gate = gate; + } + + @Override + public void execute(ThrowingRunnable<Throwable> action, String actionName, Object... actionParams) { + super.execute(() -> { + gate.await(); + action.run(); + }, actionName, actionParams); + } + } + private String mockWriteWithMetadata(long checkpointId) { String instant = requestInstantTime(checkpointId); OperatorEvent event = createOperatorEvent(0, checkpointId, instant, "par1", false, true, 0.1); diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/common/TestAbstractStreamWriteFunction.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/common/TestAbstractStreamWriteFunction.java index b87ee0aa0a95..bfd9e06184fd 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/common/TestAbstractStreamWriteFunction.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/common/TestAbstractStreamWriteFunction.java @@ -21,6 +21,7 @@ package org.apache.hudi.sink.common; import org.apache.hudi.client.HoodieFlinkWriteClient; import org.apache.hudi.common.table.HoodieTableMetaClient; import org.apache.hudi.common.table.timeline.HoodieTimeline; +import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.sink.event.Correspondent; import org.apache.hudi.sink.event.WriteMetadataEvent; import org.apache.hudi.sink.utils.MockOperatorStateStore; @@ -84,7 +85,7 @@ class TestAbstractStreamWriteFunction { initialize(-1L, attempt); assertEquals("002", function.instantToWrite(true)); - verify(correspondent).requestInstantTime(-1L); + verify(correspondent).requestInstantTime(-1L, instantRequestPollBudget()); if (attempt == 0) { assertTrue(events.isEmpty()); } else { @@ -103,7 +104,7 @@ class TestAbstractStreamWriteFunction { initialize(42L, attempt); assertEquals("002", function.instantToWrite(true)); - verify(correspondent).requestInstantTime(42L); + verify(correspondent).requestInstantTime(42L, instantRequestPollBudget()); if (attempt == 0) { assertTrue(events.isEmpty()); } else { @@ -116,7 +117,7 @@ class TestAbstractStreamWriteFunction { assertEquals("002", snapshot.getInstantTime()); assertTrue(snapshot.isBootstrap()); function.instantToWrite(true); - verify(correspondent).requestInstantTime(43L); + verify(correspondent).requestInstantTime(43L, instantRequestPollBudget()); } @Test @@ -140,7 +141,7 @@ class TestAbstractStreamWriteFunction { assertEquals(41L, bootstrap.getCheckpointId()); assertEquals("001", bootstrap.getInstantTime()); function.instantToWrite(true); - verify(correspondent).requestInstantTime(42L); + verify(correspondent).requestInstantTime(42L, instantRequestPollBudget()); } private void initialize(long checkpointId, int attempt) throws Exception { @@ -149,7 +150,7 @@ class TestAbstractStreamWriteFunction { when(context.getOperatorStateStore()).thenReturn(stateStore); when(context.isRestored()).thenReturn(checkpointId >= 0); when(context.getRestoredCheckpointId()).thenReturn(checkpointId >= 0 ? OptionalLong.of(checkpointId) : OptionalLong.empty()); - when(correspondent.requestInstantTime(anyLong())).thenReturn("002"); + when(correspondent.requestInstantTime(anyLong(), anyLong())).thenReturn("002"); function.setRuntimeContext(runtimeContext); function.setCorrespondent(correspondent); function.setOperatorEventGateway(events::add); @@ -168,6 +169,10 @@ class TestAbstractStreamWriteFunction { } } + private long instantRequestPollBudget() { + return conf.get(FlinkOptions.WRITE_COMMIT_ACK_TIMEOUT); + } + private void assertCleanupEvent(long checkpointId) { assertEquals(1, events.size()); WriteMetadataEvent bootstrap = (WriteMetadataEvent) events.get(0); 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..b8658505d25d --- /dev/null +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/event/TestCorrespondentEventModels.java @@ -0,0 +1,126 @@ +/* + * 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.hudi.exception.HoodieException; + +import org.apache.flink.runtime.jobgraph.OperatorID; +import org.apache.flink.runtime.jobgraph.tasks.TaskOperatorEventGateway; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.util.concurrent.atomic.AtomicInteger; + +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.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; + +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()); + assertNull(Correspondent.InstantTimeResponse.getInstance(null).getInstant()); + } + + @Test + void testCommitAckFactoryAndAccessors() { + CommitAckEvent event = CommitAckEvent.getInstance(11L); + assertEquals(11L, event.getCheckpointId()); + event.setCheckpointId(12L); + assertEquals(12L, event.getCheckpointId()); + } + + @Test + void testInstantRequestPollsUntilReady() { + AtomicInteger requestCount = new AtomicInteger(); + Correspondent correspondent = new Correspondent() { + @Override + protected InstantTimeResponse fetchInstantTimeResponse(long checkpointId) { + return InstantTimeResponse.getInstance(requestCount.getAndIncrement() == 0 ? null : "001"); + } + }; + + assertEquals("001", correspondent.requestInstantTime(9L, 10_000L)); + assertEquals(2, requestCount.get()); + } + + @Test + void testInstantRequestTimeoutDoesNotRetry() { + AtomicInteger requestCount = new AtomicInteger(); + Correspondent correspondent = new Correspondent() { + @Override + protected InstantTimeResponse fetchInstantTimeResponse(long checkpointId) { + requestCount.incrementAndGet(); + return InstantTimeResponse.getInstance(null); + } + }; + + HoodieException error = assertThrows(HoodieException.class, () -> correspondent.requestInstantTime(9L, 0L)); + assertEquals("Timeout waiting for the instant time from the coordinator for checkpoint 9", error.getMessage()); + assertEquals(1, requestCount.get()); + } + + @Test + void testInstantPollingPreservesInterrupt() { + Correspondent correspondent = new Correspondent() { + @Override + protected InstantTimeResponse fetchInstantTimeResponse(long checkpointId) { + return InstantTimeResponse.getInstance(null); + } + }; + + Thread.currentThread().interrupt(); + try { + HoodieException error = assertThrows(HoodieException.class, () -> correspondent.requestInstantTime(9L, 10_000L)); + assertInstanceOf(InterruptedException.class, error.getCause()); + assertTrue(Thread.currentThread().isInterrupted()); + } finally { + Thread.interrupted(); + } + } + + @Test + void testInstantRequestFailureIsNotRetried() { + AtomicInteger requestCount = new AtomicInteger(); + Correspondent correspondent = new Correspondent() { + @Override + protected InstantTimeResponse fetchInstantTimeResponse(long checkpointId) throws Exception { + requestCount.incrementAndGet(); + throw new IOException("request failed"); + } + }; + + HoodieException error = assertThrows( + HoodieException.class, () -> correspondent.requestInstantTime(9L, 10_000L)); + assertEquals("request failed", error.getCause().getMessage()); + assertEquals(1, requestCount.get()); + } +} diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/BucketStreamWriteFunctionWrapper.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/BucketStreamWriteFunctionWrapper.java index b899a90787d7..34889c114b96 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/BucketStreamWriteFunctionWrapper.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/BucketStreamWriteFunctionWrapper.java @@ -121,6 +121,7 @@ public class BucketStreamWriteFunctionWrapper<I> implements TestFunctionWrapper< public void openFunction() throws Exception { this.coordinator.start(); this.coordinator.setExecutor(new MockCoordinatorExecutor(coordinatorContext)); + this.coordinator.setInstantRequestExecutor(new MockCoordinatorExecutor(coordinatorContext)); toHoodieFunction = new RowDataToHoodieFunction<>(rowType, conf); toHoodieFunction.setRuntimeContext(runtimeContext); toHoodieFunction.open(conf); diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/BulkInsertFunctionWrapper.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/BulkInsertFunctionWrapper.java index 73fd955d3579..5b5d817d8f8f 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/BulkInsertFunctionWrapper.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/BulkInsertFunctionWrapper.java @@ -103,6 +103,7 @@ public class BulkInsertFunctionWrapper<I> implements TestFunctionWrapper<I> { public void openFunction() throws Exception { this.coordinator.start(); this.coordinator.setExecutor(new MockCoordinatorExecutor(coordinatorContext)); + this.coordinator.setInstantRequestExecutor(new MockCoordinatorExecutor(coordinatorContext)); setupWriteFunction(); setupMapFunction(); if (needSortInput) { @@ -172,6 +173,7 @@ public class BulkInsertFunctionWrapper<I> implements TestFunctionWrapper<I> { this.coordinator = new StreamWriteOperatorCoordinator(conf, this.coordinatorContext); this.coordinator.start(); this.coordinator.setExecutor(new MockCoordinatorExecutor(coordinatorContext)); + this.coordinator.setInstantRequestExecutor(new MockCoordinatorExecutor(coordinatorContext)); } public void checkpointFails(long checkpointId) { diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/InsertFunctionWrapper.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/InsertFunctionWrapper.java index ea63d58d5508..a37df1e1235b 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/InsertFunctionWrapper.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/InsertFunctionWrapper.java @@ -103,6 +103,7 @@ public class InsertFunctionWrapper<I> implements TestFunctionWrapper<I> { public void openFunction() throws Exception { this.coordinator.start(); this.coordinator.setExecutor(new MockCoordinatorExecutor(coordinatorContext)); + this.coordinator.setInstantRequestExecutor(new MockCoordinatorExecutor(coordinatorContext)); setupWriteFunction(); @@ -169,6 +170,7 @@ public class InsertFunctionWrapper<I> implements TestFunctionWrapper<I> { this.coordinator = new StreamWriteOperatorCoordinator(conf, this.coordinatorContext); this.coordinator.start(); this.coordinator.setExecutor(new MockCoordinatorExecutor(coordinatorContext)); + this.coordinator.setInstantRequestExecutor(new MockCoordinatorExecutor(coordinatorContext)); } public void checkpointFails(long checkpointId) { diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/MockCorrespondent.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/MockCorrespondent.java index 6669ed9f5ace..ac83c020794b 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/MockCorrespondent.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/MockCorrespondent.java @@ -18,7 +18,6 @@ package org.apache.hudi.sink.utils; -import org.apache.hudi.exception.HoodieException; import org.apache.hudi.sink.StreamWriteOperatorCoordinator; import org.apache.hudi.sink.event.Correspondent; @@ -32,12 +31,7 @@ public class MockCorrespondent extends Correspondent { } @Override - public String requestInstantTime(long checkpointId) { - try { - InstantTimeResponse response = CoordinationResponseSerDe.unwrap(this.coordinator.handleCoordinationRequest(InstantTimeRequest.getInstance(checkpointId)).get()); - return response.getInstant(); - } catch (Exception e) { - throw new HoodieException("Error requesting the instant time from the coordinator", e); - } + protected InstantTimeResponse fetchInstantTimeResponse(long checkpointId) throws Exception { + return CoordinationResponseSerDe.unwrap(this.coordinator.handleCoordinationRequest(InstantTimeRequest.getInstance(checkpointId)).get()); } } diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/MockCorrespondentWithTimeout.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/MockCorrespondentWithTimeout.java index 22b01a6432e1..4dcada9d0a17 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/MockCorrespondentWithTimeout.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/MockCorrespondentWithTimeout.java @@ -19,7 +19,6 @@ package org.apache.hudi.sink.utils; import org.apache.hudi.configuration.FlinkOptions; -import org.apache.hudi.exception.HoodieException; import org.apache.hudi.sink.StreamWriteOperatorCoordinator; import org.apache.hudi.sink.event.Correspondent; @@ -44,13 +43,8 @@ public class MockCorrespondentWithTimeout extends Correspondent { } @Override - public String requestInstantTime(long checkpointId) { - try { - CompletableFuture<CoordinationResponse> future = this.coordinator.handleCoordinationRequest(InstantTimeRequest.getInstance(checkpointId)); - InstantTimeResponse response = CoordinationResponseSerDe.unwrap(future.get(commitAckTimeout, TimeUnit.MILLISECONDS)); - return response.getInstant(); - } catch (Exception e) { - throw new HoodieException("Error requesting the instant time from the coordinator", e); - } + protected InstantTimeResponse fetchInstantTimeResponse(long checkpointId) throws Exception { + CompletableFuture<CoordinationResponse> future = this.coordinator.handleCoordinationRequest(InstantTimeRequest.getInstance(checkpointId)); + return CoordinationResponseSerDe.unwrap(future.get(commitAckTimeout, TimeUnit.MILLISECONDS)); } } diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/StreamWriteFunctionWrapper.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/StreamWriteFunctionWrapper.java index 3bd3b2fd3412..b14429ebae18 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/StreamWriteFunctionWrapper.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/StreamWriteFunctionWrapper.java @@ -143,6 +143,7 @@ public class StreamWriteFunctionWrapper<I> implements TestFunctionWrapper<I> { resetCoordinatorToCheckpoint(); this.coordinator.start(); this.coordinator.setExecutor(new MockCoordinatorExecutor(coordinatorContext)); + this.coordinator.setInstantRequestExecutor(new MockCoordinatorExecutor(coordinatorContext)); toHoodieFunction = new RowDataToHoodieFunction<>(rowType, conf); toHoodieFunction.setRuntimeContext(runtimeContext); toHoodieFunction.open(conf); @@ -275,6 +276,7 @@ public class StreamWriteFunctionWrapper<I> implements TestFunctionWrapper<I> { resetCoordinatorToCheckpoint(); this.coordinator.start(); this.coordinator.setExecutor(new MockCoordinatorExecutor(coordinatorContext)); + this.coordinator.setInstantRequestExecutor(new MockCoordinatorExecutor(coordinatorContext)); } public void checkpointFails(long checkpointId) { diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestNonThrownExecutor.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestNonThrownExecutor.java new file mode 100644 index 000000000000..8d268cfe85be --- /dev/null +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestNonThrownExecutor.java @@ -0,0 +1,84 @@ +/* + * 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.junit.jupiter.api.Test; +import org.slf4j.Logger; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.RejectedExecutionException; +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.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; + +class TestNonThrownExecutor { + + @Test + void testTracksTasksThroughCompletionFailureAndRejection() throws Exception { + CountDownLatch gate = new CountDownLatch(1); + AtomicInteger completed = new AtomicInteger(); + AtomicInteger failures = new AtomicInteger(); + AtomicReference<Throwable> failure = new AtomicReference<>(); + NonThrownExecutor executor = NonThrownExecutor.builder(mock(Logger.class)) + .exceptionHook((message, error) -> { + failures.incrementAndGet(); + failure.set(error); + }) + .waitForTasksFinish(true) + .build(); + try { + assertFalse(executor.hasRunningTasks()); + executor.execute(() -> { + gate.await(); + throw new IllegalStateException("test failure"); + }, "blocked task"); + executor.execute(completed::incrementAndGet, "queued task"); + assertTrue(executor.hasRunningTasks()); + } finally { + gate.countDown(); + executor.close(); + } + assertEquals(1, failures.get()); + assertEquals(1, completed.get()); + assertFalse(executor.hasRunningTasks()); + executor.execute(completed::incrementAndGet, "rejected task"); + assertEquals(2, failures.get()); + assertInstanceOf(RejectedExecutionException.class, failure.get()); + assertFalse(executor.hasRunningTasks()); + executor.executeSync(completed::incrementAndGet, "rejected synchronous task"); + assertEquals(3, failures.get()); + assertInstanceOf(RejectedExecutionException.class, failure.get()); + assertFalse(executor.hasRunningTasks()); + + AtomicInteger customFailures = new AtomicInteger(); + executor.execute(completed::incrementAndGet, (message, error) -> { + assertInstanceOf(RejectedExecutionException.class, error); + customFailures.incrementAndGet(); + }, "rejected task with custom hook"); + assertEquals(1, customFailures.get()); + assertEquals(3, failures.get()); + assertEquals(1, completed.get()); + assertFalse(executor.hasRunningTasks()); + } +}
