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 cbe95512541 HDDS-16275. OM checkpoint pause() stops flush thread but
not apply executor. (#11190)
cbe95512541 is described below
commit cbe95512541bd65be925c60168d05e14873ac24b
Author: Sadanand Shenoy <[email protected]>
AuthorDate: Mon Oct 12 10:01:29 2026 +0530
HDDS-16275. OM checkpoint pause() stops flush thread but not apply
executor. (#11190)
---
.../hadoop/ozone/om/TestOMRatisSnapshots.java | 107 +++++++++++
.../TestDirectoryDeletingServiceWithFSO.java | 18 +-
.../org/apache/hadoop/ozone/om/OzoneManager.java | 14 +-
.../ozone/om/ratis/OzoneManagerStateMachine.java | 182 +++++++++++++++---
.../om/ratis/TestOzoneManagerStateMachine.java | 208 +++++++++++++++++++++
5 files changed, 487 insertions(+), 42 deletions(-)
diff --git
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOMRatisSnapshots.java
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOMRatisSnapshots.java
index 1bfab3c3c6b..fbddb8101fd 100644
---
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOMRatisSnapshots.java
+++
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOMRatisSnapshots.java
@@ -37,8 +37,11 @@
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import java.util.Set;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
@@ -86,6 +89,7 @@
import org.rocksdb.RocksDB;
import org.rocksdb.RocksDBException;
import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.slf4j.event.Level;
/**
@@ -644,6 +648,82 @@ public void testInstallSnapshotFailedBackupRestoresDbDir()
throws Exception {
}
}
+ /**
+ * An apply offered while installCheckpoint has the state machine paused is
rejected, and Ratis moves
+ * past it. If the install then fails, the old DB is kept and Ratis does not
reload the state machine,
+ * so the entry would never be applied: the OM must exit instead of resuming
applies, so that a restart
+ * replays it from the Raft log. A failed install with no such entry must
not exit.
+ *
+ * The install is called directly on a follower that keeps receiving
entries, which is wider than a
+ * Ratis driven install: there the follower refuses appendEntries, so only
entries already committed in
+ * its log but not yet applied can be rejected.
+ */
+ @Test
+ public void testFailedInstallExitsAfterApplyRejectedDuringPause() throws
Exception {
+ final String leaderOMNodeId =
OmTestUtil.getCurrentOmProxyNodeId(objectStore);
+ OzoneManager leaderOM = cluster.getOzoneManager(leaderOMNodeId);
+ // The active follower, so it keeps receiving entries while it installs.
+ String followerNodeId = leaderOM.getPeerNodes().get(0).getNodeId();
+ if (!cluster.isOMActive(followerNodeId)) {
+ followerNodeId = leaderOM.getPeerNodes().get(1).getNodeId();
+ }
+ OzoneManager followerOM = cluster.getOzoneManager(followerNodeId);
+
+ writeKeys(1);
+ DBCheckpoint leaderDbCheckpoint =
leaderOM.getMetadataManager().getStore().getCheckpoint(true);
+ Path leaderCheckpointLocation = leaderDbCheckpoint.getCheckpointLocation();
+ assertNotNull(leaderCheckpointLocation);
+ Path omDbDir = leaderCheckpointLocation.resolve(OM_DB_NAME);
+ Files.createDirectory(omDbDir);
+ moveCheckpointContentsToOmDbDir(leaderCheckpointLocation, omDbDir);
+ TransactionInfo trxnInfo =
OzoneManagerRatisUtils.getTrxnInfoFromCheckpoint(conf, omDbDir);
+ // The follower is caught up, so claim a newer checkpoint to let the
replacement start. It fails
+ // at the first backup move, before anything from the checkpoint is
installed.
+ TransactionInfo aheadTrxnInfo =
TransactionInfo.valueOf(trxnInfo.getTerm(), trxnInfo.getTransactionIndex() +
1000);
+
+ DummyExitManager exitManager = new DummyExitManager();
+ followerOM.setExitManagerForTesting(exitManager);
+ LogCapturer updaterLog =
+
LogCapturer.captureLogs(LoggerFactory.getLogger("org.apache.ratis.server.impl.StateMachineUpdater"));
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ BlockThenThrowFaultInjector injector = new BlockThenThrowFaultInjector();
+ try {
+ // Nothing left to apply on the follower, so this install fails with no
entry rejected.
+ long leaderIndex =
leaderOM.getOmRatisServer().getLastAppliedTermIndex().getIndex();
+ GenericTestUtils.waitFor(
+ () ->
followerOM.getOmRatisServer().getLastAppliedTermIndex().getIndex() >=
leaderIndex, 100, 30_000);
+ followerOM.setCheckpointBackupInjector(new
ThrowOnNthPauseFaultInjector(1, "Simulated backup move failure"));
+ assertNull(followerOM.installCheckpoint(leaderOMNodeId,
leaderCheckpointLocation, aheadTrxnInfo));
+ assertThat(exitManager.messages).isEmpty();
+
+ followerOM.setCheckpointBackupInjector(injector);
+ Future<TermIndex> install = executor.submit(
+ () -> followerOM.installCheckpoint(leaderOMNodeId,
leaderCheckpointLocation, aheadTrxnInfo));
+ assertTrue(injector.paused.await(60, TimeUnit.SECONDS));
+ String key = createKey(ozoneBucket);
+ OmKeyArgs keyArgs = new
OmKeyArgs.Builder().setVolumeName(volumeName).setBucketName(bucketName)
+ .setKeyName(key).build();
+ long keyIndex = leaderOM.lookupKey(keyArgs).getUpdateID();
+ // Ratis logs the rejected apply of the key's commit on the follower.
+ String follower = followerNodeId + "@";
+ String index = "applying txn index=" + keyIndex + ",";
+ GenericTestUtils.waitFor(() ->
Arrays.stream(updaterLog.getOutput().split("\n"))
+ .anyMatch(line -> line.contains(follower) && line.contains(index)),
100, 30_000);
+ injector.resume();
+ assertNull(install.get(60, TimeUnit.SECONDS));
+
+ assertThat(exitManager.messages)
+ .as("follower resumed applies on its old DB after the entry at index
%d was skipped", keyIndex)
+ .singleElement().asString().contains("rejected while the state
machine was paused");
+ } finally {
+ injector.resume();
+ executor.shutdownNow();
+ updaterLog.stopCapturing();
+ followerOM.setCheckpointBackupInjector(null);
+ cluster.setupExitManagerForTesting();
+ }
+ }
+
/** Top-level entries of the OM metadata dir, ignoring backup dirs the
install creates. */
private static Set<String> topLevelNames(File metaDir) throws IOException {
try (Stream<Path> list = Files.list(metaDir.toPath())) {
@@ -870,13 +950,40 @@ private void assertLogCapture(LogCapturer logCapture,
}
private static class DummyExitManager extends ExitManager {
+ private final List<String> messages = new CopyOnWriteArrayList<>();
+
@Override
public void exitSystem(int status, String message, Throwable throwable,
Logger log) {
+ messages.add(throwable == null ? message : message + " " +
throwable.getMessage());
log.error("System Exit: " + message, throwable);
}
}
+ /**
+ * FaultInjector that blocks pause() until resume(), then throws IOException.
+ */
+ private static class BlockThenThrowFaultInjector extends FaultInjector {
+ private final CountDownLatch paused = new CountDownLatch(1);
+ private final CountDownLatch released = new CountDownLatch(1);
+
+ @Override
+ public void pause() throws IOException {
+ paused.countDown();
+ try {
+ released.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ throw new IOException("Simulated backup move failure");
+ }
+
+ @Override
+ public void resume() {
+ released.countDown();
+ }
+ }
+
/**
* FaultInjector that throws IOException on pause(), simulating a download
failure
* after the first part completes. Used to test cleanup on failed download.
diff --git
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/service/TestDirectoryDeletingServiceWithFSO.java
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/service/TestDirectoryDeletingServiceWithFSO.java
index 58d816fbabd..7e92e3bc26a 100644
---
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/service/TestDirectoryDeletingServiceWithFSO.java
+++
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/service/TestDirectoryDeletingServiceWithFSO.java
@@ -38,7 +38,6 @@
import java.io.IOException;
import java.util.List;
import java.util.Random;
-import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
@@ -465,11 +464,11 @@ public void testDeleteWithMultiLevelsBlockDoubleBuffer()
throws Exception {
assertTableRowCount(dirTable, 5);
assertTableRowCount(keyTable, 3);
- // stop daemon to simulate blocked double buffer
+ // Pause flushing to simulate a blocked double buffer
OzoneManagerStateMachine omStateMachine =
cluster.getOzoneManager().getOmRatisServer().getOmStateMachine();
OzoneManagerDoubleBuffer omDoubleBuffer =
omStateMachine.getOzoneManagerDoubleBuffer();
omDoubleBuffer.awaitFlush();
- omDoubleBuffer.stopDaemon();
+ omDoubleBuffer.pause();
OzoneVolume volume = client.getObjectStore().getVolume(volumeName);
OzoneBucket bucket = volume.getBucket(bucketName);
@@ -497,18 +496,9 @@ public void testDeleteWithMultiLevelsBlockDoubleBuffer()
throws Exception {
// verify bucket used namespace is not going -ve
assertTrue(volume.getBucket(bucketName).getUsedNamespace() >= 0);
- // re-init double buffer in state machine to resume further processing
- omStateMachine.pause();
- omStateMachine.unpause(omStateMachine.getLastAppliedTermIndex().getIndex(),
- omStateMachine.getLatestSnapshot().getIndex());
- // flush previous pending transaction manually
- omDoubleBuffer.resume();
- CompletableFuture.supplyAsync(() -> {
- omDoubleBuffer.flushTransactions();
- return null;
- });
+ // Resume flushing the pending transactions.
+ omDoubleBuffer.unpause();
omDoubleBuffer.awaitFlush();
- omDoubleBuffer.stop();
// verify deletion progress completion
assertTableRowCount(deletedDirTable, 0);
assertTableRowCount(keyTable, 0);
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java
index 45d4129dcba..5c3e54ebbae 100644
---
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java
@@ -4672,6 +4672,10 @@ public synchronized TermIndex
installSnapshotFromLeader(String leaderId) throws
if (checkpointLocation == null) {
throw new IOException("Cannot install checkpoint from leader " +
leaderId + ": checkpointLocation is null");
}
+ // The follower refuses appendEntries while this install is in progress,
so its commit index is fixed.
+ // Apply up to it before installCheckpoint pauses the state machine, so
that no committed entry is rejected.
+ long committedIndex =
omRatisServer.getServerDivision().getRaftLog().getLastCommittedIndex();
+ omRatisServer.getOmStateMachine().awaitApplied(committedIndex);
termIndex = installCheckpoint(leaderId, checkpointLocation);
} catch (Exception ex) {
LOG.error("Failed to install snapshot from Leader OM.", ex);
@@ -4737,9 +4741,7 @@ TermIndex installCheckpoint(String leaderId, Path
checkpointLocation,
stopSecretManager();
stopTrashEmptier();
omSnapshotManager.invalidateCache();
- // Pause the State Machine so that no new transactions can be applied.
- // This action also clears the OM Double Buffer so that if there are any
- // pending transactions in the buffer, they are discarded.
+ // Pause the State Machine. It flushes the transactions already applied
and rejects new ones until unpause.
omRatisServer.getOmStateMachine().pause();
} catch (Exception e) {
LOG.error("Failed to stop/ pause the services. Cannot proceed with " +
@@ -4809,6 +4811,8 @@ TermIndex installCheckpoint(String leaderId, Path
checkpointLocation,
omSnapshotManager.close();
}
+ // Keeping the old DB returns null below, so Ratis does not reload the
state machine.
+ boolean oldDbKept = lastAppliedIndex !=
checkpointTrxnInfo.getTransactionIndex();
// Reload the OM DB store with the new checkpoint.
// Restart (unpause) the state machine and update its last applied index
// to the installed checkpoint's snapshot index.
@@ -4820,7 +4824,7 @@ TermIndex installCheckpoint(String leaderId, Path
checkpointLocation,
// obtains the last entry. Publish the position used to unpause. After
a failed
// DB replacement, these values still identify the restored
pre-install state.
setTransactionInfo(TransactionInfo.valueOf(term, lastAppliedIndex));
- omRatisServer.getOmStateMachine().unpause(lastAppliedIndex, term);
+ omRatisServer.getOmStateMachine().unpause(lastAppliedIndex, term,
oldDbKept);
newMetadataManagerStarted = true;
LOG.info("Reloaded OM state with Term: {} and Index: {}. Spend {} ms",
term, lastAppliedIndex, Time.monotonicNow() - time);
@@ -4829,7 +4833,7 @@ TermIndex installCheckpoint(String leaderId, Path
checkpointLocation,
keyManager.start(configuration);
startSecretManagerIfNecessary();
startTrashEmptier(configuration);
- omRatisServer.getOmStateMachine().unpause(lastAppliedIndex, term);
+ omRatisServer.getOmStateMachine().unpause(lastAppliedIndex, term,
oldDbKept);
LOG.info("OM DB is not stopped. Started services with Term: {} and " +
"Index: {}", term, lastAppliedIndex);
}
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerStateMachine.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerStateMachine.java
index 4abdc1cc683..cb4151a41a4 100644
---
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerStateMachine.java
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerStateMachine.java
@@ -98,6 +98,11 @@ public class OzoneManagerStateMachine extends
BaseStateMachine {
private final ExecutorService installSnapshotExecutor;
private final boolean isTracingEnabled;
private final AtomicInteger statePausedCount = new AtomicInteger(0);
+ private final Object applyTransactionMonitor = new Object();
+ private volatile boolean applyTransactionPaused = false;
+ private int inFlightApplyTransactions = 0;
+ // Highest index whose apply was rejected while paused. Reset when a pause
starts (not by a nested pause).
+ private long lastRejectedApplyIndex = RaftLog.INVALID_LOG_INDEX;
private final String threadPrefix;
/**
@@ -434,12 +439,37 @@ public CompletableFuture<Message>
applyTransaction(TransactionContext trx) {
// lastAppliedIndex in OzoneManager StateMachine, even if other
// executor has completed the transactions with id more.
- //if there are too many pending requests, wait for doubleBuffer flushing
- ozoneManagerDoubleBuffer.acquireUnFlushedTransactions(1);
+ if (!tryEnterApplyTransaction(termIndex.getIndex())) {
+ return completeExceptionally(new IOException("Cannot apply transaction
while state machine is paused."));
+ }
+ try {
+ //if there are too many pending requests, wait for doubleBuffer
flushing
+ ozoneManagerDoubleBuffer.acquireUnFlushedTransactions(1);
+ } catch (Exception ex) {
+ exitApplyTransaction();
+ throw ex;
+ }
+
+ final CompletableFuture<OMResponse> runFuture;
+ try {
+ runFuture = CompletableFuture.supplyAsync(() -> {
+ try {
+ return runCommand(request, termIndex);
+ } finally {
+ exitApplyTransaction();
+ }
+ }, executorService);
+ } catch (RuntimeException ex) {
+ ozoneManagerDoubleBuffer.releaseUnFlushedTransactions(1);
+ exitApplyTransaction();
+ throw ex;
+ }
- return CompletableFuture.supplyAsync(() -> runCommand(request,
termIndex), executorService)
- .thenApply(this::processResponse);
+ return runFuture.thenApply(this::processResponse);
} catch (Exception e) {
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
return completeExceptionally(e);
}
}
@@ -485,19 +515,37 @@ public CompletableFuture<Message> query(Message request) {
}
@Override
- public synchronized void pause() {
- LOG.info("OzoneManagerStateMachine is pausing");
- statePausedCount.incrementAndGet();
- final LifeCycle.State state = getLifeCycleState();
- if (state == LifeCycle.State.PAUSED) {
- return;
- }
- if (state != LifeCycle.State.NEW) {
- getLifeCycle().transition(LifeCycle.State.PAUSING);
- getLifeCycle().transition(LifeCycle.State.PAUSED);
+ public void pause() {
+ synchronized (this) {
+ LOG.info("OzoneManagerStateMachine is pausing");
+ statePausedCount.incrementAndGet();
+ final LifeCycle.State state = getLifeCycleState();
+ if (state == LifeCycle.State.PAUSED) {
+ return;
+ }
+ if (state != LifeCycle.State.NEW) {
+ getLifeCycle().transition(LifeCycle.State.PAUSING);
+ getLifeCycle().transition(LifeCycle.State.PAUSED);
+ }
}
- ozoneManagerDoubleBuffer.stop();
+ pauseApplyTransaction();
+ boolean interrupted = false;
+ while (true) {
+ try {
+ ozoneManagerDoubleBuffer.awaitFlush();
+ break;
+ } catch (InterruptedException ex) {
+ interrupted = true;
+ LOG.warn("Interrupted while waiting for OM double buffer to flush
during pause.", ex);
+ }
+ }
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ synchronized (this) {
+ ozoneManagerDoubleBuffer.stop();
+ }
}
/**
@@ -507,15 +555,28 @@ public synchronized void pause() {
*/
public synchronized void unpause(long newLastAppliedSnaphsotIndex,
long newLastAppliedSnapShotTermIndex) {
+ unpause(newLastAppliedSnaphsotIndex, newLastAppliedSnapShotTermIndex,
false);
+ }
+
+ /**
+ * Same as {@link #unpause(long, long)}. If oldDbKept, Ratis does not reload
the state machine, so it never
+ * retries an apply rejected while paused: if one is above the restored
index, keep rejecting applies and throw.
+ */
+ public synchronized void unpause(long newLastAppliedSnaphsotIndex,
+ long newLastAppliedSnapShotTermIndex, boolean oldDbKept) {
if (statePausedCount.decrementAndGet() == 0) {
- getLifeCycle().startAndTransition(() -> {
- this.ozoneManagerDoubleBuffer = buildDoubleBufferForRatis();
- restoreLastAppliedTermIndex(TermIndex.valueOf(
- newLastAppliedSnapShotTermIndex, newLastAppliedSnaphsotIndex));
- LOG.info("{}: OzoneManagerStateMachine un-pause completed. " +
- "newLastAppliedSnapshotIndex: {}, newLastAppliedSnapShotTermIndex:
{}",
- getId(), newLastAppliedSnaphsotIndex,
newLastAppliedSnapShotTermIndex);
- });
+ try {
+ getLifeCycle().startAndTransition(() -> {
+ this.ozoneManagerDoubleBuffer = buildDoubleBufferForRatis();
+ restoreLastAppliedTermIndex(TermIndex.valueOf(
+ newLastAppliedSnapShotTermIndex, newLastAppliedSnaphsotIndex));
+ LOG.info("{}: OzoneManagerStateMachine un-pause completed. " +
+ "newLastAppliedSnapshotIndex: {},
newLastAppliedSnapShotTermIndex: {}",
+ getId(), newLastAppliedSnaphsotIndex,
newLastAppliedSnapShotTermIndex);
+ });
+ } finally {
+ resumeApplyTransaction(oldDbKept ? newLastAppliedSnaphsotIndex :
Long.MAX_VALUE);
+ }
}
}
@@ -729,6 +790,7 @@ public OzoneManagerRequestHandler getHandler() {
}
public void stop() {
+ resumeApplyTransaction();
ozoneManagerDoubleBuffer.stop();
HadoopExecutors.shutdown(executorService, LOG, 5, TimeUnit.SECONDS);
HadoopExecutors.shutdown(installSnapshotExecutor, LOG, 5,
TimeUnit.SECONDS);
@@ -746,8 +808,82 @@ public void awaitDoubleBufferFlush() throws
InterruptedException {
ozoneManagerDoubleBuffer.awaitFlush();
}
+ /** Wait until the state machine has applied and flushed up to the given
index. */
+ public void awaitApplied(long index) throws IOException {
+ while (getLastAppliedTermIndex().getIndex() < index) {
+ if (ozoneManager.isStopped()) {
+ throw new IOException("OzoneManager is already stopped: " +
ozoneManager.getNodeDetails());
+ }
+ try {
+ ozoneManagerDoubleBuffer.awaitFlush();
+ } catch (InterruptedException e) {
+ throw IOUtils.toInterruptedIOException("Interrupted
ozoneManagerDoubleBuffer.awaitFlush", e);
+ }
+ }
+ }
+
@VisibleForTesting
public OzoneManagerDoubleBuffer getOzoneManagerDoubleBuffer() {
return ozoneManagerDoubleBuffer;
}
+
+ private void pauseApplyTransaction() {
+ synchronized (applyTransactionMonitor) {
+ if (applyTransactionPaused) {
+ return;
+ }
+ applyTransactionPaused = true;
+ lastRejectedApplyIndex = RaftLog.INVALID_LOG_INDEX;
+ boolean interrupted = false;
+ while (inFlightApplyTransactions > 0) {
+ try {
+ applyTransactionMonitor.wait();
+ } catch (InterruptedException ex) {
+ LOG.warn("Interrupted while waiting for in-flight apply transactions
to complete.");
+ interrupted = true;
+ }
+ }
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+
+ private void resumeApplyTransaction() {
+ resumeApplyTransaction(Long.MAX_VALUE);
+ }
+
+ private void resumeApplyTransaction(long restoredIndex) {
+ synchronized (applyTransactionMonitor) {
+ if (!applyTransactionPaused) {
+ return;
+ }
+ if (lastRejectedApplyIndex > restoredIndex) {
+ throw new IllegalStateException("Transactions up to index " +
lastRejectedApplyIndex
+ + " were rejected while the state machine was paused, above the
restored index " + restoredIndex);
+ }
+ applyTransactionPaused = false;
+ applyTransactionMonitor.notifyAll();
+ }
+ }
+
+ private boolean tryEnterApplyTransaction(long index) {
+ synchronized (applyTransactionMonitor) {
+ if (applyTransactionPaused) {
+ lastRejectedApplyIndex = Math.max(lastRejectedApplyIndex, index);
+ return false;
+ }
+ inFlightApplyTransactions++;
+ return true;
+ }
+ }
+
+ private void exitApplyTransaction() {
+ synchronized (applyTransactionMonitor) {
+ inFlightApplyTransactions--;
+ if (inFlightApplyTransactions == 0) {
+ applyTransactionMonitor.notifyAll();
+ }
+ }
+ }
}
diff --git
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOzoneManagerStateMachine.java
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOzoneManagerStateMachine.java
index 7a0657828af..ed94136cf49 100644
---
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOzoneManagerStateMachine.java
+++
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOzoneManagerStateMachine.java
@@ -20,6 +20,8 @@
import static
org.apache.hadoop.ozone.om.lock.OzoneManagerLock.LeveledResource.VOLUME_LOCK;
import static
org.apache.hadoop.security.UserGroupInformation.AuthenticationMethod.SIMPLE;
import static org.apache.ozone.test.GenericTestUtils.waitFor;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
@@ -52,6 +54,7 @@
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.LockSupport;
@@ -271,6 +274,133 @@ public void testApplyTransactionBackpressureInterrupt()
throws Exception {
assertInstanceOf(InterruptedException.class, ex.getCause());
}
+ @Test
+ public void testPauseRejectsApplyUntilUnpause() throws Exception {
+ OzoneManagerStateMachine testSm = new OzoneManagerStateMachine(
+ om, doubleBuffer, handler, executor, null) {
+ @Override
+ public OzoneManagerDoubleBuffer buildDoubleBufferForRatis() {
+ return doubleBuffer;
+ }
+ };
+ try {
+ OMRequest request = sampleWriteRequest();
+ TransactionContext trx = mockTrx(request, 1, 5);
+
+ OMResponse expectedResponse = OMResponse.newBuilder()
+ .setCmdType(Type.CreateKey)
+ .setStatus(Status.OK)
+ .setSuccess(true)
+ .build();
+ OMClientResponse clientResponse = mock(OMClientResponse.class);
+ when(clientResponse.getOMResponse()).thenReturn(expectedResponse);
+ when(clientResponse.getOmLockDetails()).thenReturn(null);
+
+ CountDownLatch handlerInvoked = new CountDownLatch(1);
+ doAnswer(invocation -> {
+ handlerInvoked.countDown();
+ return clientResponse;
+ }).when(handler).handleWriteRequest(eq(request), any(),
eq(doubleBuffer));
+
+ testSm.pause();
+ CompletableFuture<Message> pausedFuture = testSm.applyTransaction(trx);
+ ExecutionException ex = assertThrows(ExecutionException.class,
pausedFuture::get);
+ assertInstanceOf(IOException.class, ex.getCause());
+ assertFalse(handlerInvoked.await(200, TimeUnit.MILLISECONDS),
+ "apply should be rejected while state machine is paused");
+
+ testSm.unpause(5, 1);
+
+ CompletableFuture<Message> future = testSm.applyTransaction(trx);
+ assertTrue(handlerInvoked.await(2, TimeUnit.SECONDS),
+ "apply should continue after state machine unpause");
+ assertNotNull(future.get(2, TimeUnit.SECONDS));
+ } finally {
+ testSm.stop();
+ }
+ }
+
+ @Test
+ public void testApplyTransactionExecutorRejectDoesNotBlockPause() throws
Exception {
+ ExecutorService rejectedExecutor = Executors.newSingleThreadExecutor();
+ rejectedExecutor.shutdownNow();
+ OzoneManagerStateMachine testSm = new OzoneManagerStateMachine(
+ om, doubleBuffer, handler, rejectedExecutor, null);
+ try {
+ OMRequest request = sampleWriteRequest();
+ TransactionContext trx = mockTrx(request, 1, 5);
+
+ CompletableFuture<Message> future = testSm.applyTransaction(trx);
+ ExecutionException ex = assertThrows(ExecutionException.class,
future::get);
+ assertInstanceOf(RejectedExecutionException.class, ex.getCause());
+ verify(doubleBuffer).releaseUnFlushedTransactions(1);
+
+ ExecutorService pauseExecutor = Executors.newSingleThreadExecutor();
+ try {
+ Future<?> pauseFuture = pauseExecutor.submit(testSm::pause);
+ pauseFuture.get(2, TimeUnit.SECONDS);
+ } finally {
+ pauseExecutor.shutdownNow();
+ }
+ } finally {
+ testSm.stop();
+ }
+ }
+
+ @Test
+ public void testUnpauseKeepingOldDbFailsWhenApplyAboveItWasRejected() throws
Exception {
+ OzoneManagerStateMachine testSm = new OzoneManagerStateMachine(
+ om, doubleBuffer, handler, executor, null) {
+ @Override
+ public OzoneManagerDoubleBuffer buildDoubleBufferForRatis() {
+ return doubleBuffer;
+ }
+ };
+ try {
+ OMRequest request = sampleWriteRequest();
+ // A rejected entry at or below the restored index is in the kept DB, so
applies resume.
+ testSm.pause();
+ assertRejectedWhilePaused(testSm.applyTransaction(mockTrx(request, 1,
5)));
+ testSm.unpause(5, 1, true);
+
+ // Ratis never retries an entry above the restored index, so applies
must stay rejected.
+ testSm.pause();
+ assertRejectedWhilePaused(testSm.applyTransaction(mockTrx(request, 1,
7)));
+ assertThatThrownBy(() -> testSm.unpause(6, 1, true))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("up to index 7");
+ assertRejectedWhilePaused(testSm.applyTransaction(mockTrx(request, 1,
8)));
+ } finally {
+ testSm.stop();
+ }
+ }
+
+ private static void assertRejectedWhilePaused(CompletableFuture<Message>
future) {
+ ExecutionException ex = assertThrows(ExecutionException.class,
future::get);
+
assertThat(ex.getCause()).isInstanceOf(IOException.class).hasMessageContaining("while
state machine is paused");
+ }
+
+ @Test
+ public void testUnpauseChecksAppliesRejectedWhileRebuildingDoubleBuffer() {
+ OzoneManagerStateMachine testSm = new OzoneManagerStateMachine(
+ om, doubleBuffer, handler, executor, null) {
+ @Override
+ public OzoneManagerDoubleBuffer buildDoubleBufferForRatis() {
+
assertRejectedWhilePaused(applyTransaction(mockTrx(sampleWriteRequest(), 1,
7)));
+ return doubleBuffer;
+ }
+ };
+ try {
+ testSm.pause();
+ assertThatThrownBy(() -> testSm.unpause(6, 1, true))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("up to index 7");
+
assertRejectedWhilePaused(testSm.applyTransaction(mockTrx(sampleWriteRequest(),
1, 8)));
+ } finally {
+ testSm.stop();
+ }
+ }
+
// --- runCommand tests ---
@Test
@@ -1040,6 +1170,32 @@ public void testTakeSnapshotOmStopped() {
assertThrows(IOException.class, () -> sm.takeSnapshot());
}
+ @Test
+ public void testAwaitAppliedWaitsForFlush() throws Exception {
+ sm.notifyTermIndexUpdated(1, 0);
+ when(om.isStopped()).thenReturn(false);
+ doAnswer(invocation -> {
+ sm.updateLastAppliedTermIndex(TermIndex.valueOf(1, 2));
+ return null;
+ }).when(doubleBuffer).awaitFlush();
+
+ sm.awaitApplied(2);
+ assertTermIndex(1, 2, sm.getLastAppliedTermIndex());
+ verify(doubleBuffer).awaitFlush();
+
+ // Already applied: returns without waiting.
+ sm.awaitApplied(1);
+ verify(doubleBuffer).awaitFlush();
+ }
+
+ @Test
+ public void testAwaitAppliedOmStopped() {
+ when(om.isStopped()).thenReturn(true);
+ sm.notifyTermIndexUpdated(1, 0);
+
+ assertThrows(IOException.class, () -> sm.awaitApplied(1));
+ }
+
// --- loadSnapshotInfoFromDB tests ---
@Test
@@ -1281,6 +1437,58 @@ public void testPauseAlreadyPaused() {
verify(doubleBuffer, times(2)).stop();
}
+ @Test
+ public void testPauseWaitsForSnapshotDuringFlush() throws Exception {
+ initializeStateMachine();
+ OMMetadataManager metadataManager = mock(OMMetadataManager.class);
+ when(om.getMetadataManager()).thenReturn(metadataManager);
+ when(metadataManager.getStore()).thenReturn(mock(DBStore.class));
+
+ CountDownLatch flushWaitReached = new CountDownLatch(1);
+ CountDownLatch releaseFlushWait = new CountDownLatch(1);
+ CountDownLatch snapshotWriteReached = new CountDownLatch(1);
+ CountDownLatch releaseSnapshotWrite = new CountDownLatch(1);
+ doAnswer(invocation -> {
+ flushWaitReached.countDown();
+ awaitUninterruptibly(releaseFlushWait);
+ return null;
+ }).when(doubleBuffer).awaitFlush();
+
when(doubleBuffer.persistIfNewer(any(TransactionInfo.class))).thenAnswer(invocation
-> {
+ snapshotWriteReached.countDown();
+ awaitUninterruptibly(releaseSnapshotWrite);
+ return invocation.getArgument(0);
+ });
+
+ AtomicReference<Thread> pauseThread = new AtomicReference<>();
+ ExecutorService pauseExecutor = Executors.newSingleThreadExecutor(runnable
-> {
+ Thread thread = new Thread(runnable, "test-pause-during-snapshot");
+ pauseThread.set(thread);
+ return thread;
+ });
+ ExecutorService snapshotExecutor = Executors.newSingleThreadExecutor();
+ try {
+ Future<?> pauseFuture = pauseExecutor.submit(sm::pause);
+ assertTrue(flushWaitReached.await(5, TimeUnit.SECONDS));
+ Future<Long> snapshotFuture = snapshotExecutor.submit(sm::takeSnapshot);
+ assertTrue(snapshotWriteReached.await(5, TimeUnit.SECONDS));
+
+ releaseFlushWait.countDown();
+ waitFor(() -> pauseThread.get().getState() == Thread.State.BLOCKED, 10,
5000);
+ assertFalse(pauseFuture.isDone());
+
+ releaseSnapshotWrite.countDown();
+ snapshotFuture.get(5, TimeUnit.SECONDS);
+ pauseFuture.get(5, TimeUnit.SECONDS);
+ } finally {
+ releaseFlushWait.countDown();
+ releaseSnapshotWrite.countDown();
+ pauseExecutor.shutdownNow();
+ snapshotExecutor.shutdownNow();
+ pauseExecutor.awaitTermination(5, TimeUnit.SECONDS);
+ snapshotExecutor.awaitTermination(5, TimeUnit.SECONDS);
+ }
+ }
+
/**
* Reproduces the checkpoint-pause deadlock by holding the double-buffer
* flush callback immediately before it updates the last-applied index, then
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
