You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by si...@apache.org on 2023/12/18 07:42:17 UTC
(hudi) 04/07: [HUDI-7132] Data may be lost for flink task failure (#10312)
This is an automated email from the ASF dual-hosted git repository.
sivabalan pushed a commit to branch release-0.14.0-siva-0.14.1
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 549a80bf865012e707abd045597fb9ca5a0a12a4
Author: Danny Chan <yu...@gmail.com>
AuthorDate: Tue Dec 12 18:26:03 2023 +0800
[HUDI-7132] Data may be lost for flink task failure (#10312)
---
.../hudi/sink/StreamWriteOperatorCoordinator.java | 7 +++---
.../sink/TestStreamWriteOperatorCoordinator.java | 29 ++++++++++++++++++++++
2 files changed, 32 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 55188f2cc5e..274091c88ea 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
@@ -299,9 +299,7 @@ public class StreamWriteOperatorCoordinator
@Override
public void subtaskFailed(int i, @Nullable Throwable throwable) {
- // reset the event
- this.eventBuffer[i] = null;
- LOG.warn("Reset the event for task [" + i + "]", throwable);
+ // no operation
}
@Override
@@ -376,7 +374,8 @@ public class StreamWriteOperatorCoordinator
}
private void addEventToBuffer(WriteMetadataEvent event) {
- if (this.eventBuffer[event.getTaskID()] != null) {
+ if (this.eventBuffer[event.getTaskID()] != null
+ && this.eventBuffer[event.getTaskID()].getInstantTime().equals(event.getInstantTime())) {
this.eventBuffer[event.getTaskID()].mergeWith(event);
} else {
this.eventBuffer[event.getTaskID()] = event;
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 e0e42b9d8c4..186500b1f38 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
@@ -147,6 +147,35 @@ public class TestStreamWriteOperatorCoordinator {
"Receive an unexpected event for instant abc from task 0");
}
+ @Test
+ public void testEventReset() {
+ CompletableFuture<byte[]> future = new CompletableFuture<>();
+ coordinator.checkpointCoordinator(1, future);
+ OperatorEvent event1 = WriteMetadataEvent.builder()
+ .taskID(0)
+ .instantTime("001")
+ .writeStatus(Collections.emptyList())
+ .build();
+ coordinator.handleEventFromOperator(0, event1);
+ coordinator.subtaskFailed(0, null);
+ assertNotNull(coordinator.getEventBuffer()[0], "Events should not be cleared by subTask failure");
+
+ OperatorEvent event2 = createOperatorEvent(0, "001", "par1", false, 0.1);
+ coordinator.handleEventFromOperator(0, event2);
+ coordinator.subtaskFailed(0, null);
+ assertNotNull(coordinator.getEventBuffer()[0], "Events should not be cleared by subTask failure");
+
+ OperatorEvent event3 = createOperatorEvent(0, "001", "par1", false, 0.1);
+ coordinator.handleEventFromOperator(0, event3);
+ assertThat("Multiple events of same instant should be merged",
+ coordinator.getEventBuffer()[0].getWriteStatuses().size(), is(2));
+
+ OperatorEvent event4 = createOperatorEvent(0, "002", "par1", false, 0.1);
+ coordinator.handleEventFromOperator(0, event4);
+ assertThat("The new event should override the old event",
+ coordinator.getEventBuffer()[0].getWriteStatuses().size(), is(1));
+ }
+
@Test
public void testCheckpointCompleteWithPartialEvents() {
final CompletableFuture<byte[]> future = new CompletableFuture<>();