From 483ee843e62511f6d5bc487d90aba1a636ebac22 Mon Sep 17 00:00:00 2001
From: Zhaojing Yu <yuzhaojing@bytedance.com>
Date: Fri, 25 Mar 2022 10:18:28 +0800
Subject: [PATCH] [HUDI-3703] Reset taskID in restoreWriteMetadata (#5122)

---
 .../apache/hudi/sink/common/AbstractStreamWriteFunction.java    | 2 ++
 1 file changed, 2 insertions(+)

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 f9cf938e4..4e8712b66 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
@@ -182,6 +182,8 @@ public abstract class AbstractStreamWriteFunction<I>
     boolean eventSent = false;
     for (WriteMetadataEvent event : this.writeMetadataState.get()) {
       if (Objects.equals(lastInflight, event.getInstantTime())) {
+        // Reset taskID for event
+        event.setTaskID(taskID);
         // The checkpoint succeed but the meta does not commit,
         // re-commit the inflight instant
         this.eventGateway.sendEventToCoordinator(event);