(hudi) branch master updated: [HUDI-7165] Flink multi writer not close the failed instant heartbeat (#10221)

[danny0405]

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 831c0c861fa [HUDI-7165] Flink multi writer not close the failed 
instant heartbeat (#10221)
831c0c861fa is described below

commit 831c0c861fa69266e9de065f07045af1cb437cbe
Author: ksmou <135721692+ks...@users.noreply.github.com>
AuthorDate: Fri Dec 1 08:45:52 2023 +0800

    [HUDI-7165] Flink multi writer not close the failed instant heartbeat 
(#10221)
---
 .../java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java     | 4 ++++
 1 file changed, 4 insertions(+)

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 dfc713244d5..ec612c6caf2 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
@@ -420,6 +420,10 @@ public class StreamWriteOperatorCoordinator
       }
       commitInstant(instant);
     }
+    // stop the heartbeat for old instant
+    if (writeClient.getConfig().getFailedWritesCleanPolicy().isLazy() && 
!WriteMetadataEvent.BOOTSTRAP_INSTANT.equals(this.instant)) {
+      writeClient.getHeartbeatClient().stop(this.instant);
+    }
     // starts a new instant
     startInstant();
     // upgrade downgrade