This is an automated email from the ASF dual-hosted git repository.
ArafatKhan2198 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 9d0339cbfaf HDDS-14989. Delay follower SCM DN server start until Ratis
log catch-up. (#10617).
9d0339cbfaf is described below
commit 9d0339cbfaf44ddb2833cc63825bd045b5fd196c
Author: Arafat2198 <[email protected]>
AuthorDate: Wed Jul 15 12:13:01 2026 +0530
HDDS-14989. Delay follower SCM DN server start until Ratis log catch-up.
(#10617).
Co-authored-by: Claude Haiku 4.5 <[email protected]>
Co-authored-by: XiChen <[email protected]>
---
.../container/AbstractContainerReportHandler.java | 26 +-
.../apache/hadoop/hdds/scm/ha/SCMStateMachine.java | 135 +++++++--
.../hdds/scm/server/StorageContainerManager.java | 9 +-
.../TestSCMFollowerCatchupWithContainerReport.java | 336 +++++++++++++++++++++
.../hdds/scm/safemode/TestSafeModeSCMHA.java | 18 +-
5 files changed, 495 insertions(+), 29 deletions(-)
diff --git
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/AbstractContainerReportHandler.java
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/AbstractContainerReportHandler.java
index daf821a8f43..e13b0a461d1 100644
---
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/AbstractContainerReportHandler.java
+++
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/AbstractContainerReportHandler.java
@@ -252,7 +252,7 @@ private boolean updateContainerState(final DatanodeDetails
datanode,
// If the state of a container is OPEN and a replica is in different
state, finalize the container.
if (replica.getState() != State.OPEN) {
getLogger().info("FINALIZE (i.e. CLOSING) {}", detailsForLogging);
- containerManager.updateContainerState(containerId,
LifeCycleEvent.FINALIZE);
+ updateContainerState(containerId, LifeCycleEvent.FINALIZE);
}
return false;
case CLOSING:
@@ -263,7 +263,7 @@ private boolean updateContainerState(final DatanodeDetails
datanode,
// If the replica is in QUASI_CLOSED state, move the container to
QUASI_CLOSED state.
if (replica.getState() == State.QUASI_CLOSED) {
getLogger().info("QUASI_CLOSE {}", detailsForLogging);
- containerManager.updateContainerState(containerId,
LifeCycleEvent.QUASI_CLOSE);
+ updateContainerState(containerId, LifeCycleEvent.QUASI_CLOSE);
return false;
}
@@ -287,7 +287,7 @@ private boolean updateContainerState(final DatanodeDetails
datanode,
return true;
}
getLogger().info("CLOSE {}", detailsForLogging);
- containerManager.updateContainerState(containerId,
LifeCycleEvent.CLOSE);
+ updateContainerState(containerId, LifeCycleEvent.CLOSE);
}
return false;
case QUASI_CLOSED:
@@ -300,7 +300,7 @@ private boolean updateContainerState(final DatanodeDetails
datanode,
return true;
}
getLogger().info("FORCE_CLOSE for {}", detailsForLogging);
- containerManager.updateContainerState(containerId,
LifeCycleEvent.FORCE_CLOSE);
+ updateContainerState(containerId, LifeCycleEvent.FORCE_CLOSE);
}
return false;
case CLOSED:
@@ -361,6 +361,24 @@ private boolean updateContainerState(final DatanodeDetails
datanode,
}
}
+ /**
+ * Apply a container lifecycle state transition, but only on the leader SCM.
+ * On a follower the underlying {@code containerManager.updateContainerState}
+ * is a Ratis write and would throw {@code NotLeaderException}, which would
+ * abort {@code processContainerReplica} and skip recording the replica
+ * location. Skipping the state change on a follower is safe: the leader
+ * drives the transition and it replicates back via the Ratis log.
+ */
+ private void updateContainerState(ContainerID containerID, LifeCycleEvent
event)
+ throws IOException {
+ if (scmContext.isLeader()) {
+ containerManager.updateContainerState(containerID, event);
+ } else {
+ getLogger().debug("Skipping updateContainerState on non-leader SCM,
container {} event {}",
+ containerID, event);
+ }
+ }
+
/**
* Helper method to verify that the replica's bcsId matches the container's
in SCM.
*
diff --git
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMStateMachine.java
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMStateMachine.java
index f6295fa7a7c..5036769ed3a 100644
---
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMStateMachine.java
+++
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMStateMachine.java
@@ -32,7 +32,6 @@
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicLong;
import org.apache.hadoop.hdds.protocol.proto.SCMRatisProtocol.RequestType;
import org.apache.hadoop.hdds.scm.block.DeletedBlockLog;
import org.apache.hadoop.hdds.scm.block.DeletedBlockLogImpl;
@@ -86,9 +85,14 @@ public class SCMStateMachine extends BaseStateMachine {
private DBCheckpoint installingDBCheckpoint = null;
private List<ManagedSecretKey> installingSecretKeys = null;
- private AtomicLong currentLeaderTerm = new AtomicLong(-1L);
private AtomicBoolean isStateMachineReady = new AtomicBoolean();
+ // The leader's committed index captured when this SCM (re)joins as a
+ // follower. Catch-up is measured against this fixed target rather than the
+ // leader's live commit index, which on a busy cluster keeps advancing and
+ // would never be reached. Set only while not yet ready; -1 means uncaptured.
+ private volatile long leaderCommitIndexOnStart = -1L;
+
public SCMStateMachine(final StorageContainerManager scm,
SCMHADBTransactionBuffer buffer) {
this.scm = scm;
@@ -172,9 +176,15 @@ public CompletableFuture<Message> applyTransaction(
// Ratis client, leaving SCM intact.
applyTransactionFuture.completeExceptionally(ex);
}
+
final TermIndex appliedTermIndex = TermIndex.valueOf(trx.getLogEntry());
transactionBuffer.updateLatestTrxInfo(TransactionInfo.valueOf(appliedTermIndex));
updateLastAppliedTermIndex(appliedTermIndex);
+
+ // A restarted follower may catch up by applying data-carrying entries
+ // here rather than through notifyTermIndexUpdated, so check for catch-up
+ // in both places. No-op once the datanode protocol server has started.
+ tryStartDNServerAndRefreshSafeMode();
} catch (Exception ex) {
applyTransactionFuture.completeExceptionally(ex);
ExitUtils.terminate(1, ex.getMessage(), ex, StateMachine.LOG);
@@ -278,18 +288,19 @@ public void notifyLeaderChanged(RaftGroupMemberId
groupMemberId,
return;
}
- currentLeaderTerm.set(scm.getScmHAManager().getRatisServer().getDivision()
- .getInfo().getCurrentTerm());
-
- if (isStateMachineReady.compareAndSet(false, true)) {
- // refresh and validate safe mode rules if it can exit safe mode
- // if being leader, all previous term transactions have been applied
- // if other states, just refresh safe mode rules, and transaction keeps
flushing from leader
- // and does not depend on pending transactions.
- scm.getScmSafeModeManager().refreshAndValidate();
- }
-
- if (!groupMemberId.getPeerId().equals(newLeaderId)) {
+ final boolean isLeader = groupMemberId.getPeerId().equals(newLeaderId);
+
+ if (!isLeader) {
+ // Follower: capture the (possibly new) leader's current committed index
+ // as the fixed catch-up target, then start the datanode protocol server
+ // if we are already caught up with it; otherwise applyTransaction /
+ // notifyTermIndexUpdated start it as catch-up completes. Set it always:
+ // getLeaderCommitIndex() returns -1 when the leader is not known yet,
+ // which isFollowerCaughtUp() treats as uncaptured and re-reads later.
+ if (!isStateMachineReady.get()) {
+ leaderCommitIndexOnStart = getLeaderCommitIndex();
+ }
+ tryStartDNServerAndRefreshSafeMode();
String message = "Leader changed to " + newLeaderId +
", current SCM " + scm.getScmId() + " is still follower.";
LOG.info(message);
@@ -297,15 +308,20 @@ public void notifyLeaderChanged(RaftGroupMemberId
groupMemberId,
return;
}
+ long currentTerm = scm.getScmHAManager().getRatisServer().getDivision()
+ .getInfo().getCurrentTerm();
String message = "current SCM " + scm.getScmId() +
- " becomes leader of term " + currentLeaderTerm;
+ " becomes leader of term " + currentTerm;
LOG.info(message);
addRatisEvent(message);
- scm.getScmContext().updateLeaderAndTerm(true,
- currentLeaderTerm.get());
+ scm.getScmContext().updateLeaderAndTerm(true, currentTerm);
scm.getSequenceIdGen().invalidateBatch();
+ // isLeader() is now true -> start the datanode protocol server for the new
+ // leader (a leader has applied all committed entries) and refresh safe
mode.
+ tryStartDNServerAndRefreshSafeMode();
+
try {
transactionBuffer.flush();
} catch (Exception ex) {
@@ -380,16 +396,93 @@ public void notifyTermIndexUpdated(long term, long index)
{
}
}
- if (currentLeaderTerm.get() == term) {
- // This means after a restart, all pending transactions have been
applied.
+ // As committed entries are applied (e.g. a restarted follower catching
up),
+ // start the datanode protocol server once we are caught up with the
leader's
+ // committed index. No-op once the server has already been started.
+ tryStartDNServerAndRefreshSafeMode();
+ }
+
+ /**
+ * Start the DatanodeProtocolServer and re-evaluate safe-mode rules, but only
+ * when this SCM is safe to accept datanode reports: it is the leader, or it
+ * is a follower whose state machine has caught up with the leader's
committed
+ * log. Guarded by {@code isStateMachineReady} (CAS) so the non-idempotent
+ * {@code DatanodeProtocolServer.start()} runs exactly once.
+ *
+ * <p>In HA mode {@link StorageContainerManager#start()} deliberately does
not
+ * start the datanode protocol server; it is deferred to here so datanode
+ * container reports are processed against the up-to-date container/pipeline
+ * state rather than a stale, mid-replay snapshot.
+ */
+ private void tryStartDNServerAndRefreshSafeMode() {
+ if (isStateMachineReady.get()) {
+ return;
+ }
+ if (scm.getScmContext().isLeader() || isFollowerCaughtUp()) {
if (isStateMachineReady.compareAndSet(false, true)) {
- // Refresh Safemode rules state if not already done.
+ scm.getDatanodeProtocolServer().start();
scm.getScmSafeModeManager().refreshAndValidate();
}
- currentLeaderTerm.set(-1L);
}
}
+ /**
+ * @return true if this follower's last applied index has reached the
leader's
+ * committed index captured when it (re)joined, i.e. all transactions the
+ * leader had committed at that point have been replayed. Comparing against a
+ * fixed target avoids chasing the leader's ever-advancing live commit index.
+ */
+ private boolean isFollowerCaughtUp() {
+ try {
+ long target = leaderCommitIndexOnStart;
+ if (target < 0) {
+ // Not captured at leader-change time yet; capture the leader's current
+ // commit index once here so we still compare against a fixed target.
+ target = getLeaderCommitIndex();
+ if (target < 0) {
+ // Normal transient condition during startup/catch-up; this is polled
+ // from multiple callbacks, so keep it at DEBUG to avoid log
flooding.
+ LOG.debug("Leader commit index not available yet");
+ return false;
+ }
+ leaderCommitIndexOnStart = target;
+ }
+
+ long lastAppliedIndex = scm.getScmHAManager().getRatisServer()
+ .getDivision().getInfo().getLastAppliedIndex();
+ boolean caughtUp = lastAppliedIndex >= target;
+ if (caughtUp) {
+ LOG.info("Follower caught up with leader: lastAppliedIndex={},
leaderCommitOnStart={}",
+ lastAppliedIndex, target);
+ } else {
+ LOG.debug("Follower not caught up: lastAppliedIndex={},
leaderCommitOnStart={}",
+ lastAppliedIndex, target);
+ }
+ return caughtUp;
+ } catch (Exception e) {
+ LOG.warn("Failed to check follower catch-up status", e);
+ return false;
+ }
+ }
+
+ /**
+ * @return the leader's current committed index as seen by this SCM, or -1 if
+ * the leader or its commit info is not available yet.
+ */
+ private long getLeaderCommitIndex() {
+ RaftServer.Division division = scm.getScmHAManager()
+ .getRatisServer().getDivision();
+ RaftPeerId leaderId = division.getInfo().getLeaderId();
+ if (leaderId != null) {
+ for (RaftProtos.CommitInfoProto info : division.getCommitInfos()) {
+ if (info.getServer().getId().equals(leaderId.toByteString())) {
+ return info.getCommitIndex();
+ }
+ }
+ }
+ return -1L;
+ }
+
public boolean getIsStateMachineReady() {
return isStateMachineReady.get();
}
diff --git
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/StorageContainerManager.java
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/StorageContainerManager.java
index 13e297b9cae..100c9feb9b6 100644
---
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/StorageContainerManager.java
+++
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/StorageContainerManager.java
@@ -1585,8 +1585,13 @@ public void start() throws IOException {
}
getBlockProtocolServer().start();
- // start datanode protocol server
- getDatanodeProtocolServer().start();
+ // In HA mode, defer starting the datanode protocol server until the SCM
+ // state machine has caught up with the leader's committed log entries
+ // (see SCMStateMachine#tryStartDNServerAndRefreshSafeMode). In non-HA mode
+ // there is no Ratis state machine, so start it here as before.
+ if (!scmStorageConfig.isSCMHAEnabled()) {
+ getDatanodeProtocolServer().start();
+ }
if (getSecurityProtocolServer() != null) {
getSecurityProtocolServer().start();
persistSCMCertificates();
diff --git
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestSCMFollowerCatchupWithContainerReport.java
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestSCMFollowerCatchupWithContainerReport.java
new file mode 100644
index 00000000000..1e172d7be5c
--- /dev/null
+++
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestSCMFollowerCatchupWithContainerReport.java
@@ -0,0 +1,336 @@
+/*
+ * 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.hadoop.hdds.scm;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static
org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState.CLOSED;
+import static
org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor.THREE;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+import java.io.IOException;
+import java.util.LinkedHashSet;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BooleanSupplier;
+import org.apache.hadoop.hdds.client.RatisReplicationConfig;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState;
+import org.apache.hadoop.hdds.scm.container.ContainerID;
+import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
+import org.apache.hadoop.ozone.MiniOzoneCluster;
+import org.apache.hadoop.ozone.MiniOzoneHAClusterImpl;
+import org.apache.hadoop.ozone.TestDataUtil;
+import org.apache.hadoop.ozone.client.ObjectStore;
+import org.apache.hadoop.ozone.client.OzoneBucket;
+import org.apache.hadoop.ozone.client.OzoneClient;
+import org.apache.hadoop.ozone.client.OzoneKeyDetails;
+import org.apache.hadoop.ozone.client.OzoneVolume;
+import org.apache.hadoop.ozone.client.io.OzoneInputStream;
+import org.apache.ozone.test.GenericTestUtils;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Verifies that a follower SCM correctly rebuilds container replica locations
+ * for containers that were <em>created</em> while it was offline. After the
+ * follower restarts, catches up its Ratis log, and is promoted to leader, all
+ * such containers must still have the expected replica count and all keys must
+ * be readable.
+ *
+ * <p>This class covers the create-while-down, close-while-down, and
+ * idle-cluster scenarios for HDDS-14989. It exercises
+ * the deferred datanode-server start: the restarted follower must finish Raft
+ * log replay <em>before</em> accepting datanode container reports, otherwise a
+ * report for a not-yet-replayed container is dropped with CONTAINER_NOT_FOUND
+ * and the replica location is lost until the next full container report.
+ *
+ * <p>The container report interval is set high so that, without the fix, the
+ * dropped replicas are not re-reported within the test window and the
+ * assertions fail; with the fix the datanode server is deferred until
catch-up,
+ * datanodes (re)register against the up-to-date state, and replicas are
+ * recorded immediately.
+ */
+@Timeout(300)
+public class TestSCMFollowerCatchupWithContainerReport {
+ private static final Logger LOG =
+ LoggerFactory.getLogger(TestSCMFollowerCatchupWithContainerReport.class);
+
+ private static final String OM_SERVICE_ID = "om-service-test1";
+ private static final String SCM_SERVICE_ID = "scm-service-test1";
+ private static final int NUM_OF_SCMS = 3;
+ private static final int NUM_OF_DNS = 3;
+ private static final int NUM_KEYS = 5;
+
+ // One cluster is shared by all tests in this class (built once in
@BeforeAll).
+ // Each test uses its own volume/bucket and re-discovers leader/follower, so
the
+ // restart + leadership-transfer each test performs leaves the cluster
healthy
+ // for the next one.
+ private static MiniOzoneHAClusterImpl cluster;
+
+ @BeforeAll
+ static void init() throws Exception {
+ OzoneConfiguration conf = new OzoneConfiguration();
+ // Keep the full container report interval long so a replica dropped during
+ // catch-up is not silently re-reported within the test window. This makes
+ // the regression deterministic: only the deferred-start path can
repopulate
+ // replicas in time.
+ conf.setTimeDuration("hdds.container.report.interval", 5,
TimeUnit.MINUTES);
+ // Fast datanode heartbeats so safe-mode exit at startup and replica
+ // re-reporting after the deferred DN-server start happen within ~1s.
+ conf.setTimeDuration("hdds.heartbeat.interval", 1, TimeUnit.SECONDS);
+ cluster = MiniOzoneCluster.newHABuilder(conf)
+ .setOMServiceId(OM_SERVICE_ID)
+ .setSCMServiceId(SCM_SERVICE_ID)
+ .setNumOfOzoneManagers(1)
+ .setNumOfStorageContainerManagers(NUM_OF_SCMS)
+ .setNumOfActiveSCMs(NUM_OF_SCMS)
+ .build();
+ cluster.waitForClusterToBeReady();
+ }
+
+ @AfterAll
+ static void shutdown() {
+ if (cluster != null) {
+ cluster.shutdown();
+ }
+ }
+
+ /**
+ * HDDS-14989 scenario: containers are closed while a follower SCM is
offline.
+ * After the follower restarts and is promoted to leader, each container must
+ * be CLOSED with a full replica set and all keys must remain readable.
+ */
+ @Test
+ void testFollowerCatchupAfterContainerClose() throws Exception {
+ String vol = "vol-close";
+ String buck = "buck-close";
+ byte[] keyData = "value-of-key".getBytes(UTF_8);
+ Set<Long> containerIds = createKeys(vol, buck, keyData);
+ assertFalse(containerIds.isEmpty(), "Should have created containers");
+
+ StorageContainerManager followerScm = null;
+ for (StorageContainerManager scm : cluster.getStorageContainerManagers()) {
+ if (!scm.checkLeader() && followerScm == null) {
+ followerScm = scm;
+ }
+ }
+ assertFalse(followerScm == null, "Expected to find a follower SCM");
+
+ cluster.shutdownStorageContainerManager(followerScm);
+ followerScm.join();
+
+ for (long cid : containerIds) {
+ cluster.getStorageContainerLocationClient().closeContainer(cid);
+ }
+ for (long cid : containerIds) {
+ waitForContainerState(cluster.getActiveSCM(), ContainerID.valueOf(cid),
CLOSED);
+ }
+
+ StorageContainerManager newFollower =
+ cluster.restartStorageContainerManager(followerScm, false);
+ GenericTestUtils.waitFor(() -> !newFollower.isInSafeMode(), 250, 120_000);
+
+ cluster.getStorageContainerLocationClient()
+ .transferLeadership(newFollower.getScmId());
+ GenericTestUtils.waitFor(newFollower::checkLeader, 250, 60_000);
+
+ for (long cid : containerIds) {
+ ContainerID id = ContainerID.valueOf(cid);
+ assertEquals(CLOSED,
+ newFollower.getContainerManager().getContainer(id).getState(),
+ "Container " + cid + " should be CLOSED");
+ waitForReplicaCount(newFollower, id, NUM_OF_DNS);
+ assertEquals(NUM_OF_DNS,
+ newFollower.getContainerManager().getContainerReplicas(id).size(),
+ "Container " + cid + " should have " + NUM_OF_DNS + " replicas");
+ }
+ assertKeysReadable(vol, buck, keyData);
+ }
+
+ /**
+ * Reproduces the production failure: containers are created while a follower
+ * SCM is offline. After the follower restarts and is promoted to leader, the
+ * containers must have full replica sets (not an empty replica list).
+ */
+ @Test
+ void testFollowerCatchupAfterContainerCreate() throws Exception {
+ // ---- Step 1: pick a leader and a follower ----
+ StorageContainerManager followerScm = null;
+ for (StorageContainerManager scm : cluster.getStorageContainerManagers()) {
+ if (!scm.checkLeader() && followerScm == null) {
+ followerScm = scm;
+ }
+ }
+ assertFalse(followerScm == null, "Expected to find a follower SCM");
+
+ // ---- Step 2: stop the follower BEFORE creating containers, so it misses
+ // the container-create transactions entirely ----
+ cluster.shutdownStorageContainerManager(followerScm);
+ followerScm.join();
+
+ // ---- Step 3: create keys -> new containers created while follower
offline.
+ String vol = "vol-create";
+ String buck = "buck-create";
+ byte[] keyData = "value-of-key".getBytes(UTF_8);
+ Set<Long> containerIds = createKeys(vol, buck, keyData);
+ assertFalse(containerIds.isEmpty(), "Should have created containers");
+
+ // ---- Step 4: restart the follower and wait for safe-mode exit ----
+ StorageContainerManager newFollower =
+ cluster.restartStorageContainerManager(followerScm, false);
+ BooleanSupplier safeModeExited = () -> !newFollower.isInSafeMode();
+ GenericTestUtils.waitFor(safeModeExited, 250, 120_000);
+
+ // ---- Step 5: transfer leadership to the restarted follower ----
+ cluster.getStorageContainerLocationClient()
+ .transferLeadership(newFollower.getScmId());
+ GenericTestUtils.waitFor(newFollower::checkLeader, 250, 60_000);
+ LOG.info("Leadership transferred to {}", newFollower.getScmId());
+
+ // ---- Step 6: every container must have a full replica set on the new
+ // leader (the bug shows replicas == 0) ----
+ for (long cid : containerIds) {
+ ContainerID id = ContainerID.valueOf(cid);
+ waitForReplicaCount(newFollower, id, NUM_OF_DNS);
+ assertEquals(NUM_OF_DNS,
+ newFollower.getContainerManager().getContainerReplicas(id).size(),
+ "Container " + cid + " should have " + NUM_OF_DNS + " replicas");
+ }
+
+ // ---- Step 7: every key must still be readable ----
+ assertKeysReadable(vol, buck, keyData);
+ }
+
+ /**
+ * Edge case for removing the background polling loop: on an otherwise idle
+ * cluster a restarted follower must still start its datanode server (exit
safe
+ * mode) and serve replicas after promotion, driven by Ratis heartbeats /
+ * notifyLeaderChanged rather than a steady stream of new transactions.
+ */
+ @Test
+ void testFollowerCatchupOnIdleCluster() throws Exception {
+ String vol = "vol-idle";
+ String buck = "buck-idle";
+ byte[] keyData = "value-of-key".getBytes(UTF_8);
+ Set<Long> containerIds = createKeys(vol, buck, keyData);
+ assertFalse(containerIds.isEmpty(), "Should have created containers");
+
+ StorageContainerManager followerScm = null;
+ for (StorageContainerManager scm : cluster.getStorageContainerManagers()) {
+ if (!scm.checkLeader() && followerScm == null) {
+ followerScm = scm;
+ }
+ }
+ assertFalse(followerScm == null, "Expected to find a follower SCM");
+
+ // Stop the follower, then do NO further writes (idle cluster).
+ cluster.shutdownStorageContainerManager(followerScm);
+ followerScm.join();
+
+ StorageContainerManager newFollower =
+ cluster.restartStorageContainerManager(followerScm, false);
+ // Must still exit safe mode (i.e. the datanode server started) without any
+ // new transactions to apply.
+ GenericTestUtils.waitFor(() -> !newFollower.isInSafeMode(), 250, 120_000);
+
+ cluster.getStorageContainerLocationClient()
+ .transferLeadership(newFollower.getScmId());
+ GenericTestUtils.waitFor(newFollower::checkLeader, 250, 60_000);
+
+ for (long cid : containerIds) {
+ ContainerID id = ContainerID.valueOf(cid);
+ waitForReplicaCount(newFollower, id, NUM_OF_DNS);
+ assertEquals(NUM_OF_DNS,
+ newFollower.getContainerManager().getContainerReplicas(id).size(),
+ "Container " + cid + " should have " + NUM_OF_DNS + " replicas");
+ }
+ assertKeysReadable(vol, buck, keyData);
+ }
+
+ private Set<Long> createKeys(String volumeName, String bucketName, byte[]
keyData)
+ throws IOException {
+ Set<Long> containerIds = new LinkedHashSet<>();
+ try (OzoneClient client = cluster.newClient()) {
+ ObjectStore store = client.getObjectStore();
+ store.createVolume(volumeName);
+ OzoneVolume volume = store.getVolume(volumeName);
+ volume.createBucket(bucketName);
+ OzoneBucket bucket = volume.getBucket(bucketName);
+
+ for (int i = 0; i < NUM_KEYS; i++) {
+ String keyName = "key-" + i;
+ TestDataUtil.createKey(bucket, keyName,
+ RatisReplicationConfig.getInstance(THREE), keyData);
+ OzoneKeyDetails keyDetails = bucket.getKey(keyName);
+ keyDetails.getOzoneKeyLocations()
+ .forEach(loc -> containerIds.add(loc.getContainerID()));
+ }
+ }
+ return containerIds;
+ }
+
+ private void assertKeysReadable(String volumeName, String bucketName, byte[]
keyData)
+ throws IOException {
+ try (OzoneClient client = cluster.newClient()) {
+ ObjectStore store = client.getObjectStore();
+ OzoneBucket bucket = store.getVolume(volumeName).getBucket(bucketName);
+ for (int i = 0; i < NUM_KEYS; i++) {
+ String keyName = "key-" + i;
+ try (OzoneInputStream is = bucket.readKey(keyName)) {
+ byte[] readData = new byte[keyData.length];
+ int bytesRead = is.read(readData);
+ assertEquals(keyData.length, bytesRead);
+ assertArrayEquals(keyData, readData);
+ }
+ }
+ }
+ }
+
+ private static void waitForContainerState(
+ StorageContainerManager scm, ContainerID id, LifeCycleState
expectedState)
+ throws Exception {
+ GenericTestUtils.waitFor(() -> {
+ try {
+ return scm.getContainerManager().getContainer(id).getState()
+ == expectedState;
+ } catch (Exception e) {
+ return false;
+ }
+ }, 250, 120_000);
+ }
+
+ private static void waitForReplicaCount(
+ StorageContainerManager scm, ContainerID id, int expectedCount)
+ throws Exception {
+ BooleanSupplier check = () -> {
+ try {
+ return scm.getContainerManager().getContainerReplicas(id).size()
+ == expectedCount;
+ } catch (Exception e) {
+ return false;
+ }
+ };
+ GenericTestUtils.waitFor(check, 250, 120_000);
+ }
+}
diff --git
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/safemode/TestSafeModeSCMHA.java
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/safemode/TestSafeModeSCMHA.java
index 27d0c81923e..6f85c096980 100644
---
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/safemode/TestSafeModeSCMHA.java
+++
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/safemode/TestSafeModeSCMHA.java
@@ -91,9 +91,23 @@ public void testFollowerRestartExitSafeMode() throws
Exception {
GenericTestUtils.waitFor(() ->
leaderScmStateMachine.getLastAppliedTermIndex().getIndex()
== followerScmStateMachine.getLastAppliedTermIndex().getIndex(), 1000,
60000);
- // wait for follower to exit safe mode
+ // Wait for the restarted follower to exit safe mode. A live cluster has
+ // ongoing activity that advances the Ratis log and re-drives the
follower's
+ // catch-up check (via applyTransaction / notifyTermIndexUpdated); a
totally
+ // idle cluster provides no such trigger. Generate some SCM activity by
+ // allocating containers on the leader so the follower catches up, starts
its
+ // datanode protocol server and exits safe mode.
StorageContainerManager newFollowerScm =
cluster.restartStorageContainerManager(followerScm, false);
- GenericTestUtils.waitFor(() -> !newFollowerScm.isInSafeMode(), 1000,
60000);
+ final StorageContainerManager currentLeader = leaderScm;
+ GenericTestUtils.waitFor(() -> {
+ try {
+ currentLeader.getContainerManager().allocateContainer(
+ RatisReplicationConfig.getInstance(THREE), "safemode-test");
+ } catch (Exception e) {
+ // Ignore transient errors while the follower is catching up.
+ }
+ return !newFollowerScm.isInSafeMode();
+ }, 1000, 60000);
}
private void createTestData(OzoneClient client) throws IOException {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
