Add ReplicationLogWriterTest#testBlockingWhenRingFull

Also fix testAppendAndSync

Author: apurtell

phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogWriterTest.java

@@ -18,14 +18,20 @@
 package org.apache.phoenix.replication;
 
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.assertFalse;
 import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.timeout;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.verify;
 
 import java.io.IOException;
 import java.util.concurrent.CompletableFuture;
@@ -63,9 +69,14 @@ public class ReplicationLogWriterTest {
     private TestableReplicationLogManager logManager;
     private LogFile.Writer internalWriter;
 
+    private static final int TEST_RING_BUFFER_SIZE = 32;
+
     @Before
     public void setUp() throws IOException {
         conf = HBaseConfiguration.create();
+        // Small ring buffer size for testing
+        conf.setInt(ReplicationLogWriter.REPLICATION_WRITER_RINGBUFFER_SIZE_KEY,
+            TEST_RING_BUFFER_SIZE);
         serverName = ServerName.valueOf("test", 60010, EnvironmentEdgeManager.currentTimeMillis());
         // Use a temporary folder for the standby HDFS URL
         standbyLogDir = new Path(testFolder.newFolder("standby").toURI());
@@ -159,30 +170,23 @@ public Void answer(InvocationOnMock invocation) throws Throwable {
         });
         syncThread.start();
 
-        // Happens-before ordering verification, using Mockito's inOrder.
+        // Wait for the sync() call itself to complete. Any issues here will cause the test to
+        // stall and eventually time out.
+        syncResultFuture.get();
 
-        // Verify that the appends happen before sync, and sync happened after appends
-        // Use timeout with verify to wait for the async event handler to process
+        // Happens-before ordering verification, using Mockito's inOrder. Verify that the appends
+        // happen before sync, and sync happened after appends.
         InOrder inOrder = Mockito.inOrder(internalWriter);
 
-        // Wait for appends (these might happen very quickly)
-        inOrder.verify(internalWriter, timeout(1000).times(1)).append(eq(tableName), eq(commitId1),
-            eq(put1));
-        inOrder.verify(internalWriter, timeout(1000).times(1)).append(eq(tableName), eq(commitId2),
-            eq(put2));
-        inOrder.verify(internalWriter, timeout(1000).times(1)).append(eq(tableName), eq(commitId3),
-            eq(put3));
-        inOrder.verify(internalWriter, timeout(1000).times(1)).append(eq(tableName), eq(commitId4),
-            eq(put4));
-        inOrder.verify(internalWriter, timeout(1000).times(1)).append(eq(tableName), eq(commitId5),
-            eq(put5));
-
-        // Wait for sync (this depends on the syncFuture completion)
-        inOrder.verify(internalWriter, timeout(1000).times(1)).sync();
+        // Verify the appends happened in order.
+        inOrder.verify(internalWriter, times(1)).append(eq(tableName), eq(commitId1), eq(put1));
+        inOrder.verify(internalWriter, times(1)).append(eq(tableName), eq(commitId2), eq(put2));
+        inOrder.verify(internalWriter, times(1)).append(eq(tableName), eq(commitId3), eq(put3));
+        inOrder.verify(internalWriter, times(1)).append(eq(tableName), eq(commitId4), eq(put4));
+        inOrder.verify(internalWriter, times(1)).append(eq(tableName), eq(commitId5), eq(put5));
 
-        // Wait for the sync() call itself to complete. Any issues here will cause the test to
-        // stall and eventually time out.
-        syncResultFuture.get();
+        // Verify the sync happened after the appends.
+        inOrder.verify(internalWriter, times(1)).sync();
 
         // Close the log writer.
         logWriter.close();
@@ -214,6 +218,9 @@ public void testSyncFailureAndRetry() throws Exception {
         LogFile.Writer writerAfterRoll = mock(LogFile.Writer.class);
         when(writerAfterRoll.getLength()).thenReturn(0L); // Needed for shouldRotate check
 
+        // Configure writerAfterRoll to succeed on sync
+        doAnswer(invocation -> null).when(writerAfterRoll).sync();
+
         // Configure the logManager's rotateLog method to return the second writer
         doAnswer(invocation -> {
             // Simulate the actual rotation behavior: close old, create new
@@ -225,14 +232,10 @@ public void testSyncFailureAndRetry() throws Exception {
             return logManager.currentWriter;
         }).when(logManager).rotateLog();
 
-        // Configure internalWriter2 to succeed on sync
-        doAnswer(invocation -> null).when(writerAfterRoll).sync();
-
         // Append data
         logWriter.append(tableName, commitId1, put1);
 
-        // Call sync and expect it to eventually succeed after retry. Do it in another thread so we
-        // continue to InOrder behavioral verification.
+        // Call sync and expect it to eventually succeed after retry.
         CompletableFuture<Void> syncResultFuture = new CompletableFuture<>();
         Thread syncThread = new Thread(() -> {
             try {
@@ -244,25 +247,84 @@ public void testSyncFailureAndRetry() throws Exception {
         });
         syncThread.start();
 
+        // Wait for the sync() call itself to complete successfully
+        syncResultFuture.get();
+
         // Verify the sequence: append, sync (fail), rotate, append (retry), sync (succeed)
         InOrder inOrder = Mockito.inOrder(writerBeforeRoll, writerAfterRoll, logManager);
-
-        // 1. Append to the first writer
+        // Append to the first writer.
         inOrder.verify(writerBeforeRoll, times(1)).append(eq(tableName), eq(commitId1), eq(put1));
-        // 2. First sync attempt fails
+        // First sync attempt fails.
         inOrder.verify(writerBeforeRoll, times(1)).sync();
-        // 3. Log rotation is triggered due to failure
-        inOrder.verify(logManager, times(1)).rotateLog(); // Verify rotation was called
-
-        // Wait for the sync() call itself to complete successfully
-        syncResultFuture.get();
-
-        // 4. Append is retried on the second writer
+        // Log rotation is triggered due to failure.
+        inOrder.verify(logManager, times(1)).rotateLog();
+        // Append is retried on the second writer.
         inOrder.verify(writerAfterRoll, times(1)).append(eq(tableName), eq(commitId1), eq(put1));
-        // 5. Second sync attempt succeeds
+        // Second sync attempt succeeds.
         inOrder.verify(writerAfterRoll, times(1)).sync();
 
         logWriter.close();
     }
 
+    @Test(timeout=15000)
+    public void testBlockingWhenRingFull() throws Exception {
+        final String tableName = "TBLBWRF";
+        final Mutation put = LogFileTestUtil.newPut("row", 1, 1);
+        long commitId = 0;
+
+        // Create the ReplicationLogWriter instance to be tested.
+        ReplicationLogWriter logWriter = new ReplicationLogWriter(conf, logManager);
+        logWriter.init();
+
+        // Create a slow consumer to fill up the ring buffer.
+        doAnswer(new Answer<Void>() {
+            @Override
+            public Void answer(InvocationOnMock invocation) throws Throwable {
+                Thread.sleep(50); // Simulate slow processing
+                return null;
+            }
+        }).when(internalWriter).append(anyString(), anyLong(), any(Mutation.class));
+
+        // Fill up the ring buffer by sending enough events.
+        for (int i = 0; i < TEST_RING_BUFFER_SIZE; i++) {
+            logWriter.append(tableName, commitId++, put);
+        }
+
+        // Now try to append when the ring is full. This should block until space becomes
+        // available.
+        long myCommitId = commitId++;
+        CompletableFuture<Void> startFuture = new CompletableFuture<>();
+        CompletableFuture<Void> appendFuture = new CompletableFuture<>();
+        Thread appendThread = new Thread(() -> {
+            try {
+                startFuture.complete(null);
+                logWriter.append(tableName, myCommitId, put);
+                appendFuture.complete(null);
+            } catch (IOException e) {
+                appendFuture.completeExceptionally(e);
+            }
+        });
+        appendThread.start();
+
+        // Wait for the append thread.
+        startFuture.get();
+
+        // Verify the append is still blocked
+        assertFalse("Append should be blocked when ring is full", appendFuture.isDone());
+
+        // Let some events process to free up space.
+        Thread.sleep(1000);
+
+        // Now the append should complete. Any issues and we will time out here.
+        appendFuture.get();
+        assertTrue("Append should have completed", appendFuture.isDone());
+
+        // Verify the append eventually happens on the writer. We may have to wait for a while
+        // because of our slow consumer.
+        verify(internalWriter, timeout(10000).times(1)).append(eq(tableName), eq(myCommitId),
+            eq(put));
+
+        logWriter.close();
+    }
+
 }