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 <[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]> (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: [email protected] For additional commands, e-mail: [email protected]
