This is an automated email from the ASF dual-hosted git repository.
smengcl 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 bcf9463253e HDDS-16132. Retry follower SCM DN server start after
catch-up (#10999)
bcf9463253e is described below
commit bcf9463253eb504457ee819a5de802f131fb1e6d
Author: Siyao Meng <[email protected]>
AuthorDate: Sat Oct 10 00:41:16 2026 -0700
HDDS-16132. Retry follower SCM DN server start after catch-up (#10999)
Generated-by: Codex (GPT-5.6 Sol)
---
.../apache/hadoop/hdds/scm/ha/SCMStateMachine.java | 129 +++++++++-
.../hdds/scm/server/StorageContainerManager.java | 7 +
.../hadoop/hdds/scm/ha/TestSCMStateMachine.java | 280 +++++++++++++++++++--
.../TestSCMFollowerCatchupWithContainerReport.java | 6 +-
4 files changed, 397 insertions(+), 25 deletions(-)
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 eb1b8ec0577..546f5b63dfe 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
@@ -23,6 +23,7 @@
import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.io.IOException;
+import java.time.Duration;
import java.util.Collection;
import java.util.EnumMap;
import java.util.List;
@@ -30,8 +31,11 @@
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.LongSupplier;
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;
@@ -82,6 +86,24 @@ public class SCMStateMachine extends BaseStateMachine {
private final boolean isInitialized;
private ExecutorService installSnapshotExecutor;
+ // Interval for the fallback retry that starts the deferred datanode-server
on
+ // a follower (see retryStartDNServerUntilReady).
+ private static final Duration DN_SERVER_START_RETRY_INTERVAL =
+ Duration.ofSeconds(1);
+ private static final Duration DN_SERVER_START_INITIAL_WARNING_DELAY =
Duration.ofSeconds(30);
+ private static final Duration DN_SERVER_START_WARNING_INTERVAL =
Duration.ofMinutes(5);
+ private final Duration dnServerStartRetryInterval;
+ private final LongSupplier monotonicClock;
+
+ // Periodically retries the deferred datanode-server start on a follower so
an
+ // idle cluster (no new transactions, only Ratis heartbeats) still exits safe
+ // mode once catch-up becomes observable.
+ private ScheduledExecutorService dnServerStartRetryExecutor;
+ private ScheduledFuture<?> dnServerStartRetryFuture;
+ private long dnServerStartRetryAttempts;
+ private long dnServerStartRetryStartedAt;
+ private long dnServerStartNextWarningAt;
+
private DBCheckpoint installingDBCheckpoint = null;
private List<ManagedSecretKey> installingSecretKeys = null;
@@ -95,10 +117,18 @@ public class SCMStateMachine extends BaseStateMachine {
public SCMStateMachine(final StorageContainerManager scm,
SCMHADBTransactionBuffer buffer) {
+ this(scm, buffer, DN_SERVER_START_RETRY_INTERVAL, Time::monotonicNow);
+ }
+
+ @VisibleForTesting
+ SCMStateMachine(final StorageContainerManager scm,
+ SCMHADBTransactionBuffer buffer, Duration retryInterval, LongSupplier
monotonicClock) {
this.scm = scm;
this.invokers = new EnumMap<>(RequestType.class);
this.transactionBuffer = buffer;
this.metrics = scm.getMetrics();
+ this.dnServerStartRetryInterval = retryInterval;
+ this.monotonicClock = monotonicClock;
TransactionInfo latestTrxInfo = this.transactionBuffer.getLatestTrxInfo();
if (!latestTrxInfo.isDefault()) {
updateLastAppliedTermIndex(latestTrxInfo.getTerm(),
@@ -111,12 +141,21 @@ public SCMStateMachine(final StorageContainerManager scm,
.setNameFormat(scm.threadNamePrefix() + "SCMInstallSnapshot-%d")
.build()
);
+ this.dnServerStartRetryExecutor = HadoopExecutors.newScheduledThreadPool(1,
+ new ThreadFactoryBuilder()
+ .setNameFormat(scm.threadNamePrefix() + "SCMDNServerStartRetry-%d")
+ .setDaemon(true)
+ .build());
isInitialized = true;
}
+ // Used only by SCMRatisServerImpl.initialize() to bootstrap Ratis storage;
+ // there is no SCM instance or background work to start or stop.
public SCMStateMachine() {
isInitialized = false;
this.metrics = null;
+ this.dnServerStartRetryInterval = DN_SERVER_START_RETRY_INTERVAL;
+ this.monotonicClock = Time::monotonicNow;
}
private void addRatisEvent(String message) {
@@ -303,6 +342,7 @@ public void notifyLeaderChanged(RaftGroupMemberId
groupMemberId,
leaderCommitIndexOnStart = getLeaderCommitIndex();
}
tryStartDNServerAndRefreshSafeMode();
+ scheduleDNServerStartRetry();
String message = "Leader changed to " + newLeaderId +
", current SCM " + scm.getScmId() + " is still follower.";
LOG.info(message);
@@ -416,18 +456,102 @@ public void notifyTermIndexUpdated(long term, long
index) {
* 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()) {
+ private synchronized void tryStartDNServerAndRefreshSafeMode() {
+ if (scm.isStopped() || isStateMachineReady.get()) {
return;
}
if (scm.getScmContext().isLeader() || isFollowerCaughtUp()) {
if (isStateMachineReady.compareAndSet(false, true)) {
+ // No further retry is needed. shutdown() lets the current retry, if
+ // this method was called by it, finish normally without cancellation.
+ dnServerStartRetryExecutor.shutdown();
scm.getDatanodeProtocolServer().start();
scm.getScmSafeModeManager().refreshAndValidate();
}
}
}
+ /**
+ * Periodic fallback for the deferred datanode-server start on a follower.
The
+ * Ratis callbacks ({@code applyTransaction}, {@code notifyTermIndexUpdated},
+ * {@code notifyLeaderChanged}) only fire on new activity, so on an idle
+ * cluster a restarted follower can miss the moment the leader's committed
+ * index becomes observable (it is not yet known when {@code
notifyLeaderChanged}
+ * fires) and never start its datanode server, leaving it stuck in safe mode.
+ * This re-checks until the server starts. Starting stays
+ * gated by the same {@link #tryStartDNServerAndRefreshSafeMode()}
predicate, so
+ * it cannot start the server before genuine catch-up.
+ */
+ private void retryStartDNServerUntilReady() {
+ try {
+ synchronized (this) {
+ if (!scm.isStopped() && !isStateMachineReady.get()) {
+ dnServerStartRetryAttempts++;
+ tryStartDNServerAndRefreshSafeMode();
+ logPendingDNServerStart();
+ }
+ }
+ } catch (Exception e) {
+ // Do not let a failed readiness check terminate the fallback. The
+ // finally block only schedules another attempt if startup is pending.
+ LOG.warn("Deferred datanode-server start retry failed", e);
+ } finally {
+ synchronized (this) {
+ dnServerStartRetryFuture = null;
+ scheduleDNServerStartRetry();
+ }
+ }
+ }
+
+ private void logPendingDNServerStart() {
+ if (scm.isStopped() || isStateMachineReady.get()) {
+ return;
+ }
+ long now = monotonicClock.getAsLong();
+ if (now >= dnServerStartNextWarningAt) {
+ long lastAppliedIndex =
scm.getScmHAManager().getRatisServer().getDivision().getInfo().getLastAppliedIndex();
+ LOG.warn("Deferred datanode-server start is still waiting for follower
catch-up: " +
+ "attempts={}, elapsed={}ms, lastAppliedIndex={},
leaderCommitIndexOnStart={}",
+ dnServerStartRetryAttempts, now - dnServerStartRetryStartedAt,
lastAppliedIndex, leaderCommitIndexOnStart);
+ dnServerStartNextWarningAt = now +
DN_SERVER_START_WARNING_INTERVAL.toMillis();
+ }
+ }
+
+ private synchronized void scheduleDNServerStartRetry() {
+ if (scm.isStopped() || isStateMachineReady.get() ||
+ dnServerStartRetryExecutor.isShutdown() || dnServerStartRetryFuture !=
null) {
+ return;
+ }
+ if (dnServerStartRetryAttempts == 0) {
+ dnServerStartRetryStartedAt = monotonicClock.getAsLong();
+ dnServerStartNextWarningAt = dnServerStartRetryStartedAt +
DN_SERVER_START_INITIAL_WARNING_DELAY.toMillis();
+ }
+ dnServerStartRetryFuture = dnServerStartRetryExecutor.schedule(
+ this::retryStartDNServerUntilReady,
+ dnServerStartRetryInterval.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
+ public void stopDNServerStartRetry() {
+ if (!isInitialized) {
+ return;
+ }
+ synchronized (this) {
+ dnServerStartRetryExecutor.shutdownNow();
+ dnServerStartRetryFuture = null;
+ }
+ HadoopExecutors.shutdown(dnServerStartRetryExecutor, LOG, 5,
TimeUnit.SECONDS);
+ }
+
+ @VisibleForTesting
+ boolean isDNServerStartRetryStopped() {
+ return dnServerStartRetryExecutor.isShutdown();
+ }
+
+ @VisibleForTesting
+ boolean isDNServerStartRetryTerminated() {
+ return dnServerStartRetryExecutor.isTerminated();
+ }
+
/**
* @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
@@ -570,6 +694,7 @@ public void close() throws IOException {
transactionBuffer.close();
HadoopExecutors.
shutdown(installSnapshotExecutor, LOG, 5, TimeUnit.SECONDS);
+ stopDNServerStartRetry();
} else if (!scm.isStopped()) {
scm.shutDown("scm statemachine is closed by ratis, terminate SCM");
}
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 0fa0e38bfa7..fa32b0b9705 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
@@ -117,6 +117,7 @@
import org.apache.hadoop.hdds.scm.ha.SCMRatisServerImpl;
import org.apache.hadoop.hdds.scm.ha.SCMServiceException;
import org.apache.hadoop.hdds.scm.ha.SCMServiceManager;
+import org.apache.hadoop.hdds.scm.ha.SCMStateMachine;
import org.apache.hadoop.hdds.scm.ha.SequenceIdGenerator;
import org.apache.hadoop.hdds.scm.ha.StatefulServiceStateManager;
import org.apache.hadoop.hdds.scm.ha.StatefulServiceStateManagerImpl;
@@ -1627,6 +1628,12 @@ public void stop() {
stopReplicationManager(); // started eagerly
return;
}
+
+ SCMStateMachine stateMachine = getScmHAManager().getRatisServer()
+ .getSCMStateMachine();
+ if (stateMachine != null) {
+ stateMachine.stopDNServerStartRetry();
+ }
try {
if (containerBalancer.isBalancerRunning()) {
LOG.info("Stopping Container Balancer service.");
diff --git
a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/ha/TestSCMStateMachine.java
b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/ha/TestSCMStateMachine.java
index 0bf82bad2b5..075e6f10223 100644
---
a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/ha/TestSCMStateMachine.java
+++
b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/ha/TestSCMStateMachine.java
@@ -17,18 +17,30 @@
package org.apache.hadoop.hdds.scm.ha;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.after;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.timeout;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import java.time.Duration;
+import java.util.Collections;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+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.container.placement.metrics.SCMMetrics;
import org.apache.hadoop.hdds.scm.exceptions.SCMException;
@@ -39,37 +51,265 @@
import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
import org.apache.hadoop.hdds.utils.TransactionInfo;
import org.apache.hadoop.ozone.upgrade.UpgradeException;
+import org.apache.hadoop.util.concurrent.ExecutorHelper;
+import org.apache.ozone.test.GenericTestUtils;
+import org.apache.ozone.test.GenericTestUtils.LogCapturer;
import org.apache.ratis.proto.RaftProtos;
import org.apache.ratis.proto.RaftProtos.LogEntryProto;
import org.apache.ratis.proto.RaftProtos.StateMachineLogEntryProto;
import org.apache.ratis.protocol.Message;
+import org.apache.ratis.protocol.RaftGroupId;
+import org.apache.ratis.protocol.RaftGroupMemberId;
+import org.apache.ratis.protocol.RaftPeerId;
+import org.apache.ratis.server.DivisionInfo;
+import org.apache.ratis.server.RaftServer;
import org.apache.ratis.server.protocol.TermIndex;
import org.apache.ratis.statemachine.TransactionContext;
import org.apache.ratis.util.ExitUtils;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.InOrder;
/**
- * Test SCMStateMachine events recording.
+ * Tests SCMStateMachine events and deferred datanode-server startup.
*/
public class TestSCMStateMachine {
+ private static final Duration RETRY_INTERVAL = Duration.ofMillis(100);
- @Test
- public void testRatisEventsRecording() throws Exception {
+ private final AtomicBoolean commitIndexAvailable = new AtomicBoolean();
+ private final AtomicBoolean scmStopped = new AtomicBoolean();
+ private final AtomicLong lastAppliedIndex = new AtomicLong(5L);
+ private final AtomicLong leaderCommitIndex = new AtomicLong(5L);
+ private final AtomicLong retryClock = new AtomicLong(1000L);
+ private final RaftPeerId followerId = RaftPeerId.valueOf("follower");
+ private final RaftPeerId leaderId = RaftPeerId.valueOf("leader");
+
+ private SCMMetrics metrics;
+ private SCMDatanodeProtocolServer datanodeProtocolServer;
+ private SCMSafeModeManager safeModeManager;
+ private SCMStateMachine stateMachine;
+
+ @BeforeEach
+ void setUp() {
StorageContainerManager scm = mock(StorageContainerManager.class);
- SCMMetrics metrics = SCMMetrics.create();
+ metrics = SCMMetrics.create();
+ SCMHADBTransactionBuffer buffer = mock(SCMHADBTransactionBuffer.class);
+ datanodeProtocolServer = mock(SCMDatanodeProtocolServer.class);
+ safeModeManager = mock(SCMSafeModeManager.class);
+
+ SCMContext scmContext = mock(SCMContext.class);
+ SCMHAManager haManager = mock(SCMHAManager.class);
+ SCMRatisServer ratisServer = mock(SCMRatisServer.class);
+ RaftServer.Division division = mock(RaftServer.Division.class);
+ DivisionInfo divisionInfo = mock(DivisionInfo.class);
+
when(scm.getMetrics()).thenReturn(metrics);
+ when(scm.isStopped()).thenAnswer(invocation -> scmStopped.get());
+ when(scm.getScmContext()).thenReturn(scmContext);
+ when(scm.getScmHAManager()).thenReturn(haManager);
+ when(scm.getDatanodeProtocolServer()).thenReturn(datanodeProtocolServer);
+ when(scm.getScmSafeModeManager()).thenReturn(safeModeManager);
+ when(haManager.getRatisServer()).thenReturn(ratisServer);
+ when(ratisServer.getDivision()).thenReturn(division);
+ when(division.getInfo()).thenReturn(divisionInfo);
+ when(divisionInfo.getLeaderId()).thenReturn(leaderId);
+ when(divisionInfo.getLastAppliedIndex()).thenAnswer(invocation ->
lastAppliedIndex.get());
+ when(division.getCommitInfos()).thenAnswer(invocation ->
commitIndexAvailable.get()
+ ? Collections.singletonList(RaftProtos.CommitInfoProto.newBuilder()
+
.setServer(RaftProtos.RaftPeerProto.newBuilder().setId(leaderId.toByteString()))
+ .setCommitIndex(leaderCommitIndex.get())
+ .build())
+ : Collections.emptyList());
+ when(buffer.getLatestTrxInfo()).thenReturn(
+ TransactionInfo.valueOf(TermIndex.valueOf(0, 0)));
- SCMHADBTransactionBuffer buffer = mock(SCMHADBTransactionBuffer.class);
-
when(buffer.getLatestTrxInfo()).thenReturn(TransactionInfo.valueOf(TermIndex.valueOf(0,
0)));
+ stateMachine = new SCMStateMachine(scm, buffer, RETRY_INTERVAL,
retryClock::get);
+ }
- // The contents of the state machine are mocked, so SCMStateMachine#close
is a no-op.
- SCMStateMachine stateMachine = new SCMStateMachine(scm, buffer);
- stateMachine.notifyConfigurationChanged(1, 1,
RaftProtos.RaftConfigurationProto.getDefaultInstance());
+ @AfterEach
+ void tearDown() {
+ stateMachine.stopDNServerStartRetry();
+ metrics.unRegister();
+ }
+ @Test
+ void testRatisEventsRecording() {
+ stateMachine.notifyConfigurationChanged(1, 1,
RaftProtos.RaftConfigurationProto.getDefaultInstance());
assertTrue(metrics.getRatisEvents().contains("Configuration changed at
term index"));
+ }
- metrics.unRegister();
+ @Test
+ void testRetryStartsDNServerWhenLeaderCommitIndexBecomesAvailable() throws
Exception {
+ LogCapturer executorLogs = LogCapturer.captureLogs(ExecutorHelper.class);
+ try {
+ stateMachine.notifyLeaderChanged(memberId(), leaderId);
+ verify(datanodeProtocolServer, never()).start();
+
+ commitIndexAvailable.set(true);
+
+ verify(datanodeProtocolServer, timeout(2000)).start();
+ verify(safeModeManager, timeout(2000)).refreshAndValidate();
+ assertThat(stateMachine.getIsStateMachineReady()).isTrue();
+ assertThat(stateMachine.isDNServerStartRetryStopped()).isTrue();
+ GenericTestUtils.waitFor(stateMachine::isDNServerStartRetryTerminated,
10, 2000);
+
assertThat(executorLogs.getOutput()).doesNotContain("CancellationException");
+ } finally {
+ executorLogs.stopCapturing();
+ }
+ }
+
+ @Test
+ void testRetryWaitsUntilFollowerCatchesUp() {
+ commitIndexAvailable.set(true);
+ lastAppliedIndex.set(4L);
+
+ stateMachine.notifyLeaderChanged(memberId(), leaderId);
+
+ verify(datanodeProtocolServer, after(RETRY_INTERVAL.toMillis() *
3).never()).start();
+ lastAppliedIndex.set(5L);
+ verify(datanodeProtocolServer, timeout(2000)).start();
+ verify(safeModeManager, timeout(2000)).refreshAndValidate();
+ }
+
+ @Test
+ void testRetryUsesCapturedCommitIndexWhenLeaderAdvances() {
+ commitIndexAvailable.set(true);
+ lastAppliedIndex.set(4L);
+ stateMachine.notifyLeaderChanged(memberId(), leaderId);
+
+ leaderCommitIndex.set(100L);
+ lastAppliedIndex.set(5L);
+
+ verify(datanodeProtocolServer, timeout(2000).times(1)).start();
+ verify(safeModeManager, timeout(2000).times(1)).refreshAndValidate();
+ assertThat(stateMachine.getIsStateMachineReady()).isTrue();
+ }
+
+ @Test
+ void testPendingRetryWarningsAreThrottledAndStopWhenReady() throws Exception
{
+ commitIndexAvailable.set(true);
+ lastAppliedIndex.set(4L);
+ LogCapturer logs = LogCapturer.captureLogs(SCMStateMachine.class);
+ try {
+ stateMachine.notifyLeaderChanged(memberId(), leaderId);
+ retryClock.set(30999L);
+ verify(datanodeProtocolServer, after(RETRY_INTERVAL.toMillis() *
2).never()).start();
+ assertThat(logs.getOutput()).doesNotContain("still waiting for follower
catch-up");
+
+ retryClock.set(31000L);
+ GenericTestUtils.waitFor(() ->
logs.getOutput().contains("elapsed=30000ms"), 10, 2000);
+ assertThat(logs.getOutput()).containsPattern(
+ "attempts=[1-9][0-9]*, elapsed=30000ms, lastAppliedIndex=4,
leaderCommitIndexOnStart=5");
+
+ retryClock.set(330999L);
+ verify(datanodeProtocolServer, after(RETRY_INTERVAL.toMillis() *
2).never()).start();
+ assertThat(logs.getOutput()).containsOnlyOnce("still waiting for
follower catch-up");
+
+ retryClock.set(331000L);
+ GenericTestUtils.waitFor(() ->
logs.getOutput().contains("elapsed=330000ms"), 10, 2000);
+ assertThat(logs.getOutput()).containsOnlyOnce("elapsed=330000ms");
+
+ lastAppliedIndex.set(5L);
+ verify(datanodeProtocolServer, timeout(2000)).start();
+ GenericTestUtils.waitFor(stateMachine::isDNServerStartRetryTerminated,
10, 2000);
+ logs.clearOutput();
+ retryClock.set(1000000L);
+ stateMachine.notifyLeaderChanged(memberId(), leaderId);
+ assertThat(logs.getOutput()).doesNotContain("still waiting for follower
catch-up");
+ } finally {
+ logs.stopCapturing();
+ }
+ }
+
+ @Test
+ void testPendingRetryWarningIncludesUnavailableLeaderCommitIndex() throws
Exception {
+ LogCapturer logs = LogCapturer.captureLogs(SCMStateMachine.class);
+ try {
+ stateMachine.notifyLeaderChanged(memberId(), leaderId);
+ retryClock.set(31000L);
+
+ GenericTestUtils.waitFor(() ->
logs.getOutput().contains("elapsed=30000ms"), 10, 2000);
+ assertThat(logs.getOutput()).contains("lastAppliedIndex=5,
leaderCommitIndexOnStart=-1");
+ verify(datanodeProtocolServer, never()).start();
+
+ scmStopped.set(true);
+ stateMachine.stopDNServerStartRetry();
+ logs.clearOutput();
+ retryClock.set(1000000L);
+ assertThat(stateMachine.isDNServerStartRetryTerminated()).isTrue();
+ assertThat(logs.getOutput()).doesNotContain("still waiting for follower
catch-up");
+ } finally {
+ logs.stopCapturing();
+ }
+ }
+
+ @Test
+ void testBootstrapStateMachineSkipsSCMCallbacksAndRetryStop() throws
Exception {
+ try (SCMStateMachine bootstrap = new SCMStateMachine()) {
+ bootstrap.notifyLeaderChanged(memberId(), leaderId);
+ bootstrap.notifyLeaderChanged(memberId(), followerId);
+ bootstrap.notifyTermIndexUpdated(1, 1);
+ bootstrap.notifyLeaderReady();
+ bootstrap.notifyNotLeader(Collections.emptyList());
+ bootstrap.stopDNServerStartRetry();
+ bootstrap.stopDNServerStartRetry();
+
+ assertThat(bootstrap.getIsStateMachineReady()).isFalse();
+ assertThat(bootstrap.getLatestSnapshot()).isNull();
+ }
+ }
+
+ @Test
+ void testAlreadyCaughtUpStartsDNServerExactlyOnce() throws Exception {
+ commitIndexAvailable.set(true);
+
+ stateMachine.notifyLeaderChanged(memberId(), leaderId);
+ stateMachine.notifyLeaderChanged(memberId(), leaderId);
+
+ verify(datanodeProtocolServer, times(1)).start();
+ verify(safeModeManager, times(1)).refreshAndValidate();
+ assertThat(stateMachine.getIsStateMachineReady()).isTrue();
+ GenericTestUtils.waitFor(stateMachine::isDNServerStartRetryTerminated, 10,
2000);
+ }
+
+ @Test
+ void testStopPreventsPendingRetryFromStartingDNServer() {
+ stateMachine.notifyLeaderChanged(memberId(), leaderId);
+ scmStopped.set(true);
+
+ stateMachine.stopDNServerStartRetry();
+
+ assertThat(stateMachine.isDNServerStartRetryStopped()).isTrue();
+ commitIndexAvailable.set(true);
+ verify(datanodeProtocolServer, after(RETRY_INTERVAL.toMillis() *
2).never()).start();
+ }
+
+ @Test
+ void testStopWaitsForRunningRetry() throws Exception {
+ CountDownLatch startEntered = new CountDownLatch(1);
+ CountDownLatch allowStart = new CountDownLatch(1);
+ doAnswer(invocation -> {
+ startEntered.countDown();
+ assertTrue(allowStart.await(2, TimeUnit.SECONDS));
+ return null;
+ }).when(datanodeProtocolServer).start();
+ stateMachine.notifyLeaderChanged(memberId(), leaderId);
+ commitIndexAvailable.set(true);
+ assertTrue(startEntered.await(2, TimeUnit.SECONDS));
+
+ scmStopped.set(true);
+ CompletableFuture<Void> stopFuture =
CompletableFuture.runAsync(stateMachine::stopDNServerStartRetry);
+ assertThrows(TimeoutException.class, () -> stopFuture.get(100,
TimeUnit.MILLISECONDS));
+
+ allowStart.countDown();
+ stopFuture.get(2, TimeUnit.SECONDS);
+ assertThat(stateMachine.isDNServerStartRetryTerminated()).isTrue();
+ verify(datanodeProtocolServer, times(1)).start();
+ }
+
+ private RaftGroupMemberId memberId() {
+ return RaftGroupMemberId.valueOf(followerId, RaftGroupId.randomId());
}
/**
@@ -85,11 +325,11 @@ public void testUpgradeExceptionDuringApplyTerminates()
throws Exception {
ScmInvoker<?> invoker = mock(ScmInvoker.class);
when(invoker.invokeLocal(any(), any())).thenThrow(
new
UpgradeException(UpgradeException.ResultCodes.FINALIZE_UPGRADE_ACTION_FAILED));
- SCMStateMachine stateMachine = newStateMachine(buffer,
RequestType.FINALIZE, invoker);
+ SCMStateMachine applyingStateMachine = newStateMachine(buffer,
RequestType.FINALIZE, invoker);
// terminate throws ExitException when system exit is disabled
assertThrows(ExitUtils.ExitException.class,
- () ->
stateMachine.applyTransaction(newTransaction(RequestType.FINALIZE, 1)));
+ () ->
applyingStateMachine.applyTransaction(newTransaction(RequestType.FINALIZE, 1)));
} finally {
ExitUtils.clear();
}
@@ -98,9 +338,9 @@ public void testUpgradeExceptionDuringApplyTerminates()
throws Exception {
@Test
public void testApplyTransactionFlushesAfterRecordingTransactionInfo()
throws Exception {
SCMHADBTransactionBuffer buffer = mock(SCMHADBTransactionBuffer.class);
- SCMStateMachine stateMachine = newStateMachine(buffer,
RequestType.PIPELINE, succeedingInvoker());
+ SCMStateMachine applyingStateMachine = newStateMachine(buffer,
RequestType.PIPELINE, succeedingInvoker());
- CompletableFuture<Message> result =
stateMachine.applyTransaction(newTransaction(RequestType.PIPELINE, 7));
+ CompletableFuture<Message> result =
applyingStateMachine.applyTransaction(newTransaction(RequestType.PIPELINE, 7));
assertTrue(result.isDone() && !result.isCompletedExceptionally());
InOrder order = inOrder(buffer);
@@ -113,10 +353,10 @@ public void
testApplyTransactionFlushesAfterRecordingTransactionInfo() throws Ex
@Test
public void testFlushCheckedForEveryAppliedTransaction() throws Exception {
SCMHADBTransactionBuffer buffer = mock(SCMHADBTransactionBuffer.class);
- SCMStateMachine stateMachine = newStateMachine(buffer,
RequestType.PIPELINE, succeedingInvoker());
+ SCMStateMachine applyingStateMachine = newStateMachine(buffer,
RequestType.PIPELINE, succeedingInvoker());
for (int i = 1; i <= 5; i++) {
- stateMachine.applyTransaction(newTransaction(RequestType.PIPELINE, i));
+
applyingStateMachine.applyTransaction(newTransaction(RequestType.PIPELINE, i));
}
verify(buffer, times(5)).flushIfPendingLimitReached();
@@ -129,9 +369,9 @@ public void
testFlushStillCheckedWhenTransactionIsLogicallyRejected() throws Exc
ScmInvoker<?> invoker = mock(ScmInvoker.class);
when(invoker.invokeLocal(anyString(), any()))
.thenThrow(new SCMException("rejected",
ResultCodes.FAILED_TO_FIND_CONTAINER));
- SCMStateMachine stateMachine = newStateMachine(buffer,
RequestType.PIPELINE, invoker);
+ SCMStateMachine applyingStateMachine = newStateMachine(buffer,
RequestType.PIPELINE, invoker);
- CompletableFuture<Message> result =
stateMachine.applyTransaction(newTransaction(RequestType.PIPELINE, 3));
+ CompletableFuture<Message> result =
applyingStateMachine.applyTransaction(newTransaction(RequestType.PIPELINE, 3));
assertTrue(result.isCompletedExceptionally());
InOrder order = inOrder(buffer);
@@ -146,10 +386,10 @@ public void
testFlushFailureTerminatesAndClosesApplyingWindow() throws Exception
try {
SCMHADBTransactionBuffer buffer = mock(SCMHADBTransactionBuffer.class);
doThrow(new IllegalStateException("injected flush
failure")).when(buffer).flushIfPendingLimitReached();
- SCMStateMachine stateMachine = newStateMachine(buffer,
RequestType.PIPELINE, succeedingInvoker());
+ SCMStateMachine applyingStateMachine = newStateMachine(buffer,
RequestType.PIPELINE, succeedingInvoker());
assertThrows(ExitUtils.ExitException.class,
- () ->
stateMachine.applyTransaction(newTransaction(RequestType.PIPELINE, 9)));
+ () ->
applyingStateMachine.applyTransaction(newTransaction(RequestType.PIPELINE, 9)));
verify(buffer).endApplyingTransaction();
} finally {
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
index 8ab1f6ae8ed..faf602ee3e8 100644
---
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
@@ -223,10 +223,10 @@ void testFollowerCatchupAfterContainerCreate() throws
Exception {
}
/**
- * Edge case for removing the background polling loop: on an otherwise idle
+ * Edge case for the fallback retry: 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.
+ * mode) and serve replicas after promotion without a steady stream of new
+ * transactions.
*/
@Test
void testFollowerCatchupOnIdleCluster() throws Exception {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
