This is an automated email from the ASF dual-hosted git repository.
tkhurana pushed a commit to branch PHOENIX-7562-feature-new
in repository https://gitbox.apache.org/repos/asf/phoenix.git
The following commit(s) were added to refs/heads/PHOENIX-7562-feature-new by
this push:
new 4ccb7a3d52 PHOENIX-7960 ReplicationLogGroup.close() racing init() can
swallow the disruptor halt alert, leaking the writer (#2579)
4ccb7a3d52 is described below
commit 4ccb7a3d522489301a7d74cadfb1f60347cdaf9e
Author: tkhurana <[email protected]>
AuthorDate: Wed Jul 22 10:15:37 2026 -0700
PHOENIX-7960 ReplicationLogGroup.close() racing init() can swallow the
disruptor halt alert, leaking the writer (#2579)
A close() that immediately follows init() could halt() the disruptor before
its consumer thread entered BatchEventProcessor.run(). run()'s opening
clearAlert() then wiped the alert raised by halt(), the wait loop parked
forever, and onShutdown never ran, so the underlying writer was never
closed.
This surfaced as a flaky testClose (Mockito "wanted but not invoked:
close()") and latently affects every recreateLogGroup()/tearDown() that
closes a zero-append group right after creation.
Add a startedLatch counted down in LogEventHandler.onStart() and await it
(bounded by shutdownTimeoutMs) in close() before disruptor.shutdown(), so
the
consumer is guaranteed past clearAlert() when halt() raises the alert.
Extract createDisruptorExecutor() as a @VisibleForTesting seam and add
testCloseImmediatelyAfterInitClosesWriterWhenConsumerStartsLate, which
delays
the consumer start to force halt() into the pre-clearAlert window and
verifies
the writer is still closed.
---
.../phoenix/replication/ReplicationLogGroup.java | 36 ++++++++-
.../replication/ReplicationLogGroupTest.java | 89 +++++++++++++++++++++-
2 files changed, 118 insertions(+), 7 deletions(-)
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogGroup.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogGroup.java
index 9959ecc8aa..aee7143761 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogGroup.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogGroup.java
@@ -50,6 +50,7 @@ import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -555,8 +556,7 @@ public class ReplicationLogGroup {
protected void initializeDisruptor() throws IOException {
int ringBufferSize =
conf.getInt(REPLICATION_LOG_RINGBUFFER_SIZE_KEY,
DEFAULT_REPLICATION_LOG_RINGBUFFER_SIZE);
- this.disruptorExecutor = Executors.newCachedThreadPool(new
ThreadFactoryBuilder()
- .setNameFormat("ReplicationLogGroup-" + getHAGroupName() +
"-%d").setDaemon(true).build());
+ this.disruptorExecutor = createDisruptorExecutor();
disruptor = new Disruptor<>(LogEvent.EVENT_FACTORY, ringBufferSize,
disruptorExecutor,
ProducerType.MULTI, new YieldingWaitStrategy());
eventHandler = new LogEventHandler();
@@ -567,6 +567,12 @@ public class ReplicationLogGroup {
ringBuffer = disruptor.start();
}
+ @VisibleForTesting
+ protected ExecutorService createDisruptorExecutor() {
+ return Executors.newCachedThreadPool(new ThreadFactoryBuilder()
+ .setNameFormat("ReplicationLogGroup-" + getHAGroupName() +
"-%d").setDaemon(true).build());
+ }
+
/**
* Append a mutation to the log. This method is non-blocking and returns
quickly, unless the ring
* buffer is full. The actual write happens asynchronously. We expect
multiple append() calls
@@ -757,6 +763,19 @@ public class ReplicationLogGroup {
}
LOG.info("Closing HAGroup {}", this);
INSTANCES.remove(instanceKey(serverName, haGroupName));
+ // Wait for the consumer thread to enter its run loop before shutting
down. Otherwise a close()
+ // that races init() can halt() before the consumer starts; run()'s
opening clearAlert() then
+ // wipes the alert and onShutdown (writer close) never runs.
+ try {
+ if (!eventHandler.awaitStarted(shutdownTimeoutMs,
TimeUnit.MILLISECONDS)) {
+ LOG.warn("HAGroup {} consumer thread did not start within {}ms before
close", this,
+ shutdownTimeoutMs);
+ }
+ } catch (InterruptedException e) {
+ LOG.warn("HAGroup {} interrupted while awaiting consumer thread start; "
+ + "proceeding with shutdown without the start gate", this, e);
+ Thread.currentThread().interrupt();
+ }
try {
disruptor.shutdown(shutdownTimeoutMs, TimeUnit.MILLISECONDS);
} catch (com.lmax.disruptor.TimeoutException e) {
@@ -1052,6 +1071,10 @@ public class ReplicationLogGroup {
private volatile IOException fatalException;
// Counts events drained per Disruptor batch. Single-threaded access from
onEvent.
private int batchEventCount;
+ // Counted down by onStart() once the consumer thread has entered its run
loop (past
+ // BatchEventProcessor.run()'s opening clearAlert()). close() awaits this
before halting so
+ // halt()'s alert cannot be swallowed by a clearAlert() that races an
immediate close.
+ private final CountDownLatch startedLatch = new CountDownLatch(1);
public LogEventHandler() {
}
@@ -1299,7 +1322,14 @@ public class ReplicationLogGroup {
@Override
public void onStart() {
- // no-op
+ // run() has already cleared any pending alert by the time onStart runs,
so a subsequent
+ // halt() from close() is guaranteed to be observed by the wait loop.
+ startedLatch.countDown();
+ }
+
+ /** Await (bounded) until the consumer thread has entered its run loop. */
+ boolean awaitStarted(long timeout, TimeUnit unit) throws
InterruptedException {
+ return startedLatch.await(timeout, unit);
}
@Override
diff --git
a/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogGroupTest.java
b/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogGroupTest.java
index d74eab43ba..7a7ae208a0 100644
---
a/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogGroupTest.java
+++
b/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogGroupTest.java
@@ -48,13 +48,19 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.phoenix.jdbc.HAGroupStoreManager;
import org.apache.phoenix.jdbc.HAGroupStoreRecord;
import org.apache.phoenix.jdbc.HAGroupStoreRecord.HAGroupState;
import org.apache.phoenix.jdbc.HighAvailabilityPolicy;
@@ -496,6 +502,81 @@ public class ReplicationLogGroupTest extends
ReplicationLogBaseTest {
logGroup.close();
}
+ /**
+ * Regression test for the start/shutdown clearAlert race: a close() that
immediately follows
+ * init() must still close the underlying writer even when the disruptor
consumer thread is slow
+ * to start. Without the startedLatch gate in close(),
disruptor.shutdown()'s halt() raises an
+ * alert before the consumer thread enters BatchEventProcessor.run();
run()'s opening clearAlert()
+ * then wipes that alert, the wait loop spins forever, and onShutdown
(writer close) never runs.
+ * <p>
+ * This mirrors the recreateLogGroup()/tearDown() path where a group with
zero appends is closed
+ * right after creation, so we intentionally do NOT append (a backlog would
make shutdown() wait
+ * for the consumer regardless and mask the race).
+ */
+ @Test(timeout = 60000)
+ public void
testCloseImmediatelyAfterInitClosesWriterWhenConsumerStartsLate() throws
Exception {
+ // Delay the consumer thread's start by ~500ms so close()'s halt() lands
in the window before
+ // run()'s clearAlert(). The startedLatch gate in close() must wait this
out.
+ ReplicationLogGroup lateGroup = new LateStartLogGroup(conf, serverName,
haGroupName,
+ haGroupStoreManager, useAlignedRotation(), 500L);
+ try {
+ lateGroup.init();
+
+ // currentWriter is created synchronously in ReplicationLog.init(), so
it exists even though
+ // the consumer thread has not started yet.
+ LogFileWriter innerWriter = lateGroup.getActiveLog().getWriter();
+ assertNotNull("Inner writer should not be null", innerWriter);
+
+ // Close immediately, before the delayed consumer thread has entered its
run loop.
+ lateGroup.close();
+
+ // The gate ensures onShutdown ran, so the writer must have been closed.
+ verify(innerWriter, times(1)).close();
+ } finally {
+ lateGroup.close();
+ }
+ }
+
+ /**
+ * A ReplicationLogGroup whose disruptor consumer thread is delayed before
it starts executing,
+ * used to force close()'s halt() into the pre-clearAlert window.
+ */
+ static class LateStartLogGroup extends
ReplicationLogBaseTest.TestableLogGroup {
+ private final long startDelayMs;
+
+ LateStartLogGroup(Configuration conf, ServerName serverName, String
haGroupName,
+ HAGroupStoreManager haGroupStoreManager, boolean useAlignedRotation,
long startDelayMs) {
+ super(conf, serverName, haGroupName, haGroupStoreManager,
useAlignedRotation);
+ this.startDelayMs = startDelayMs;
+ }
+
+ @Override
+ protected ExecutorService createDisruptorExecutor() {
+ ThreadFactory factory = new ThreadFactory() {
+ private int count = 0;
+
+ @Override
+ public Thread newThread(Runnable r) {
+ Thread t = new Thread(r, "LateStartLogGroup-" + getHAGroupName() +
"-" + (count++));
+ t.setDaemon(true);
+ return t;
+ }
+ };
+ // Sleep before the consumer Runnable executes so run()'s clearAlert()
is delayed.
+ return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L,
TimeUnit.SECONDS,
+ new SynchronousQueue<>(), factory) {
+ @Override
+ protected void beforeExecute(Thread t, Runnable r) {
+ try {
+ Thread.sleep(startDelayMs);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ };
+ }
+ }
+
/**
* Tests the automatic rotation task. Verifies that the background rotation
task correctly rotates
* log files based on the configured rotation time. The writer swap happens
pre-action in apply(),
@@ -2371,10 +2452,10 @@ public class ReplicationLogGroupTest extends
ReplicationLogBaseTest {
// publish can floor to a 0ns delta, so assert presence (>= 0) rather than
strict positivity.
assertTrue("appendTime should be >= 0, got " + values.getAppendTimeMax(),
values.getAppendTimeMax() >= 0);
- assertTrue("ringBufferTime should be > 0, got " +
values.getRingBufferTimeMax(),
- values.getRingBufferTimeMax() > 0);
- assertTrue("pendingSyncWaitTime should be > 0, got " +
values.getPendingSyncWaitTimeMax(),
- values.getPendingSyncWaitTimeMax() > 0);
+ assertTrue("ringBufferTime should be >= 0, got " +
values.getRingBufferTimeMax(),
+ values.getRingBufferTimeMax() >= 0);
+ assertTrue("pendingSyncWaitTime should be >= 0, got " +
values.getPendingSyncWaitTimeMax(),
+ values.getPendingSyncWaitTimeMax() >= 0);
// Counts.
assertTrue("batchSize should be > 0, got " + values.getBatchSizeMax(),
values.getBatchSizeMax() > 0);
