You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ozone.apache.org by we...@apache.org on 2026/08/27 18:20:37 UTC
(ozone) 04/19: HDDS-14989. Delay follower SCM DN server start until Ratis log catch-up. (#10617).
This is an automated email from the ASF dual-hosted git repository.
jojochuang pushed a commit to branch ozone-2.1
in repository https://gitbox.apache.org/repos/asf/ozone.git
commit 173fea0d5b6eaf2fdd87d066c041b4c4e1c74f44
Author: Arafat2198 <ar...@gmail.com>
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 <no...@anthropic.com>
Co-authored-by: XiChen <32...@users.noreply.github.com>
(cherry picked from commit 9d0339cbfaf44ddb2833cc63825bd045b5fd196c)
Change-Id: I14150c5ea9cc6bb5789599b1ee680c6ec663df6e
---
.../container/AbstractContainerReportHandler.java | 26 +-
.../apache/hadoop/hdds/scm/ha/SCMStateMachine.java | 153 ++++++++--
.../hdds/scm/server/StorageContainerManager.java | 5 +-
.../TestSCMFollowerCatchupWithContainerReport.java | 336 +++++++++++++++++++++
4 files changed, 489 insertions(+), 31 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 214731f16d6..5fd12999074 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
@@ -250,7 +250,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:
@@ -261,7 +261,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;
}
@@ -286,7 +286,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:
@@ -299,7 +299,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:
@@ -330,6 +330,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 ef9ffc03f26..a516a217c08 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
@@ -34,7 +34,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;
@@ -85,8 +84,13 @@ public class SCMStateMachine extends BaseStateMachine {
private DBCheckpoint installingDBCheckpoint = null;
private List<ManagedSecretKey> installingSecretKeys = null;
- private AtomicLong currentLeaderTerm = new AtomicLong(-1L);
- private AtomicBoolean refreshedAfterLeaderReady = new AtomicBoolean();
+ 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) {
@@ -162,14 +166,15 @@ public CompletableFuture<Message> applyTransaction(
applyTransactionFuture.completeExceptionally(ex);
}
- // After previous term transactions are applied, still in safe mode,
- // perform refreshAndValidate to update the safemode rule state.
- if (scm.isInSafeMode() && refreshedAfterLeaderReady.get()) {
- scm.getScmSafeModeManager().refreshAndValidate();
- }
+
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);
@@ -282,20 +287,41 @@ public void notifyLeaderChanged(RaftGroupMemberId groupMemberId,
return;
}
- currentLeaderTerm.set(scm.getScmHAManager().getRatisServer().getDivision()
- .getInfo().getCurrentTerm());
-
- if (!groupMemberId.getPeerId().equals(newLeaderId)) {
- LOG.info("leader changed, yet current SCM is still follower.");
+ 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();
+ LOG.info("Leader changed to {}, current SCM {} is still follower.",
+ newLeaderId, scm.getScmId());
return;
}
- LOG.info("current SCM becomes leader of term {}.", currentLeaderTerm);
+ long currentTerm = scm.getScmHAManager().getRatisServer().getDivision()
+ .getInfo().getCurrentTerm();
+ LOG.info("current SCM {} becomes leader of term {}.", scm.getScmId(), currentTerm);
- 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) {
+ ExitUtils.terminate(1, "Failed to flush transactionBuffer", ex, StateMachine.LOG);
+ }
+
DeletedBlockLog deletedBlockLog = scm.getScmBlockManager()
.getDeletedBlockLog();
Preconditions.checkArgument(
@@ -354,22 +380,97 @@ public void notifyTermIndexUpdated(long term, long index) {
transactionBuffer.updateLatestTrxInfo(TransactionInfo.valueOf(term, index));
}
- if (currentLeaderTerm.get() == term) {
- // Means all transactions before this term have been applied.
- // This means after a restart, all pending transactions have been applied.
- // Perform
- // 1. Refresh Safemode rules state.
- // 2. Start DN Rpc server.
- if (!refreshedAfterLeaderReady.get()) {
- scm.getScmSafeModeManager().refresh();
+ // 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)) {
scm.getDatanodeProtocolServer().start();
+ scm.getScmSafeModeManager().refreshAndValidate();
+ }
+ }
+ }
+
+ /**
+ * @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;
+ }
- refreshedAfterLeaderReady.set(true);
+ 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);
}
- currentLeaderTerm.set(-1L);
+ 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();
+ }
+
@Override
public void notifyLeaderReady() {
if (!isInitialized) {
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 4253e7b1c11..eb40f906274 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
@@ -1530,7 +1530,10 @@ public void start() throws IOException {
}
getBlockProtocolServer().start();
- // If HA is enabled, start datanode protocol server once leader is ready.
+ // 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();
}
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);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@ozone.apache.org
For additional commands, e-mail: commits-help@ozone.apache.org