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 7007a9804b PHOENIX-7989 Fix close-resurrection double-writer in
ReplicationLogGroup (#2601)
7007a9804b is described below
commit 7007a9804b318aa0adf56b801eb167407f2f09f9
Author: tkhurana <[email protected]>
AuthorDate: Tue Aug 18 11:15:28 2026 -0700
PHOENIX-7989 Fix close-resurrection double-writer in ReplicationLogGroup
(#2601)
close() removed the instance from INSTANCES at the top, before a
multi-second disruptor drain. A concurrent get() during that window
missed the cache and constructed a second live instance for the same
shard, producing two writers (confirmed in a captured abort log).
Remove from the cache only after teardown, in a finally so a throwing
teardown step cannot strand a closed instance (which would break the
active->standby->active re-promotion path without a JVM restart). The
removal is value-specific (remove(key, this)) so a later instance for
the same key is never clobbered. The closed flag is still set first, so
a get() during the drain returns the still-closing instance whose
append()/sync() fail fast rather than a fresh writer.
Add a cacheForTesting() seam and two regression tests:
- testConcurrentGetDuringCloseDoesNotResurrectInstance
- testCloseRemovesFromCacheEvenWhenTeardownThrows
---
.../phoenix/replication/ReplicationLogGroup.java | 65 ++++++---
.../replication/ReplicationLogGroupTest.java | 149 +++++++++++++++++++++
2 files changed, 198 insertions(+), 16 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 add7d514d8..d77977b7a9 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
@@ -243,6 +243,17 @@ public class ReplicationLogGroup {
return serverIdentity(serverName) + "|" + haGroupName;
}
+ /**
+ * Register a pre-built (typically subclassed/spied) group in the instance
cache under its own
+ * identity, so a subsequent {@link #get} for the same (server, HA group)
returns it and
+ * {@link #close()} removes it via the normal path. Test-only seam that
avoids exposing the cache
+ * map or its key scheme.
+ */
+ @VisibleForTesting
+ static void cacheForTesting(ReplicationLogGroup group) {
+ INSTANCES.put(instanceKey(group.serverName, group.haGroupName), group);
+ }
+
protected final Configuration conf;
protected final ServerName serverName;
protected final String haGroupName;
@@ -995,15 +1006,16 @@ public class ReplicationLogGroup {
* Close the ReplicationLogGroup. Drains pending events from the Disruptor
with a bounded timeout
* (which triggers onShutdown → modeImpl.onExit → ReplicationLog.close),
then cleans up all
* resources. When the event handler has a fatal exception the drain
completes instantly because
- * each event hits the short-circuit path. The instance is removed from the
INSTANCES cache and
- * subsequent append()/sync() calls throw IOException.
+ * each event hits the short-circuit path. The closed flag is set
immediately so append()/sync()
+ * calls throw IOException, but the instance is removed from the INSTANCES
cache only after
+ * teardown completes, so a concurrent get() during the drain cannot
resurrect a second live
+ * instance.
*/
public void close() {
if (!closed.compareAndSet(false, true)) {
return;
}
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.
@@ -1017,20 +1029,41 @@ public class ReplicationLogGroup {
+ "proceeding with shutdown without the start gate", this, e);
Thread.currentThread().interrupt();
}
- // Remove every LOCAL state subscription registered in
subscribeToStateChanges().
- stateUnsubscribers.forEach(Runnable::run);
- stateUnsubscribers.clear();
-
try {
- disruptor.shutdown(shutdownTimeoutMs, TimeUnit.MILLISECONDS);
- } catch (com.lmax.disruptor.TimeoutException e) {
- LOG.warn("HAGroup {} shutdown timed out after {}ms, halting", this,
shutdownTimeoutMs);
- disruptor.halt();
- }
- shutdownDisruptorExecutor();
- logForwarder.stop();
- logForwarder.close();
- metrics.close();
+ // Remove every LOCAL state subscription registered in
subscribeToStateChanges(). Swallow an
+ // unchecked throw here: this runs before disruptor.shutdown() closes
the writer, so letting
+ // it
+ // propagate would skip the writer teardown and jump straight to the
finally cache removal --
+ // leaving a live consumer + writer while a racing get() mints a second
instance, the exact
+ // double-writer this close ordering exists to prevent.
+ try {
+ stateUnsubscribers.forEach(Runnable::run);
+ stateUnsubscribers.clear();
+ } catch (RuntimeException e) {
+ LOG.warn("HAGroup {} error unsubscribing from state changes during
close; "
+ + "continuing with disruptor shutdown", this, e);
+ }
+
+ try {
+ disruptor.shutdown(shutdownTimeoutMs, TimeUnit.MILLISECONDS);
+ } catch (com.lmax.disruptor.TimeoutException e) {
+ LOG.warn("HAGroup {} shutdown timed out after {}ms, halting", this,
shutdownTimeoutMs);
+ disruptor.halt();
+ }
+ shutdownDisruptorExecutor();
+ logForwarder.stop();
+ logForwarder.close();
+ metrics.close();
+ } finally {
+ // Remove from the cache only after teardown, so a concurrent get()
during the (potentially
+ // multi-second) drain finds this still-closing instance -- whose
append()/sync() fail fast on
+ // the closed flag -- instead of a cache miss that would mint a second
live instance for the
+ // same group, producing a double-writer against the same shard. In a
finally so a teardown
+ // step that throws can never strand a closed instance: on an
active->standby->active cycle
+ // without a JVM restart, get() would otherwise return the dead instance
forever. Value-
+ // specific remove so a later instance created for this key is never
clobbered.
+ INSTANCES.remove(instanceKey(serverName, haGroupName), this);
+ }
LOG.info("HAGroup {} closed", this);
}
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 855fe9dccb..b59e4cb96d 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
@@ -25,6 +25,7 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -36,6 +37,7 @@ import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.timeout;
@@ -75,6 +77,7 @@ import
org.apache.phoenix.replication.log.LogFileReaderContext;
import org.apache.phoenix.replication.log.LogFileTestUtil;
import org.apache.phoenix.replication.log.LogFileWriter;
import org.apache.phoenix.replication.log.LogFileWriterContext;
+import org.apache.phoenix.replication.metrics.MetricsReplicationLogGroupSource;
import org.apache.phoenix.replication.metrics.ReplicationLogMetricValues;
import org.junit.Assume;
import org.junit.Test;
@@ -1461,6 +1464,152 @@ public class ReplicationLogGroupTest extends
ReplicationLogBaseTest {
g1_3.close();
}
+ /**
+ * A get() concurrent with a still-draining close() must NOT resurrect a
second live instance for
+ * the same group. close() sets the closed flag first but removes the
instance from the cache only
+ * after teardown, so a get() during that window returns the same (closing)
instance rather than
+ * minting a new one -- avoiding the double-writer against the same shard.
+ * <p>
+ * The window is made deterministic by blocking the last teardown step
(metrics.close()) on a
+ * latch, so close() is provably parked inside teardown -- after the closed
flag is set, before
+ * the cache removal -- while the racing get() runs.
+ */
+ @Test(timeout = 60000)
+ public void testConcurrentGetDuringCloseDoesNotResurrectInstance() throws
Exception {
+ final String haGroupId = "testResurrectionGuard";
+ final CountDownLatch closeInTeardown = new CountDownLatch(1);
+ final CountDownLatch releaseClose = new CountDownLatch(1);
+ MetricsReplicationLogGroupSource blockingMetrics =
mock(MetricsReplicationLogGroupSource.class);
+ doAnswer(inv -> {
+ closeInTeardown.countDown();
+ releaseClose.await();
+ return null;
+ }).when(blockingMetrics).close();
+
+ ReplicationLogGroup group =
+ new TestableLogGroup(conf, serverName, haGroupId, haGroupStoreManager,
useAlignedRotation()) {
+ @Override
+ protected MetricsReplicationLogGroupSource createMetricsSource() {
+ return blockingMetrics;
+ }
+ };
+ group.init();
+ ReplicationLogGroup.cacheForTesting(group);
+ try {
+ Thread closer = new Thread(group::close, "closer");
+ closer.start();
+
+ // Wait until close() is provably parked in teardown (flag set, cache
entry not yet removed).
+ assertTrue("close() should reach the teardown step",
+ closeInTeardown.await(30, TimeUnit.SECONDS));
+ assertTrue("closed flag must be set before teardown removes the cache
entry",
+ group.isClosed());
+
+ // A get() in this window must return the SAME (closing) instance, not a
fresh one.
+ ReplicationLogGroup duringClose =
+ ReplicationLogGroup.get(conf, serverName, haGroupId,
haGroupStoreManager).get();
+ assertSame("A get() during close()'s teardown must not resurrect a
second instance", group,
+ duringClose);
+
+ releaseClose.countDown();
+ closer.join(10000);
+ assertFalse("closer thread should have finished", closer.isAlive());
+
+ // After close() fully completes, the entry is gone, so a later get()
creates a fresh
+ // instance.
+ ReplicationLogGroup afterClose =
+ ReplicationLogGroup.get(conf, serverName, haGroupId,
haGroupStoreManager).get();
+ assertNotSame("A get() after close() completes must create a new
instance", group,
+ afterClose);
+ afterClose.close();
+ } finally {
+ releaseClose.countDown();
+ group.close();
+ }
+ }
+
+ /**
+ * A teardown step that throws inside close() must still remove the instance
from the cache;
+ * otherwise a dangling closed instance is stranded and an
active->standby->active cycle (which
+ * calls close() on demotion) would, on re-promotion without a JVM restart,
hand the dead instance
+ * back to get() and every append()/sync() would throw "Closed" forever. The
removal lives in a
+ * finally to guarantee this.
+ */
+ @Test
+ public void testCloseRemovesFromCacheEvenWhenTeardownThrows() throws
Exception {
+ final String haGroupId = "testTeardownThrows";
+ // metrics.close() is the last teardown step in close(); make it throw.
+ MetricsReplicationLogGroupSource throwingMetrics =
mock(MetricsReplicationLogGroupSource.class);
+ doThrow(new RuntimeException("simulated metrics.close()
failure")).when(throwingMetrics)
+ .close();
+
+ ReplicationLogGroup group =
+ new TestableLogGroup(conf, serverName, haGroupId, haGroupStoreManager,
useAlignedRotation()) {
+ @Override
+ protected MetricsReplicationLogGroupSource createMetricsSource() {
+ return throwingMetrics;
+ }
+ };
+ group.init();
+ ReplicationLogGroup.cacheForTesting(group);
+
+ try {
+ group.close();
+ fail("close() should propagate the teardown failure");
+ } catch (RuntimeException expected) {
+ assertEquals("simulated metrics.close() failure", expected.getMessage());
+ }
+
+ assertTrue("Group should still be marked closed", group.isClosed());
+ // Despite the thrown teardown, the cache entry must be gone: a fresh
get() creates a NEW,
+ // usable instance rather than returning the stranded closed one.
+ ReplicationLogGroup afterClose =
+ ReplicationLogGroup.get(conf, serverName, haGroupId,
haGroupStoreManager).get();
+ assertNotSame("A stranded closed instance must not be returned after a
throwing close()", group,
+ afterClose);
+ assertFalse("Re-created instance must be usable, not closed",
afterClose.isClosed());
+ afterClose.close();
+ }
+
+ /**
+ * An unchecked throw from the state-unsubscribe step -- which runs before
disruptor.shutdown()
+ * closes the writer -- must not skip the writer teardown. If it propagated,
close() would jump
+ * straight to the finally cache removal while the consumer thread and
writer were still alive,
+ * and a racing get() would mint a second live instance: the double-writer
this close ordering
+ * exists to prevent. The throw is swallowed so shutdown still runs and the
writer is closed.
+ */
+ @Test(timeout = 60000)
+ public void testCloseUnsubscribeThrowStillClosesWriter() throws Exception {
+ final String haGroupId = "testUnsubscribeThrows";
+ ReplicationLogGroup group =
+ new TestableLogGroup(conf, serverName, haGroupId, haGroupStoreManager,
useAlignedRotation());
+ group.init();
+ ReplicationLogGroup.cacheForTesting(group);
+
+ LogFileWriter innerWriter = group.getActiveLog().getWriter();
+ assertNotNull("Inner writer should not be null", innerWriter);
+
+ // subscribeToStateChanges() registers unsubscribers that call
unsubscribeFromTargetState;
+ // make the first one throw so the unsubscribe step throws before
disruptor.shutdown().
+ doThrow(new RuntimeException("simulated unsubscribe
failure")).when(haGroupStoreManager)
+ .unsubscribeFromTargetState(eq(haGroupId), any(HAGroupState.class),
eq(ClusterType.LOCAL),
+ any(HAGroupStateListener.class));
+
+ // close() must not propagate the unsubscribe failure.
+ group.close();
+
+ // The writer was still closed despite the thrown unsubscribe.
+ verify(innerWriter, times(1)).close();
+ assertTrue("Group should be marked closed", group.isClosed());
+
+ // Cache entry removed: a fresh get() creates a NEW instance.
+ ReplicationLogGroup afterClose =
+ ReplicationLogGroup.get(conf, serverName, haGroupId,
haGroupStoreManager).get();
+ assertNotSame("close() must remove the cache entry even when unsubscribe
throws", group,
+ afterClose);
+ afterClose.close();
+ }
+
@Test
public void testInFlightAppendsReplayAfterModeSwitch() throws Exception {
final String tableName = "TESTTBL";
