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 a6b8744d7fb2 fix(flink): defer instant creation until earlier
checkpoints finish (#20389)
a6b8744d7fb2 is described below
commit a6b8744d7fb26c71fd22cb8982d32fdf1b990109
Author: Vova Kolmakov <[email protected]>
AuthorDate: Tue Oct 20 12:04:53 2026 +0700
fix(flink): defer instant creation until earlier checkpoints finish (#20389)
* fix(flink): wait for earlier instants before starting a new one in
blocking mode
* fix(flink): rely on task polling for pending instant creation
* fix(flink): defer instant creation only for earlier checkpoints
---------
Co-authored-by: Vova Kolmakov <[email protected]>
Co-authored-by: danny0405 <[email protected]>
---
.../hudi/sink/StreamWriteOperatorCoordinator.java | 14 ++-
.../org/apache/hudi/sink/utils/CommitGuard.java | 105 -----------------
.../org/apache/hudi/sink/utils/EventBuffers.java | 19 +---
.../sink/TestStreamWriteOperatorCoordinator.java | 125 +++++++++++++++++++++
.../apache/hudi/sink/utils/TestCommitGuard.java | 78 -------------
.../apache/hudi/sink/utils/TestEventBuffers.java | 61 +++++-----
6 files changed, 161 insertions(+), 241 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 e8c5ded34d70..5a56001e6437 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
@@ -427,11 +427,9 @@ public class StreamWriteOperatorCoordinator
long checkpointId = request.getCheckpointId();
// Existing instants must remain available while another checkpoint's
creation is blocked.
Pair<String, EventBuffer> instantTimeAndEventBuffer =
this.eventBuffers.getInstantAndEventBuffer(checkpointId);
- if (instantTimeAndEventBuffer == null &&
!instantRequestExecutor.hasRunningTasks()) {
+ if (instantTimeAndEventBuffer == null &&
!isInstantCreationBlocked(checkpointId) &&
!instantRequestExecutor.hasRunningTasks()) {
instantRequestExecutor.execute(() -> {
- if (this.eventBuffers.getInstantAndEventBuffer(checkpointId) == null) {
- // Wait until previous instants are committed.
- eventBuffers.awaitAllInstantsToCompleteIfNecessary();
+ if (this.eventBuffers.getInstantAndEventBuffer(checkpointId) == null
&& !isInstantCreationBlocked(checkpointId)) {
this.eventBuffers.initNewEventBuffer(checkpointId, startInstant());
}
}, "request instant time");
@@ -441,6 +439,14 @@ public class StreamWriteOperatorCoordinator
return
CompletableFuture.completedFuture(CoordinationResponseSerDe.wrap(Correspondent.InstantTimeResponse.getInstance(instantTime)));
}
+ /**
+ * Defers new instant creation while an earlier checkpoint has a pending
instant, even without write metadata events.
+ * Later checkpoints belong to an abandoned attempt when a writer requests
an earlier checkpoint after recovery.
+ */
+ private boolean isInstantCreationBlocked(long checkpointId) {
+ return OptionsResolver.isBlockingInstantGeneration(conf) &&
!eventBuffers.getPendingInstantsBefore(checkpointId).isEmpty();
+ }
+
private CompletableFuture<CoordinationResponse>
handleInFlightInstantsRequest(Correspondent.InflightInstantsRequest request) {
CoordinationResponse coordinationResponse =
Correspondent.InflightInstantsResponse.getInstance(eventBuffers.getAllCheckpointIdAndInstants());
return
CompletableFuture.completedFuture(CoordinationResponseSerDe.wrap(coordinationResponse));
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/CommitGuard.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/CommitGuard.java
deleted file mode 100644
index 76836b6648b9..000000000000
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/CommitGuard.java
+++ /dev/null
@@ -1,105 +0,0 @@
-/*
- * 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 java.util.List;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.locks.Condition;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
-import java.util.function.Supplier;
-
-/**
- * The commit guard used for blocking instant time generation.
- *
- * <p>A new instant time can not be generated until all the pending instants
are committed.
- */
-public class CommitGuard {
- /**
- * A lock used to block instant generation until being signaled after
pending instant is committed.
- */
- private final Lock lock;
- private final Condition condition;
- private final long commitAckTimeout;
-
- private CommitGuard(long commitAckTimeout) {
- this.lock = new ReentrantLock();
- this.condition = lock.newCondition();
- this.commitAckTimeout = commitAckTimeout;
- }
-
- public static CommitGuard create(long commitAckTimeout) {
- return new CommitGuard(commitAckTimeout);
- }
-
- /**
- * Wait until all the pending instants are committed.
- *
- * @param instants Current pending instants
- */
- public void blockFor(String instants) {
- lock.lock();
- try {
- if (!condition.await(commitAckTimeout, TimeUnit.MILLISECONDS)) {
- throw new HoodieException("Timeout(" + commitAckTimeout + "ms) while
waiting for instants [" + instants + "] to commit");
- }
- } catch (InterruptedException e) {
- throw new HoodieException("Blocking for instants completion is
interrupted.", e);
- } finally {
- lock.unlock();
- }
- }
-
- /**
- * Wait until the pending instants are committed.
- *
- * @param pendingInstants Supplier to get the pending instants
- */
- public void blockFor(Supplier<List<String>> pendingInstants) {
- lock.lock();
- long nanos = TimeUnit.MILLISECONDS.toNanos(commitAckTimeout);
- try {
- while (!pendingInstants.get().isEmpty()) {
- if (nanos <= 0L) {
- throw new HoodieException("Timeout(" + commitAckTimeout + "ms) while
waiting for instants ["
- + pendingInstants.get() + "] to commit");
- }
- nanos = condition.awaitNanos(nanos);
- }
- } catch (InterruptedException e) {
- throw new HoodieException("Blocking for instants completion is
interrupted.", e);
- } finally {
- lock.unlock();
- }
- }
-
- /**
- * Signals all waited threads which are waiting on the condition.
- */
- public void unblock() {
- lock.lock();
- try {
- condition.signalAll();
- } finally {
- lock.unlock();
- }
- }
-}
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/EventBuffers.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/EventBuffers.java
index 8b0fc3cdbf9f..9985d73fe225 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/EventBuffers.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/EventBuffers.java
@@ -21,7 +21,6 @@ package org.apache.hudi.sink.utils;
import org.apache.hudi.client.WriteStatus;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.collection.Pair;
-import org.apache.hudi.configuration.FlinkOptions;
import org.apache.hudi.configuration.OptionsResolver;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.sink.event.WriteMetadataEvent;
@@ -48,26 +47,21 @@ public class EventBuffers implements Serializable {
// {checkpointId -> (instant, data write events, index write events)}
private final Map<Long, Pair<String, EventBuffer>> eventBuffers;
- private final Option<CommitGuard> commitGuardOption;
private final int dataWriteParallelism;
private final int indexWriteParallelism;
private EventBuffers(
Map<Long, Pair<String, EventBuffer>> eventBuffers,
- Option<CommitGuard> commitGuardOption,
int dataWriteParallelism,
int indexWriteParallelism) {
this.eventBuffers = eventBuffers;
- this.commitGuardOption = commitGuardOption;
this.dataWriteParallelism = dataWriteParallelism;
this.indexWriteParallelism = indexWriteParallelism;
}
public static EventBuffers getInstance(Configuration conf, int
dataWriteParallelism) {
- final Option<CommitGuard> commitGuardOpt =
OptionsResolver.isBlockingInstantGeneration(conf)
- ?
Option.of(CommitGuard.create(conf.get(FlinkOptions.WRITE_COMMIT_ACK_TIMEOUT)))
: Option.empty();
final int indexWriteParallelism =
OptionsResolver.indexWriteParallelism(conf);
- return new EventBuffers(new ConcurrentSkipListMap<>(), commitGuardOpt,
dataWriteParallelism, indexWriteParallelism);
+ return new EventBuffers(new ConcurrentSkipListMap<>(),
dataWriteParallelism, indexWriteParallelism);
}
public EventBuffer addEventToBuffer(WriteMetadataEvent event) {
@@ -137,15 +131,8 @@ public class EventBuffers implements Serializable {
this.eventBuffers.put(checkpointId, Pair.of(instantTime, new
EventBuffer(dataWriteParallelism, indexWriteParallelism)));
}
- public void awaitAllInstantsToCompleteIfNecessary() {
- if (this.commitGuardOption.isPresent() && nonEmpty()) {
- this.commitGuardOption.get().blockFor(getPendingInstants());
- }
- }
-
public void reset(long checkpointId) {
this.eventBuffers.remove(checkpointId);
- this.commitGuardOption.ifPresent(CommitGuard::unblock);
}
public boolean nonEmpty() {
@@ -162,10 +149,6 @@ public class EventBuffers implements Serializable {
.collect(Collectors.toList());
}
- public String getPendingInstants() {
- return
this.eventBuffers.values().stream().map(Pair::getKey).collect(Collectors.joining(","));
- }
-
/**
* Get write metadata events where there exists no event sent by eager
flushing from writers.
*/
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 1e68d9f2842e..b32b62d6dfbc 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
@@ -1139,6 +1139,7 @@ public class TestStreamWriteOperatorCoordinator {
@Test
void testReadyInstantRemainsAvailableWhileAnotherCreationIsBlocked() throws
Exception {
+
resetToMergeOnRead(TestConfigurations.getDefaultConf(tempFile.getAbsolutePath()));
String existingInstant = requestInstantTime(1L);
MockOperatorCoordinatorContext ctx = (MockOperatorCoordinatorContext)
coordinator.getContext();
CountDownLatch gate = new CountDownLatch(1);
@@ -1162,6 +1163,130 @@ public class TestStreamWriteOperatorCoordinator {
}
}
+ @Test
+ void testEarlierPendingInstantDefersCreationUntilRetry() throws Exception {
+ String instant1 = requestInstantTime(1L);
+ MockOperatorCoordinatorContext ctx = (MockOperatorCoordinatorContext)
coordinator.getContext();
+ NonThrownExecutor worker =
Mockito.spy(NonThrownExecutor.builder(Mockito.mock(Logger.class))
+ .exceptionHook((errMsg, t) -> ctx.failJob(new HoodieException(errMsg,
t)))
+ .build());
+ coordinator.setInstantRequestExecutor(worker);
+
+ // An earlier checkpoint with an empty event buffer still represents a
pending instant.
+ Correspondent.InstantTimeResponse pending =
CoordinationResponseSerDe.unwrap(
+
coordinator.handleCoordinationRequest(Correspondent.InstantTimeRequest.getInstance(2L))
+ .get(1, TimeUnit.SECONDS));
+ assertNull(pending.getInstant());
+ Mockito.verify(worker, Mockito.never()).execute(Mockito.any(),
Mockito.eq("request instant time"));
+ assertEquals(instant1, requestInstantTime(1L), "Matching checkpoint
requests must reuse the pending instant");
+ assertEquals(0,
TestUtils.getCompletedInstantCount(tempFile.getAbsolutePath(),
HoodieTimeline.ROLLBACK_ACTION));
+
+ coordinator.handleEventFromOperator(0, createOperatorEvent(0, 1L,
instant1, "par1", false, true, 0.1));
+ coordinator.handleEventFromOperator(1, createOperatorEvent(1, 1L,
instant1, "par2", false, true, 0.1));
+ coordinator.notifyCheckpointComplete(2L);
+ assertEquals(instant1,
TestUtils.getLastCompleteInstant(tempFile.getAbsolutePath()));
+ assertFalse(worker.hasRunningTasks(), "Commit completion must not create
an instant for a previously deferred request");
+ assertNull(coordinator.getEventBuffer(2L));
+
+ assertNotEquals(instant1, requestInstantTime(2L), "A new task-side poll
must create the next instant");
+ assertEquals(0,
TestUtils.getCompletedInstantCount(tempFile.getAbsolutePath(),
HoodieTimeline.ROLLBACK_ACTION));
+ assertFalse(ctx.isJobFailed());
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"regional", "global"})
+ void testRecoveredWriterReusesPendingInstantAndRetries(String failover)
throws Exception {
+ String instant1 = requestInstantTime(1L);
+ Correspondent.InstantTimeResponse pending =
CoordinationResponseSerDe.unwrap(
+
coordinator.handleCoordinationRequest(Correspondent.InstantTimeRequest.getInstance(2L)).get(1,
TimeUnit.SECONDS));
+ assertNull(pending.getInstant());
+
+ if ("regional".equals(failover)) {
+ coordinator.subtaskReset(0, 1L);
+ } else {
+ CompletableFuture<byte[]> checkpoint = new CompletableFuture<>();
+ coordinator.checkpointCoordinator(1L, checkpoint);
+ coordinator.resetToCheckpoint(1L, checkpoint.get());
+ }
+ assertEquals(instant1, requestInstantTime(1L));
+ assertNull(coordinator.getEventBuffer(2L));
+
+ coordinator.handleEventFromOperator(0, createOperatorEvent(0, 1L,
instant1, "par1", false, true, 0.1));
+ coordinator.handleEventFromOperator(1, createOperatorEvent(1, 1L,
instant1, "par2", false, true, 0.1));
+ coordinator.notifyCheckpointComplete(2L);
+ assertNotEquals(instant1, requestInstantTime(2L));
+ assertEquals(0,
TestUtils.getCompletedInstantCount(tempFile.getAbsolutePath(),
HoodieTimeline.ROLLBACK_ACTION));
+ assertFalse(((MockOperatorCoordinatorContext)
coordinator.getContext()).isJobFailed());
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"regional", "global"})
+ void testRecoveredWriterRollsBackLaterInstantAndContinues(String failover)
throws Exception {
+ CompletableFuture<byte[]> checkpoint = new CompletableFuture<>();
+ coordinator.checkpointCoordinator(1L, checkpoint);
+ String abandonedInstant = requestInstantTime(2L);
+ coordinator.handleEventFromOperator(1, createOperatorEvent(1, 2L,
abandonedInstant, "par2", false, true, 0.1));
+
+ if ("regional".equals(failover)) {
+ coordinator.subtaskReset(0, 1L);
+ coordinator.subtaskReset(1, 1L);
+ } else {
+ coordinator.resetToCheckpoint(1L, checkpoint.get());
+ }
+ // One restored writer requests its instant before the other writer clears
its old metadata.
+ coordinator.handleEventFromOperator(0,
WriteMetadataEvent.emptyBootstrap(0, 1L, false));
+
+ String recoveredInstant = requestInstantTime(1L);
+ assertNotEquals(abandonedInstant, recoveredInstant);
+ assertEquals(recoveredInstant, requestInstantTime(1L));
+ assertEquals(1,
TestUtils.getCompletedInstantCount(tempFile.getAbsolutePath(),
HoodieTimeline.ROLLBACK_ACTION));
+ assertEquals(recoveredInstant,
TestUtils.getLastPendingInstant(tempFile.getAbsolutePath()));
+
+ coordinator.handleEventFromOperator(1,
WriteMetadataEvent.emptyBootstrap(1, 1L, false));
+ coordinator.handleEventFromOperator(0, createOperatorEvent(0, 1L,
recoveredInstant, "par1", false, true, 0.1));
+ coordinator.handleEventFromOperator(1, createOperatorEvent(1, 1L,
recoveredInstant, "par2", false, true, 0.1));
+ coordinator.notifyCheckpointComplete(3L);
+ assertEquals(recoveredInstant,
TestUtils.getLastCompleteInstant(tempFile.getAbsolutePath()));
+ assertNull(coordinator.getEventBuffer(2L), "Checkpoint completion must
remove the abandoned empty buffer");
+ assertNotEquals(recoveredInstant, requestInstantTime(3L));
+ assertFalse(((MockOperatorCoordinatorContext)
coordinator.getContext()).isJobFailed());
+ }
+
+ @Test
+ void testInstantCreationRechecksPendingBuffersOnWorker() throws Exception {
+ 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());
+ // Bootstrap can restore a pending buffer after admission but before the
creation worker runs.
+ Field buffersField =
StreamWriteOperatorCoordinator.class.getDeclaredField("eventBuffers");
+ buffersField.setAccessible(true);
+ EventBuffers buffers = (EventBuffers) buffersField.get(coordinator);
+ buffers.initNewEventBuffer(1L, "001");
+ } finally {
+ gate.countDown();
+ }
+ worker.executeSync(() -> { }, "drain instant requests");
+ assertNull(coordinator.getEventBuffer(2L));
+ assertNotNull(coordinator.getEventBuffer(1L));
+ assertFalse(ctx.isJobFailed());
+ }
+
+ @Test
+ void testNonblockingInstantGenerationAllowsPendingBuffers() throws Exception
{
+
resetToMergeOnRead(TestConfigurations.getDefaultConf(tempFile.getAbsolutePath()));
+ String instant1 = requestInstantTime(1L);
+ assertNotEquals(instant1, requestInstantTime(2L));
+ assertNotNull(coordinator.getEventBuffer(1L));
+ assertNotNull(coordinator.getEventBuffer(2L));
+ assertFalse(((MockOperatorCoordinatorContext)
coordinator.getContext()).isJobFailed());
+ }
+
// -------------------------------------------------------------------------
// Utilities
// -------------------------------------------------------------------------
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestCommitGuard.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestCommitGuard.java
deleted file mode 100644
index 3fc00e76d9ff..000000000000
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestCommitGuard.java
+++ /dev/null
@@ -1,78 +0,0 @@
-/*
- * 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.junit.jupiter.api.Test;
-
-import java.util.Collections;
-import java.util.List;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-
-/**
- * Test cases for {@link CommitGuard}.
- */
-public class TestCommitGuard {
-
- @Test
- void testBlockForRechecksPendingInstantsAfterSignal() throws Exception {
- CommitGuard guard = CommitGuard.create(5_000L);
- AtomicBoolean pending = new AtomicBoolean(true);
- ExecutorService executor = Executors.newSingleThreadExecutor();
- try {
- CompletableFuture<Void> waitingFuture1 = CompletableFuture.runAsync(
- () -> guard.blockFor(() -> pending.get() ? List.of("001", "002") :
Collections.emptyList()), executor);
- CompletableFuture<Void> waitingFuture2 = CompletableFuture.runAsync(
- () -> guard.blockFor(() -> pending.get() ? List.of("001", "002") :
Collections.emptyList()), executor);
-
- Thread.sleep(100);
- guard.unblock();
- Thread.sleep(100);
- assertFalse(waitingFuture1.isDone(), "Should still block because pending
instants are not empty");
- assertFalse(waitingFuture2.isDone(), "Should still block because pending
instants are not empty");
-
- pending.set(false);
- guard.unblock();
- waitingFuture1.get(2, TimeUnit.SECONDS);
- assertTrue(waitingFuture1.isDone(), "Should finish when pending instants
become empty");
- waitingFuture2.get(2, TimeUnit.SECONDS);
- assertTrue(waitingFuture2.isDone(), "Should finish when pending instants
become empty");
- } finally {
- executor.shutdownNow();
- }
- }
-
- @Test
- void testBlockForTimeoutWhenPendingInstantsRemain() {
- CommitGuard guard = CommitGuard.create(100L);
- HoodieException hoodieException = assertThrows(
- HoodieException.class,
- () -> guard.blockFor(() ->
Collections.singletonList("timeout-instant")));
- assertTrue(hoodieException.getMessage().contains("timeout-instant"));
- }
-}
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestEventBuffers.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestEventBuffers.java
index c2c7f7d338f3..580c889bba93 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestEventBuffers.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestEventBuffers.java
@@ -18,17 +18,15 @@
package org.apache.hudi.sink.utils;
-import org.apache.hudi.configuration.FlinkOptions;
import org.apache.hudi.sink.event.WriteMetadataEvent;
import org.apache.flink.configuration.Configuration;
import org.junit.jupiter.api.Test;
+import java.util.Arrays;
import java.util.Collections;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -38,42 +36,33 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
public class TestEventBuffers {
@Test
- void testAwaitAllInstantsWaitsUntilAllPendingInstantsReset() throws
Exception {
- Configuration conf = new Configuration();
- conf.set(FlinkOptions.WRITE_COMMIT_ACK_TIMEOUT, 5_000L);
- EventBuffers eventBuffers = EventBuffers.getInstance(conf, 1);
+ void testPendingInstantsBeforeIncludeBuffersWithoutEvents() {
+ EventBuffers buffers = EventBuffers.getInstance(new Configuration(), 1);
+ assertTrue(buffers.getPendingInstantsBefore(3L).isEmpty());
- eventBuffers.initNewEventBuffer(1L, "001");
- eventBuffers.addEventToBuffer(newWriteEvent(1L, "001"));
+ buffers.initNewEventBuffer(1L, "001");
+ assertEquals(Collections.singletonList("001"),
buffers.getPendingInstantsBefore(2L));
+ assertFalse(buffers.nonEmpty());
+ assertTrue(buffers.getPendingInstantsBefore(1L).isEmpty());
- ExecutorService executor = Executors.newSingleThreadExecutor();
- try {
- // two tasks block waiting on commit.
- CompletableFuture<Void> waitingFuture1 = CompletableFuture.runAsync(
- eventBuffers::awaitAllInstantsToCompleteIfNecessary, executor);
- CompletableFuture<Void> waitingFuture2 = CompletableFuture.runAsync(
- eventBuffers::awaitAllInstantsToCompleteIfNecessary, executor);
-
- Thread.sleep(100);
- assertFalse(waitingFuture1.isDone());
- assertFalse(waitingFuture2.isDone());
-
- eventBuffers.reset(1L);
- Thread.sleep(100);
- assertTrue(waitingFuture1.isDone());
- assertTrue(waitingFuture2.isDone());
- } finally {
- executor.shutdownNow();
- }
- }
-
- private static WriteMetadataEvent newWriteEvent(long checkpointId, String
instant) {
- return WriteMetadataEvent.builder()
+ buffers.addEventToBuffer(WriteMetadataEvent.builder()
.taskID(0)
- .checkpointId(checkpointId)
- .instantTime(instant)
+ .checkpointId(1L)
+ .instantTime("001")
.writeStatus(Collections.emptyList())
.lastBatch(true)
- .build();
+ .build());
+ assertEquals(Collections.singletonList("001"),
buffers.getPendingInstantsBefore(2L));
+ assertTrue(buffers.nonEmpty());
+
+ buffers.initNewEventBuffer(2L, "002");
+ assertEquals(Arrays.asList("001", "002"),
buffers.getPendingInstantsBefore(3L));
+ buffers.reset(1L);
+ assertTrue(buffers.getPendingInstantsBefore(2L).isEmpty());
+ assertEquals(Collections.singletonList("002"),
buffers.getPendingInstantsBefore(3L));
+ assertFalse(buffers.nonEmpty());
+
+ buffers.reset(2L);
+ assertTrue(buffers.getPendingInstantsBefore(3L).isEmpty());
}
}
