nsivabalan commented on code in PR #19867:
URL: https://github.com/apache/hudi/pull/19867#discussion_r4050377299
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/heartbeat/HoodieHeartbeatClient.java:
##########
@@ -196,15 +196,34 @@ private boolean isHeartbeatStarted(Heartbeat heartbeat) {
private void stopHeartbeatScheduler(Heartbeat heartbeat) {
log.info("Stopping heartbeat for instant {}", heartbeat.getInstantTime());
shutdownHeartbeatScheduler(heartbeat);
+ // Callers delete the heartbeat file next. A refresh landing after that
delete recreates the
+ // file, and on storage that enforces preconditions it also changes the
object generation, so a
+ // generation-matched delete is rejected (e.g. GCS 412 conditionNotMet).
+ awaitHeartbeatSchedulerTermination(heartbeat);
heartbeat.setHeartbeatStopped(true);
log.info("Stopped heartbeat for instant {}", heartbeat.getInstantTime());
}
+ /** Stops further refreshes without waiting; safe to call from the scheduler
thread itself. */
private void shutdownHeartbeatScheduler(Heartbeat heartbeat) {
if (heartbeat.getScheduledFuture() != null) {
heartbeat.getScheduledFuture().cancel(false);
}
- heartbeat.getHeartbeatScheduler().shutdownNow();
+ heartbeat.getHeartbeatScheduler().shutdown();
+ }
+
+ private void awaitHeartbeatSchedulerTermination(Heartbeat heartbeat) {
+ // An in-flight tick can be parked on the bounded write, so allow for that
plus one interval.
+ long timeoutMs = this.heartbeatWriteTimeoutMs + this.heartbeatIntervalInMs;
+ try {
+ if (!heartbeat.getHeartbeatScheduler().awaitTermination(timeoutMs,
TimeUnit.MILLISECONDS)) {
+ log.warn("Timed out after {} ms awaiting an in-flight heartbeat
refresh for instant {}",
Review Comment:
shouldn't we throw here?
--
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]
