szetszwo commented on code in PR #9953:
URL: https://github.com/apache/ozone/pull/9953#discussion_r3010895923
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/safemode/AbstractContainerSafeModeRule.java:
##########
@@ -133,16 +153,87 @@ public double getCurrentContainerThreshold() {
return total == 0 ? 1 : ((double) getNumberOfContainersWithMinReplica() /
total);
}
+
+
+ private void startRefreshExecutor() {
+ if (refreshInterval <= 0) {
+ SCMSafeModeManager.getLogger().info(
+ "Container safe mode rule incremental sync is disabled ({}=0).",
+ HDDS_SCM_SAFEMODE_CONTAINER_RULE_REFRESH_INTERVAL);
+ return;
+ }
+ refreshExecutor = Executors.newSingleThreadScheduledExecutor(
+ new ThreadFactoryBuilder()
+ .setDaemon(true)
+ .setNameFormat(
+ "ContainerSafeModeRule-" + getContainerType() + "-refresh-%d")
+ .build());
+ refreshExecutor.scheduleAtFixedRate(
+ this::runRefresh,
+ refreshInterval,
+ refreshInterval,
+ TimeUnit.MILLISECONDS);
+ }
+
+ /**
+ * Background task: reconcile open/closed container tracking with
+ * {@link ContainerManager} while SCM is in safe mode.
+ */
+ private void runRefresh() {
+ synchronized (this) {
+ if (!scmInSafeMode()) {
+ return;
+ }
+ refreshExpectedContainers();
+ }
+ }
+
@Override
public synchronized void refresh(boolean forceRefresh) {
- if (forceRefresh || !validate()) {
+ if (forceRefresh) {
initializeRule();
}
}
+ /**
+ * Aligns tracked open/closed sets with current {@link ContainerManager}
state.
+ * Runs on the incremental sync thread when configured.
+ */
+ private void refreshExpectedContainers() {
+ if (getSafeModeManager().isScmRatisApplyCaughtUpToCommit()) {
+ return;
+ }
+ if (!validate()) {
+ initializeRule();
+ }
+ }
+
+
@Override
protected void cleanup() {
- getContainers().clear();
+ stopIncrementalSyncExecutor();
+ synchronized (this) {
+ if (containers != null) {
+ containers.clear();
+ }
+ }
+ }
+
+ private void stopIncrementalSyncExecutor() {
+ if (refreshExecutor != null) {
+ refreshExecutor.shutdownNow();
+ try {
+ refreshExecutor.awaitTermination(5, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ refreshExecutor = null;
+ }
+ }
+
+ @VisibleForTesting
+ void runIncrementalContainerSyncForTesting() {
Review Comment:
The method is unused. Let's remove it.
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/safemode/AbstractContainerSafeModeRule.java:
##########
@@ -51,12 +57,27 @@ public abstract class AbstractContainerSafeModeRule extends
SafeModeExitRule<Nod
private final AtomicInteger totalContainers = new AtomicInteger();
private final AtomicInteger containersWithMinReplicas = new AtomicInteger();
+ private final long refreshInterval;
+ private volatile ScheduledExecutorService refreshExecutor;
public AbstractContainerSafeModeRule(ConfigurationSource conf,
SCMSafeModeManager safeModeManager,
ContainerManager containerManager, EventQueue eventQueue) {
super(safeModeManager, eventQueue);
this.containerManager = containerManager;
this.safeModeCutoff = getSafeModeCutoff(conf);
+ this.refreshInterval = conf.getTimeDuration(
+ HDDS_SCM_SAFEMODE_CONTAINER_RULE_REFRESH_INTERVAL,
+ HDDS_SCM_SAFEMODE_CONTAINER_RULE_REFRESH_INTERVAL_DEFAULT,
+ TimeUnit.MILLISECONDS);
initializeRule();
+ startRefreshExecutor();
+ }
+
+ public ContainerManager getContainerManager() {
+ return containerManager;
+ }
+
+ void incrementTotalContainers() {
+ totalContainers.getAndIncrement();
Review Comment:
These two methods are unused. Let's remove them.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
