zaynt4606 commented on code in PR #3810: URL: https://github.com/apache/celeborn/pull/3810#discussion_r4312979005
########## master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/HeartbeatAggregator.java: ########## @@ -0,0 +1,153 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.celeborn.service.deploy.master.clustermeta.ha; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.ReentrantLock; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.celeborn.common.CelebornConf; +import org.apache.celeborn.common.client.MasterClient; +import org.apache.celeborn.common.util.ThreadUtils; +import org.apache.celeborn.service.deploy.master.clustermeta.ResourceProtos; +import org.apache.celeborn.service.deploy.master.clustermeta.ResourceProtos.ResourceRequest; +import org.apache.celeborn.service.deploy.master.clustermeta.ResourceProtos.Type; + +public class HeartbeatAggregator { + private static final Logger LOG = LoggerFactory.getLogger(HeartbeatAggregator.class); + + private final HARaftServer ratisServer; + private final long batchIntervalMs; + + private final ReentrantLock pendingLock = new ReentrantLock(); + private Map<String, ResourceProtos.WorkerHeartbeatRequest> workerHeartbeats = new HashMap<>(); + private Map<String, ResourceProtos.AppHeartbeatRequest> appHeartbeats = new HashMap<>(); + + private final ScheduledExecutorService flushExecutor; + + public HeartbeatAggregator(HARaftServer ratisServer, CelebornConf conf) { + this.ratisServer = ratisServer; + this.batchIntervalMs = conf.masterHaHeartbeatBatchIntervalMs(); + this.flushExecutor = + ThreadUtils.newDaemonSingleThreadScheduledExecutor("master-heartbeat-aggregator"); + this.flushExecutor.scheduleWithFixedDelay( + this::flushSafely, batchIntervalMs, batchIntervalMs, TimeUnit.MILLISECONDS); + LOG.info("HeartbeatAggregator started, flush interval {} ms.", batchIntervalMs); + } + + public void offerWorkerHeartbeat(ResourceProtos.WorkerHeartbeatRequest heartbeat) { + pendingLock.lock(); + try { + workerHeartbeats.put(workerKey(heartbeat), heartbeat); + } finally { + pendingLock.unlock(); + } + } + + public void offerAppHeartbeat(ResourceProtos.AppHeartbeatRequest heartbeat) { + pendingLock.lock(); + try { + appHeartbeats.put(heartbeat.getAppId(), heartbeat); Review Comment: [P2] Preserve application heartbeat deltas when coalescing The application heartbeat fields are not all snapshots. `CommitHandler.commitMetrics()` uses `sumThenReset()` for written bytes/file counts, and `LifecycleManager` also resets the shuffle/application counters and fallback counts after collecting them. `updateAppHeartbeatMeta()` adds these deltas. Consequently, replacing the pending heartbeat for the same app loses the earlier heartbeat's counters permanently, even though it has already been acknowledged successfully. This can happen when a slow Raft submission delays subsequent flushes, or the configured batch interval exceeds the application's heartbeat interval. The next heartbeat cannot recover the discarded deltas. I reproduced this with the actual metadata manager/handler and a mocked synchronous Raft submission: two heartbeats reporting 100 and 200 written bytes produce 200 with batching enabled, versus 300 with batching disabled. File, shuffle, application and fallback counts are also undercounted. Besides metrics, missing written bytes/file counts can bias partition-size estimation and disk-slot capacity. Could we accumulate the delta fields while retaining the latest timestamp, and update `testDuplicateAppHeartbeatKeepsNewest` to verify that distinct heartbeat deltas are conserved? ########## master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/HAMasterMetaManager.java: ########## @@ -168,22 +177,27 @@ public void handleAppHeartbeat( Map<String, Long> applicationFallbackCounts, long time, String requestId) { + ResourceProtos.AppHeartbeatRequest appHeartbeatRequest = + ResourceProtos.AppHeartbeatRequest.newBuilder() + .setAppId(appId) + .setTime(time) + .setTotalWritten(totalWritten) + .setFileCount(fileCount) + .setShuffleCount(shuffleCount) + .setApplicationCount(applicationCount) + .putAllShuffleFallbackCounts(shuffleFallbackCounts) + .putAllApplicationFallbackCounts(applicationFallbackCounts) + .build(); + if (heartbeatAggregator != null) { + heartbeatAggregator.offerAppHeartbeat(appHeartbeatRequest); + return; Review Comment: [P2] Preserve ordering between buffered heartbeats and lifecycle operations Returning after enqueueing allows a successfully acknowledged heartbeat to be applied after a subsequent lifecycle operation, because `AppLost`, `WorkerLost` and `RegisterWorker` still submit directly to Raft. Two reproducible sequences with batching enabled are: - `AppHeartbeat -> AppLost -> flush`: `AppLost` removes the application, but the buffered heartbeat then recreates its `appHeartbeatTime` entry. A normal client shutdown can hit this sequence after stopping heartbeats and unregistering. The running-application count becomes incorrect and the app expires again later; its removed shuffle registrations are not restored. - `WorkerHeartbeat -> WorkerLost -> RegisterWorker` using the same host/ports, followed by `flush`: the old heartbeat overwrites the newly registered worker's workload state and `lastHeartbeat`. Other snapshot fields, including disk/status information, follow the same update path, potentially affecting slot allocation. I reproduced both using the actual metadata manager/handler with a mocked synchronous Raft submission. The corresponding non-batched controls preserve application removal and the new worker state. These sequences do not require a leader failure: the old synchronous path completes the heartbeat submission before acknowledging it, whereas the batched path leaves it pending. Could we preserve ordering across heartbeat batches and these lifecycle operations, including already-drained batches, and add regression coverage for both sequences? -- 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]
