You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "zhangyue19921010 (via GitHub)" <gi...@apache.org> on 2025/07/08 06:13:36 UTC
[PR] [HUDI-9570] Flink AsyncInstant may lost data when trigger recommit [hudi]
zhangyue19921010 opened a new pull request, #13530:
URL: https://github.com/apache/hudi/pull/13530
### Change Logs
1. TM existed after snapshot state but before notify checkpoint complete.
2. Job restarted automatically with `attemptId++`
3. Not Trigger re-commit
4. Data lost because Kafka source has commit offset during snapshot state.
### Impact
flink write
### Risk level (write none, low medium or high below)
low
### Documentation Update
_Describe any necessary documentation update if there is any new feature, config, or user-facing change. If not, put "none"._
- _The config description must be updated if new configs are added or the default value of the configs are changed_
- _Any new feature or user-facing change requires updating the Hudi website. Please create a Jira ticket, attach the
ticket number here and follow the [instruction](https://hudi.apache.org/contribute/developer-setup#website) to make
changes to the website._
### Contributor's checklist
- [ ] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute)
- [ ] Change Logs and Impact were stated clearly
- [ ] Adequate tests were added if applicable
- [ ] CI passed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-9570] Flink AsyncInstant may lost data when trigger recommit [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #13530:
URL: https://github.com/apache/hudi/pull/13530#discussion_r2194782527
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/common/AbstractStreamWriteFunction.java:
##########
@@ -207,31 +207,33 @@ private void initCheckpointId(int attemptId, long restoredCheckpointId) throws E
this.checkpointId = restoredCheckpointId;
}
- private void sendBootstrapEvent(int attemptId, boolean isRestored) throws Exception {
- if (attemptId <= 0) {
- if (isRestored) {
- HoodieTimeline pendingTimeline = this.metaClient.getActiveTimeline().filterPendingExcludingCompaction();
- // if the task is initially started, resend the pending event.
- for (WriteMetadataEvent event : this.writeMetadataState.get()) {
- // Must filter out the completed instants in case it is a partial failover,
- // the write status should not be accumulated in such case.
- if (pendingTimeline.containsInstant(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);
- LOG.info("Send uncommitted write metadata event to coordinator, task[{}].", taskID);
- }
- }
- }
- } else {
- // otherwise sends an empty bootstrap event instead.
+ private void sendBootstrapEvent(boolean isRestored) throws Exception {
+ if (!isRestored || !sendPendingCommitEvents()) {
this.eventGateway.sendEventToCoordinator(WriteMetadataEvent.emptyBootstrap(taskID, checkpointId));
LOG.info("Send bootstrap write metadata event to coordinator, task[{}].", taskID);
}
}
+ private boolean sendPendingCommitEvents() throws Exception {
+ boolean eventSent = false;
Review Comment:
> Also It seems there’s no harm in resending it once :)
It could incur duplicate metadata event then duplicates for incremental read.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-9570] Flink AsyncInstant may lost data when trigger recommit [hudi]
Posted by "cshuo (via GitHub)" <gi...@apache.org>.
cshuo commented on code in PR #13530:
URL: https://github.com/apache/hudi/pull/13530#discussion_r2196829895
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/event/Correspondent.java:
##########
@@ -108,4 +123,36 @@ public String getInstant() {
return instant;
}
}
+
+ /**
+ * A request for pending checkpoint ids.
+ */
+ public static class PendingCheckpointsRequest implements CoordinationRequest {
+
+ private PendingCheckpointsRequest() {
+ }
+
+ public static PendingCheckpointsRequest getInstance() {
+ return new PendingCheckpointsRequest();
+ }
+ }
+
+ /**
+ * A response with pending checkpoint ids.
+ */
+ public static class PendingCheckpointsResponse implements CoordinationResponse {
+ private final Set<Long> ckpIds;
Review Comment:
`ckpIds` should be guaranteed as serializable, note that `Set` doesn't extends `Serializable`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-9570] Flink AsyncInstant may lost data when trigger recommit [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13530:
URL: https://github.com/apache/hudi/pull/13530#issuecomment-3047513232
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "863b04d77449572c71c5704bd37b7dd6946d1c41",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=6646",
"triggerID" : "863b04d77449572c71c5704bd37b7dd6946d1c41",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 863b04d77449572c71c5704bd37b7dd6946d1c41 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=6646)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-9570] Flink AsyncInstant may lost data when trigger recommit [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on code in PR #13530:
URL: https://github.com/apache/hudi/pull/13530#discussion_r2194484996
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/common/AbstractStreamWriteFunction.java:
##########
@@ -207,31 +207,33 @@ private void initCheckpointId(int attemptId, long restoredCheckpointId) throws E
this.checkpointId = restoredCheckpointId;
}
- private void sendBootstrapEvent(int attemptId, boolean isRestored) throws Exception {
- if (attemptId <= 0) {
- if (isRestored) {
- HoodieTimeline pendingTimeline = this.metaClient.getActiveTimeline().filterPendingExcludingCompaction();
- // if the task is initially started, resend the pending event.
- for (WriteMetadataEvent event : this.writeMetadataState.get()) {
- // Must filter out the completed instants in case it is a partial failover,
- // the write status should not be accumulated in such case.
- if (pendingTimeline.containsInstant(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);
- LOG.info("Send uncommitted write metadata event to coordinator, task[{}].", taskID);
- }
- }
- }
- } else {
- // otherwise sends an empty bootstrap event instead.
+ private void sendBootstrapEvent(boolean isRestored) throws Exception {
+ if (!isRestored || !sendPendingCommitEvents()) {
this.eventGateway.sendEventToCoordinator(WriteMetadataEvent.emptyBootstrap(taskID, checkpointId));
LOG.info("Send bootstrap write metadata event to coordinator, task[{}].", taskID);
}
}
+ private boolean sendPendingCommitEvents() throws Exception {
+ boolean eventSent = false;
Review Comment:
Hi Danny,
Below are the specific scenarios of data loss encountered during the long-running test:
Background: HDFS experienced instability for one hour, leading to frequent write timeouts.
Step 1: For Commit 1, after the Flink job completed the snapshot, the JM (JobManager) encountered a timeout exception while committing data via notifyCheckpointComplete.
Step 2: The task automatically restarted and relaunched. It began writing data based on Commit 2. Since the attempt count was > 0, no recommit operation was performed.
Step 3: For Commit 2, after the Flink job completed the snapshot, the JM hung while committing data via notifyCheckpointComplete (due to HDFS instability).
Step 4: Manual intervention occurred at this point, restarting the job from the checkpoint.
In this case, Commit 2 was successfully recommitted, but data from Commit 1 was lost.
It seems there’s no harm in resending it once.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-9570] Flink AsyncInstant may lost data when trigger recommit [hudi]
Posted by "cshuo (via GitHub)" <gi...@apache.org>.
cshuo commented on PR #13530:
URL: https://github.com/apache/hudi/pull/13530#issuecomment-3072241488
> > One concern is whether we should avoid introducing new RPC calls as much as possible, since @danny0405 also mentioned there actually exists chance the RPC event sending may fail.
> > Can we perform committing of the residual write metadata events in `handleBootstrapEvent` after cleaning legacy events for the restarting case: attemptId > 0. I think it can also solve the problem happened in step 2.
> > > Step 2: The task automatically restarted and relaunched. It began writing data based on Commit 2. Since the attempt count was > 0, no recommit operation was performed.
>
> Agreed. However, in this scenario, if a transmission failure occurs, the pending instant still won't be successfully re-committed, and thus will remain retained in the state. Therefore, this shouldn't pose significant issues.
Yes, the fix in this pr also works well even RPC failure occurs, but it makes the writing/committing more complicated.
After discussing the @danny0405 offline, we came up with another [fix](https://github.com/apache/hudi/pull/13543) on the writer coordinator side, relying on checkpoint of coordinator to persist the uncommitted write meta events and recommit them during job restoring. It would be great if you can help verify it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-9570] Flink AsyncInstant may lost data when trigger recommit [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #13530:
URL: https://github.com/apache/hudi/pull/13530#discussion_r2191685711
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/common/AbstractStreamWriteFunction.java:
##########
@@ -207,31 +207,33 @@ private void initCheckpointId(int attemptId, long restoredCheckpointId) throws E
this.checkpointId = restoredCheckpointId;
}
- private void sendBootstrapEvent(int attemptId, boolean isRestored) throws Exception {
- if (attemptId <= 0) {
- if (isRestored) {
- HoodieTimeline pendingTimeline = this.metaClient.getActiveTimeline().filterPendingExcludingCompaction();
- // if the task is initially started, resend the pending event.
- for (WriteMetadataEvent event : this.writeMetadataState.get()) {
- // Must filter out the completed instants in case it is a partial failover,
- // the write status should not be accumulated in such case.
- if (pendingTimeline.containsInstant(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);
- LOG.info("Send uncommitted write metadata event to coordinator, task[{}].", taskID);
- }
- }
- }
- } else {
- // otherwise sends an empty bootstrap event instead.
+ private void sendBootstrapEvent(boolean isRestored) throws Exception {
+ if (!isRestored || !sendPendingCommitEvents()) {
this.eventGateway.sendEventToCoordinator(WriteMetadataEvent.emptyBootstrap(taskID, checkpointId));
LOG.info("Send bootstrap write metadata event to coordinator, task[{}].", taskID);
}
}
+ private boolean sendPendingCommitEvents() throws Exception {
+ boolean eventSent = false;
Review Comment:
This is by design, there is no need to resend the metadata events when `attemptId > 0` because the event is already in the coordinator, we only need to send an empty event to clean the legacy events.
But there is still very little chance, the send of the first attempt fails and this could incur data loss, the odds should be very low: 1. first it needs a restart of the whole job, 2. then it needs a failure of the event send. If 2 happens, it usually indicates that there is network issue, that is why I want to keep the logic simple at first. My estimation for this incidence is 0.2(for manual restart of the job) * 0.001(for the network issue) = 0.00002.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-9570] Flink AsyncInstant may lost data when trigger recommit [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13530:
URL: https://github.com/apache/hudi/pull/13530#issuecomment-3055002176
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "863b04d77449572c71c5704bd37b7dd6946d1c41",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=6646",
"triggerID" : "863b04d77449572c71c5704bd37b7dd6946d1c41",
"triggerType" : "PUSH"
}, {
"hash" : "8f00e4dcb9869ad49aa063645b7919f4a8df60c9",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "8f00e4dcb9869ad49aa063645b7919f4a8df60c9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 863b04d77449572c71c5704bd37b7dd6946d1c41 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=6646)
* 8f00e4dcb9869ad49aa063645b7919f4a8df60c9 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-9570] Flink AsyncInstant may lost data when trigger recommit [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13530:
URL: https://github.com/apache/hudi/pull/13530#issuecomment-3055005839
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "863b04d77449572c71c5704bd37b7dd6946d1c41",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=6646",
"triggerID" : "863b04d77449572c71c5704bd37b7dd6946d1c41",
"triggerType" : "PUSH"
}, {
"hash" : "8f00e4dcb9869ad49aa063645b7919f4a8df60c9",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=6664",
"triggerID" : "8f00e4dcb9869ad49aa063645b7919f4a8df60c9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 863b04d77449572c71c5704bd37b7dd6946d1c41 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=6646)
* 8f00e4dcb9869ad49aa063645b7919f4a8df60c9 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=6664)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-9570] Flink AsyncInstant may lost data when trigger recommit [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 closed pull request #13530: [HUDI-9570] Flink AsyncInstant may lost data when trigger recommit
URL: https://github.com/apache/hudi/pull/13530
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-9570] Flink AsyncInstant may lost data when trigger recommit [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on PR #13530:
URL: https://github.com/apache/hudi/pull/13530#issuecomment-3059942684
> One concern is whether we should avoid introducing new RPC calls as much as possible, since @danny0405 also mentioned there actually exists chance the RPC event sending may fail.
>
> Can we perform committing of the residual write metadata events in `handleBootstrapEvent` after cleaning legacy events for the restarting case: attemptId > 0. I think it can also solve the problem happened in step 2.
>
> > Step 2: The task automatically restarted and relaunched. It began writing data based on Commit 2. Since the attempt count was > 0, no recommit operation was performed.
Agreed. However, in this scenario, if a transmission failure occurs, the pending instant still won't be successfully re-committed, and thus will remain retained in the state. Therefore, this shouldn't pose significant issues.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-9570] Flink AsyncInstant may lost data when trigger recommit [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13530:
URL: https://github.com/apache/hudi/pull/13530#issuecomment-3055171605
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "863b04d77449572c71c5704bd37b7dd6946d1c41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=6646",
"triggerID" : "863b04d77449572c71c5704bd37b7dd6946d1c41",
"triggerType" : "PUSH"
}, {
"hash" : "8f00e4dcb9869ad49aa063645b7919f4a8df60c9",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=6664",
"triggerID" : "8f00e4dcb9869ad49aa063645b7919f4a8df60c9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8f00e4dcb9869ad49aa063645b7919f4a8df60c9 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=6664)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-9570] Flink AsyncInstant may lost data when trigger recommit [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13530:
URL: https://github.com/apache/hudi/pull/13530#issuecomment-3047642618
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "863b04d77449572c71c5704bd37b7dd6946d1c41",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=6646",
"triggerID" : "863b04d77449572c71c5704bd37b7dd6946d1c41",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 863b04d77449572c71c5704bd37b7dd6946d1c41 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=6646)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-9570] Flink AsyncInstant may lost data when trigger recommit [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on PR #13530:
URL: https://github.com/apache/hudi/pull/13530#issuecomment-3082091273
close pr since https://github.com/apache/hudi/pull/13543
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-9570] Flink AsyncInstant may lost data when trigger recommit [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on PR #13530:
URL: https://github.com/apache/hudi/pull/13530#issuecomment-3055820631
> @zhangyue19921010 I have fired a fix for your scenarios, can you help to validate?
Hi @danny0405 Test passed. Nice fix~
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-9570] Flink AsyncInstant may lost data when trigger recommit [hudi]
Posted by "cshuo (via GitHub)" <gi...@apache.org>.
cshuo commented on PR #13530:
URL: https://github.com/apache/hudi/pull/13530#issuecomment-3056094832
One concern is whether we should avoid introducing new RPC calls as much as possible, since @danny0405 also mentioned there actually exists chance the RPC event sending may fail.
Can we perform committing of the residual write metadata events in `handleBootstrapEvent` after cleaning legacy events for the restarting case: attemptId > 0. I think it can also solve the problem happened in step 2.
> Step 2: The task automatically restarted and relaunched. It began writing data based on Commit 2. Since the attempt count was > 0, no recommit operation was performed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-9570] Flink AsyncInstant may lost data when trigger recommit [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13530:
URL: https://github.com/apache/hudi/pull/13530#issuecomment-3047509772
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "863b04d77449572c71c5704bd37b7dd6946d1c41",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "863b04d77449572c71c5704bd37b7dd6946d1c41",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 863b04d77449572c71c5704bd37b7dd6946d1c41 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-9570] Flink AsyncInstant may lost data when trigger recommit [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on code in PR #13530:
URL: https://github.com/apache/hudi/pull/13530#discussion_r2194484996
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/common/AbstractStreamWriteFunction.java:
##########
@@ -207,31 +207,33 @@ private void initCheckpointId(int attemptId, long restoredCheckpointId) throws E
this.checkpointId = restoredCheckpointId;
}
- private void sendBootstrapEvent(int attemptId, boolean isRestored) throws Exception {
- if (attemptId <= 0) {
- if (isRestored) {
- HoodieTimeline pendingTimeline = this.metaClient.getActiveTimeline().filterPendingExcludingCompaction();
- // if the task is initially started, resend the pending event.
- for (WriteMetadataEvent event : this.writeMetadataState.get()) {
- // Must filter out the completed instants in case it is a partial failover,
- // the write status should not be accumulated in such case.
- if (pendingTimeline.containsInstant(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);
- LOG.info("Send uncommitted write metadata event to coordinator, task[{}].", taskID);
- }
- }
- }
- } else {
- // otherwise sends an empty bootstrap event instead.
+ private void sendBootstrapEvent(boolean isRestored) throws Exception {
+ if (!isRestored || !sendPendingCommitEvents()) {
this.eventGateway.sendEventToCoordinator(WriteMetadataEvent.emptyBootstrap(taskID, checkpointId));
LOG.info("Send bootstrap write metadata event to coordinator, task[{}].", taskID);
}
}
+ private boolean sendPendingCommitEvents() throws Exception {
+ boolean eventSent = false;
Review Comment:
Hi Danny,
Below are the specific scenarios of data loss encountered during the long-running test:
Background: HDFS experienced instability for one hour, leading to frequent write timeouts.
Step 1: For Commit 1, after the Flink job completed the snapshot, the JM (JobManager) encountered a timeout exception while committing data via notifyCheckpointComplete.
Step 2: The task automatically restarted and relaunched. It began writing data based on Commit 2. Since the attempt count was > 0, no recommit operation was performed.
Step 3: For Commit 2, after the Flink job completed the snapshot, the JM hung while committing data via notifyCheckpointComplete (due to HDFS instability).
Step 4: Manual intervention occurred at this point, restarting the job from the checkpoint.
In this case, Commit 2 was successfully recommitted, but data from Commit 1 was lost.
Also It seems there’s no harm in resending it once :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-9570] Flink AsyncInstant may lost data when trigger recommit [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #13530:
URL: https://github.com/apache/hudi/pull/13530#discussion_r2194782527
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/common/AbstractStreamWriteFunction.java:
##########
@@ -207,31 +207,33 @@ private void initCheckpointId(int attemptId, long restoredCheckpointId) throws E
this.checkpointId = restoredCheckpointId;
}
- private void sendBootstrapEvent(int attemptId, boolean isRestored) throws Exception {
- if (attemptId <= 0) {
- if (isRestored) {
- HoodieTimeline pendingTimeline = this.metaClient.getActiveTimeline().filterPendingExcludingCompaction();
- // if the task is initially started, resend the pending event.
- for (WriteMetadataEvent event : this.writeMetadataState.get()) {
- // Must filter out the completed instants in case it is a partial failover,
- // the write status should not be accumulated in such case.
- if (pendingTimeline.containsInstant(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);
- LOG.info("Send uncommitted write metadata event to coordinator, task[{}].", taskID);
- }
- }
- }
- } else {
- // otherwise sends an empty bootstrap event instead.
+ private void sendBootstrapEvent(boolean isRestored) throws Exception {
+ if (!isRestored || !sendPendingCommitEvents()) {
this.eventGateway.sendEventToCoordinator(WriteMetadataEvent.emptyBootstrap(taskID, checkpointId));
LOG.info("Send bootstrap write metadata event to coordinator, task[{}].", taskID);
}
}
+ private boolean sendPendingCommitEvents() throws Exception {
+ boolean eventSent = false;
Review Comment:
> Also It seems there’s no harm in resending it once :)
It could incur duplicate metadata event then duplicates for incremental read.
And the fix in the patch does not really solve the issue you raised here, we need to add instead of override the metadata events state if the exiting instant does not commit yet.
My idea is in the request instant time PRC, we return all the pending instants altogether with the new instant time, then check the state to only remove those that is already committed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-9570] Flink AsyncInstant may lost data when trigger recommit [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on PR #13530:
URL: https://github.com/apache/hudi/pull/13530#issuecomment-3076435702
> > > One concern is whether we should avoid introducing new RPC calls as much as possible, since @danny0405 also mentioned there actually exists chance the RPC event sending may fail.
> > > Can we perform committing of the residual write metadata events in `handleBootstrapEvent` after cleaning legacy events for the restarting case: attemptId > 0. I think it can also solve the problem happened in step 2.
> > > > Step 2: The task automatically restarted and relaunched. It began writing data based on Commit 2. Since the attempt count was > 0, no recommit operation was performed.
> >
> >
> > Agreed. However, in this scenario, if a transmission failure occurs, the pending instant still won't be successfully re-committed, and thus will remain retained in the state. Therefore, this shouldn't pose significant issues.
>
> Yes, the fix in this pr also works well even RPC failure occurs, but it makes the writing/committing more complicated.
>
> After discussing the @danny0405 offline, we came up with another [fix](https://github.com/apache/hudi/pull/13543) on the writer coordinator side, relying on checkpoint of coordinator to persist the uncommitted write meta events and recommit them during job restoring. It would be great if you can help verify it.
will test it asap ~
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-9570] Flink AsyncInstant may lost data when trigger recommit [hudi]
Posted by "zhangyue19921010 (via GitHub)" <gi...@apache.org>.
zhangyue19921010 commented on PR #13530:
URL: https://github.com/apache/hudi/pull/13530#issuecomment-3055282578
> @zhangyue19921010 I have fired a fix for your scenarios, can you help to validate?
Sure thing, Will let u know once get it done(asap) ~
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-9570] Flink AsyncInstant may lost data when trigger recommit [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13530:
URL: https://github.com/apache/hudi/pull/13530#issuecomment-3057177400
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "863b04d77449572c71c5704bd37b7dd6946d1c41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=6646",
"triggerID" : "863b04d77449572c71c5704bd37b7dd6946d1c41",
"triggerType" : "PUSH"
}, {
"hash" : "8f00e4dcb9869ad49aa063645b7919f4a8df60c9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=6664",
"triggerID" : "8f00e4dcb9869ad49aa063645b7919f4a8df60c9",
"triggerType" : "PUSH"
}, {
"hash" : "c9a4e56af53825dea2b2de76c9b5172ee0aaab47",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=6670",
"triggerID" : "c9a4e56af53825dea2b2de76c9b5172ee0aaab47",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c9a4e56af53825dea2b2de76c9b5172ee0aaab47 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=6670)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-9570] Flink AsyncInstant may lost data when trigger recommit [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13530:
URL: https://github.com/apache/hudi/pull/13530#issuecomment-3056920177
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "863b04d77449572c71c5704bd37b7dd6946d1c41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=6646",
"triggerID" : "863b04d77449572c71c5704bd37b7dd6946d1c41",
"triggerType" : "PUSH"
}, {
"hash" : "8f00e4dcb9869ad49aa063645b7919f4a8df60c9",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=6664",
"triggerID" : "8f00e4dcb9869ad49aa063645b7919f4a8df60c9",
"triggerType" : "PUSH"
}, {
"hash" : "c9a4e56af53825dea2b2de76c9b5172ee0aaab47",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "c9a4e56af53825dea2b2de76c9b5172ee0aaab47",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8f00e4dcb9869ad49aa063645b7919f4a8df60c9 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=6664)
* c9a4e56af53825dea2b2de76c9b5172ee0aaab47 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-9570] Flink AsyncInstant may lost data when trigger recommit [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13530:
URL: https://github.com/apache/hudi/pull/13530#issuecomment-3056929120
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "863b04d77449572c71c5704bd37b7dd6946d1c41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=6646",
"triggerID" : "863b04d77449572c71c5704bd37b7dd6946d1c41",
"triggerType" : "PUSH"
}, {
"hash" : "8f00e4dcb9869ad49aa063645b7919f4a8df60c9",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=6664",
"triggerID" : "8f00e4dcb9869ad49aa063645b7919f4a8df60c9",
"triggerType" : "PUSH"
}, {
"hash" : "c9a4e56af53825dea2b2de76c9b5172ee0aaab47",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=6670",
"triggerID" : "c9a4e56af53825dea2b2de76c9b5172ee0aaab47",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8f00e4dcb9869ad49aa063645b7919f4a8df60c9 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=6664)
* c9a4e56af53825dea2b2de76c9b5172ee0aaab47 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=6670)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org