This is an automated email from the ASF dual-hosted git repository. jojochuang pushed a commit to branch ozone-2.1.2 in repository https://gitbox.apache.org/repos/asf/ozone.git
commit 8a1ae0edbb6a797b9f1d83ad5fc87de4234c1f37 Author: Wei-Chiu Chuang <[email protected]> AuthorDate: Fri Aug 28 17:08:41 2026 -0700 Revert "HDDS-14989. Delay follower SCM DN server start until Ratis log catch-up. (#10617)." This reverts commit 173fea0d5b6eaf2fdd87d066c041b4c4e1c74f44. --- .../container/AbstractContainerReportHandler.java | 26 +- .../apache/hadoop/hdds/scm/ha/SCMStateMachine.java | 153 ++-------- .../hdds/scm/server/StorageContainerManager.java | 5 +- .../TestSCMFollowerCatchupWithContainerReport.java | 336 --------------------- 4 files changed, 31 insertions(+), 489 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 5fd12999074..214731f16d6 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); - updateContainerState(containerId, LifeCycleEvent.FINALIZE); + containerManager.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); - updateContainerState(containerId, LifeCycleEvent.QUASI_CLOSE); + containerManager.updateContainerState(containerId, LifeCycleEvent.QUASI_CLOSE); return false; } @@ -286,7 +286,7 @@ private boolean updateContainerState(final DatanodeDetails datanode, return true; } getLogger().info("CLOSE {}", detailsForLogging); - updateContainerState(containerId, LifeCycleEvent.CLOSE); + containerManager.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); - updateContainerState(containerId, LifeCycleEvent.FORCE_CLOSE); + containerManager.updateContainerState(containerId, LifeCycleEvent.FORCE_CLOSE); } return false; case CLOSED: @@ -330,24 +330,6 @@ 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 a516a217c08..ef9ffc03f26 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,6 +34,7 @@ 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; @@ -84,13 +85,8 @@ public class SCMStateMachine extends BaseStateMachine { private DBCheckpoint installingDBCheckpoint = null; private List<ManagedSecretKey> installingSecretKeys = null; - 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; + private AtomicLong currentLeaderTerm = new AtomicLong(-1L); + private AtomicBoolean refreshedAfterLeaderReady = new AtomicBoolean(); public SCMStateMachine(final StorageContainerManager scm, SCMHADBTransactionBuffer buffer) { @@ -166,15 +162,14 @@ 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); @@ -287,41 +282,20 @@ public void notifyLeaderChanged(RaftGroupMemberId groupMemberId, return; } - 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()); + currentLeaderTerm.set(scm.getScmHAManager().getRatisServer().getDivision() + .getInfo().getCurrentTerm()); + + if (!groupMemberId.getPeerId().equals(newLeaderId)) { + LOG.info("leader changed, yet current SCM is still follower."); return; } - long currentTerm = scm.getScmHAManager().getRatisServer().getDivision() - .getInfo().getCurrentTerm(); - LOG.info("current SCM {} becomes leader of term {}.", scm.getScmId(), currentTerm); + LOG.info("current SCM becomes leader of term {}.", currentLeaderTerm); - scm.getScmContext().updateLeaderAndTerm(true, currentTerm); + scm.getScmContext().updateLeaderAndTerm(true, + currentLeaderTerm.get()); 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( @@ -380,97 +354,22 @@ public void notifyTermIndexUpdated(long term, long index) { transactionBuffer.updateLatestTrxInfo(TransactionInfo.valueOf(term, index)); } - // 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)) { + 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(); 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; - } - 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); + refreshedAfterLeaderReady.set(true); } - return caughtUp; - } catch (Exception e) { - LOG.warn("Failed to check follower catch-up status", e); - return false; + currentLeaderTerm.set(-1L); } } - /** - * @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 eb40f906274..4253e7b1c11 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,10 +1530,7 @@ public void start() throws IOException { } getBlockProtocolServer().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 HA is enabled, start datanode protocol server once leader is ready. 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 deleted file mode 100644 index 1e172d7be5c..00000000000 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestSCMFollowerCatchupWithContainerReport.java +++ /dev/null @@ -1,336 +0,0 @@ -/* - * 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]
