You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by da...@apache.org on 2023/07/20 08:19:21 UTC
[hudi] branch master updated: [HUDI-6570] Cancel to send commit ack event for flink batch job (#9238)
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 ffdcb24d3e6 [HUDI-6570] Cancel to send commit ack event for flink batch job (#9238)
ffdcb24d3e6 is described below
commit ffdcb24d3e6dc942d0af80336b9139e45f811d31
Author: hehuiyuan <47...@qq.com>
AuthorDate: Thu Jul 20 16:19:08 2023 +0800
[HUDI-6570] Cancel to send commit ack event for flink batch job (#9238)
---
.../java/org/apache/hudi/common/model/EventTimeAvroPayload.java | 6 +++---
.../java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java | 4 +++-
2 files changed, 6 insertions(+), 4 deletions(-)
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/model/EventTimeAvroPayload.java b/hudi-common/src/main/java/org/apache/hudi/common/model/EventTimeAvroPayload.java
index b750cffb6ae..a5de6034632 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/model/EventTimeAvroPayload.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/model/EventTimeAvroPayload.java
@@ -46,6 +46,9 @@ public class EventTimeAvroPayload extends DefaultHoodieRecordPayload {
@Override
public Option<IndexedRecord> combineAndGetUpdateValue(IndexedRecord currentValue, Schema schema, Properties properties) throws IOException {
+ /*
+ * Check if the incoming record is a delete record.
+ */
if (recordBytes.length == 0 || isDeletedRecord) {
return Option.empty();
}
@@ -58,9 +61,6 @@ public class EventTimeAvroPayload extends DefaultHoodieRecordPayload {
return Option.of(currentValue);
}
- /*
- * Now check if the incoming record is a delete record.
- */
return Option.of(incomingRecord);
}
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 39f5dbd75d7..dd9b7e9141e 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
@@ -532,7 +532,9 @@ public class StreamWriteOperatorCoordinator
// Send commit ack event to the write function to unblock the flushing
// If this checkpoint has no inputs while the next checkpoint has inputs,
// the 'isConfirming' flag should be switched with the ack event.
- sendCommitAckEvents(checkpointId);
+ if (checkpointId != -1) {
+ sendCommitAckEvents(checkpointId);
+ }
return false;
}
doCommit(instant, writeResults);