danny0405 commented on code in PR #20445:
URL: https://github.com/apache/hudi/pull/20445#discussion_r4488474380
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/AsyncCompactService.java:
##########
@@ -78,7 +80,23 @@ protected Pair<CompletableFuture, ExecutorService>
startService() {
if (null != instantTime) {
log.info("Starting Compaction for instant {}", instantTime);
- compactor.compact(instantTime);
+ BaseHoodieWriteClient runClient = compactor.compactionClient;
+ try {
+ compactor.compact(instantTime);
+ } catch (Exception e) {
+ if (!isRetryableFailure(runClient, e)) {
+ throw e;
+ }
+ log.warn("Compaction for instant {} failed with a write client
that was replaced, retrying it with the new write client",
+ instantTime, e);
+ awaitWriteClientReplaced(runClient);
+ if (isPendingCompaction(instantTime)) {
+ enqueuePendingAsyncServiceInstant(instantTime);
Review Comment:
[P2] Account for the retired client's heartbeat before retrying compaction.
With OCC/multi-writer mode enabled, a running compaction has a heartbeat.
Closing the old client stops its heartbeat timers and clears its heartbeat map,
but leaves the heartbeat file. If the run fails after close completes, its
subsequent heartbeatClient.stop(instant) cannot delete that file because its
map entry is already gone. The replacement client's compact() calls
validateHeartBeat() before rolling back the inflight instant, so an immediate
retry is rejected with HoodieLockException while that heartbeat is still fresh.
The replacement client is not retired, making this second failure fatal to the
service/query.
I reproduced old.close(); old.stop(instant) leaving heartbeatExists=true and
newClient.isHeartbeatExpired=false. Could we coordinate cleanup/ownership of
the failed run's heartbeat before retrying, and add an OCC variant of the
client-swap test? The current integration test uses single-writer mode and does
not exercise this check.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/AsyncClusteringService.java:
##########
@@ -76,7 +78,23 @@ protected Pair<CompletableFuture, ExecutorService>
startService() {
final String instant = fetchNextAsyncServiceInstant();
if (null != instant) {
log.info("Starting clustering for instant {}", instant);
- clusteringClient.cluster(instant);
+ BaseHoodieWriteClient runClient =
clusteringClient.clusteringClient;
+ try {
+ clusteringClient.cluster(instant);
+ } catch (Exception e) {
+ if (!isRetryableFailure(runClient, e)) {
+ throw e;
+ }
+ log.warn("Clustering for instant {} failed with a write client
that was replaced, retrying it with the new write client",
+ instant, e);
+ awaitWriteClientReplaced(runClient);
+ if (isPendingClustering(instant)) {
+ enqueuePendingAsyncServiceInstant(instant);
+ } else {
+ log.info("Clustering for instant {} is no longer pending, not
retrying it", instant);
+ }
+ continue;
Review Comment:
Optional simplification: the failure classification, wait-for-replacement,
pending-instant check, and requeue logic are nearly identical to
AsyncCompactService. Consider extracting a small shared helper into
HoodieAsyncTableService, leaving execution and pending-instant detection as
service-specific hooks. That would keep fixes to client capture and
shutdown/retry behavior consistent between compaction and clustering.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/AsyncCompactService.java:
##########
@@ -78,7 +80,23 @@ protected Pair<CompletableFuture, ExecutorService>
startService() {
if (null != instantTime) {
log.info("Starting Compaction for instant {}", instantTime);
- compactor.compact(instantTime);
+ BaseHoodieWriteClient runClient = compactor.compactionClient;
+ try {
+ compactor.compact(instantTime);
Review Comment:
[P2] Use the same client snapshot for execution and failure classification.
There are two independent reads here: this capture of
compactor.compactionClient, followed by another read inside
HoodieSparkCompactor.compact(). A handoff between them lets runClient be
retired client A while compact() actually runs on current client B. An
unrelated failure from B is then classified against A and retried on B, even
though B was never retired. AsyncClusteringService/HoodieSparkClusteringClient
has the same race.
A latch-controlled reproduction produced exactly this sequence: the
operation used B, the failure hook received A, and the service retried. Could
we capture once and pass that exact client into compact()/cluster() as well as
the failure hook? Holding the service lock across the operation would
reintroduce the ingestion blocking this change is intended to avoid.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
