smengcl commented on code in PR #10995:
URL: https://github.com/apache/ozone/pull/10995#discussion_r3762801594
##########
hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOzoneManagerStateMachine.java:
##########
@@ -918,6 +924,73 @@ public void testPauseAlreadyPaused() {
verify(doubleBuffer, times(2)).stop();
}
+ /**
+ * Reproduces the checkpoint-pause deadlock by holding the double-buffer
+ * flush callback immediately before it updates the last-applied index, then
+ * pausing the state machine while it waits for the flush thread to exit.
+ * Releasing the callback must allow both the index update and pause to
finish.
+ */
+ @Test
+ public void testPauseWhileDoubleBufferUpdatesLastAppliedIndex(@TempDir Path
tmpDir) throws Exception {
+ OzoneConfiguration conf = new OzoneConfiguration();
+ conf.set(OMConfigKeys.OZONE_OM_DB_DIRS,
tmpDir.toAbsolutePath().toString());
+ OzoneManager testOm = mock(OzoneManager.class);
+ when(testOm.getConfiguration()).thenReturn(conf);
+ when(testOm.getConfig()).thenReturn(conf.getObject(OmConfig.class));
+ OmMetadataManagerImpl metadataManager = new OmMetadataManagerImpl(conf,
testOm);
+ when(testOm.getMetadataManager()).thenReturn(metadataManager);
+
+ CountDownLatch flushCallbackReached = new CountDownLatch(1);
+ CountDownLatch releaseFlushCallback = new CountDownLatch(1);
+ AtomicReference<OzoneManagerStateMachine> stateMachineRef = new
AtomicReference<>();
+ OzoneManagerDoubleBuffer testDoubleBuffer =
OzoneManagerDoubleBuffer.newBuilder()
+ .setOmMetadataManager(metadataManager)
+ .setUpdateLastAppliedIndex(termIndex -> {
+ flushCallbackReached.countDown();
+ awaitUninterruptibly(releaseFlushCallback);
+ stateMachineRef.get().updateLastAppliedTermIndex(termIndex);
+ })
+ .setMaxUnFlushedTransactionCount(10)
+ .build()
+ .start();
+ ExecutorService stateMachineExecutor = Executors.newSingleThreadExecutor();
+ OzoneManagerStateMachine testStateMachine = new OzoneManagerStateMachine(
+ testOm, testDoubleBuffer, mock(RequestHandler.class),
stateMachineExecutor, null);
+ stateMachineRef.set(testStateMachine);
+
+ AtomicReference<Thread> pauseThread = new AtomicReference<>();
+ ExecutorService pauseExecutor = Executors.newSingleThreadExecutor(runnable
-> {
+ Thread thread = new Thread(runnable, "test-state-machine-pause");
+ pauseThread.set(thread);
+ return thread;
+ });
+ Future<?> pauseFuture = null;
+ try {
+ OMResponse response = OMResponse.newBuilder()
+ .setCmdType(Type.CreateKey)
+ .setStatus(Status.OK)
+ .setSuccess(true)
+ .build();
+ testDoubleBuffer.add(new DummyOMClientResponse(response),
TermIndex.valueOf(1, 1));
+ assertTrue(flushCallbackReached.await(5, TimeUnit.SECONDS));
+
+ pauseFuture = pauseExecutor.submit(testStateMachine::pause);
+ waitFor(() -> pauseThread.get() != null && pauseThread.get().getState()
== Thread.State.WAITING, 10, 5000);
+
+ releaseFlushCallback.countDown();
+ pauseFuture.get(5, TimeUnit.SECONDS);
+ assertEquals(1, testStateMachine.getLastAppliedTermIndex().getIndex());
+ } finally {
+ releaseFlushCallback.countDown();
+ if (pauseFuture != null && !pauseFuture.isDone() && pauseThread.get() !=
null) {
+ pauseThread.get().interrupt();
+ }
+ pauseExecutor.shutdownNow();
+ pauseExecutor.awaitTermination(5, TimeUnit.SECONDS);
+ testStateMachine.stop();
+ }
Review Comment:
Fixed. The test now stops the metadata manager after stopping the state
machine, closing the underlying RocksDB store.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
