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 c9ea66c177 PHOENIX-8030 Close replication log groups after the regions
close on RegionServer stop (#2642)
c9ea66c177 is described below
commit c9ea66c1779ed1969518242767b419b11db9196d
Author: tkhurana <[email protected]>
AuthorDate: Wed Oct 28 16:32:01 2026 -0700
PHOENIX-8030 Close replication log groups after the regions close on
RegionServer stop (#2642)
PhoenixRegionServerEndpoint.stop() ran ReplicationLogGroup.closeAll() in
preStop, before HBase closes the regions. A live write already committed
to the WAL could then fail replication with IOException("Closed"); with a
clean region close the WAL is never replayed, so the edit was never
shipped.
The endpoint's stop now closes the groups only if the server has no open
IndexRegionObserver region, e.g. after the regions were moved away before
a graceful stop (ReplicationLogGroup.closeAllIfNoOpenRegions). Otherwise
the close is deferred: IndexRegionObserver counts its open regions per
server (postOpen/postClose), and when the last one closes while the
server is stopped, stopping or aborted, ReplicationLogGroup.regionClosed()
closes that server's groups. HBase calls postClose after the region close
has drained in-flight writes, including their replication sync, so every
committed write is shipped and the writers close cleanly with a trailer.
A failed open runs postClose, possibly without postOpen, so a per-region
flag keeps the count from being undercounted. A count that reaches zero
in normal operation (regions moved away) leaves the groups open.
A kill skips the region close, so testWALRestore still closes the killed
RS's groups by hand. testGracefulRegionServerStopClosesLogGroups holds a
write after its WAL commit across a stop and checks that the stopping
RS's own log carries it; it fails with the old preStop close.
testRegionServerStopAfterUnloadClosesLogGroups moves the regions away
first and checks that the stop closes the groups.
---
.../coprocessor/PhoenixRegionServerEndpoint.java | 35 ++--
.../phoenix/hbase/index/IndexRegionObserver.java | 25 +++
.../phoenix/replication/ReplicationLogGroup.java | 56 ++++-
.../phoenix/replication/ReplicationLogGroupIT.java | 231 +++++++++++++++++++++
.../replication/ReplicationLogGroupTest.java | 135 ++++++++++++
5 files changed, 465 insertions(+), 17 deletions(-)
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/PhoenixRegionServerEndpoint.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/PhoenixRegionServerEndpoint.java
index d271ec5fa0..9aae123630 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/PhoenixRegionServerEndpoint.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/PhoenixRegionServerEndpoint.java
@@ -108,21 +108,26 @@ public class PhoenixRegionServerEndpoint extends
@Override
public void stop(CoprocessorEnvironment env) throws IOException {
- // Stop replication log replay
- ReplicationLogReplayService.getInstance(conf).stop();
- // Close all ReplicationLogGroup instances belonging to this server
- if (serverName != null) {
- ReplicationLogGroup.closeAll(serverName);
- }
- RegionServerCoprocessor.super.stop(env);
- if (uncoveredIndexThreadPool != null) {
- uncoveredIndexThreadPool
- .stop("PhoenixRegionServerEndpoint is stopping. Shutting down
uncovered index threadpool.");
- }
- ServerUtil.ConnectionFactory.shutdown();
- // Stop prewarming executor
- if (prewarmExecutor != null) {
- prewarmExecutor.shutdownNow();
+ try {
+ // Stop replication log replay
+ ReplicationLogReplayService.getInstance(conf).stop();
+ RegionServerCoprocessor.super.stop(env);
+ if (uncoveredIndexThreadPool != null) {
+ uncoveredIndexThreadPool.stop(
+ "PhoenixRegionServerEndpoint is stopping. Shutting down uncovered
index threadpool.");
+ }
+ ServerUtil.ConnectionFactory.shutdown();
+ // Stop prewarming executor
+ if (prewarmExecutor != null) {
+ prewarmExecutor.shutdownNow();
+ }
+ } finally {
+ // On a graceful stop this runs in preStop, before HBase closes the
regions, so close the
+ // groups only if no IndexRegionObserver region is open; otherwise the
last region close
+ // does it.
+ if (serverName != null) {
+ ReplicationLogGroup.closeAllIfNoOpenRegions(serverName);
+ }
}
}
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java
index a11b5ec144..6cbd8e7a54 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java
@@ -68,6 +68,7 @@ import org.apache.hadoop.hbase.CoprocessorEnvironment;
import org.apache.hadoop.hbase.DoNotRetryIOException;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.Server;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Durability;
@@ -622,6 +623,8 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
private boolean stopped;
private boolean disabled;
+ // Whether postOpen counted this region; postClose decrements only then.
+ private boolean countedOpen;
private long slowIndexPrepareThreshold;
private long slowPreIncrementThreshold;
private int rowLockWaitDuration;
@@ -823,6 +826,28 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
}
}
+ @Override
+ public void postOpen(ObserverContext<RegionCoprocessorEnvironment> c) {
+ if (this.disabled) {
+ return;
+ }
+ // Count open regions so the server's replication log groups close only
after its last region
+ // has drained (see ReplicationLogGroup.regionClosed).
+ ReplicationLogGroup.regionOpened(c.getEnvironment().getServerName());
+ this.countedOpen = true;
+ }
+
+ @Override
+ public void postClose(ObserverContext<RegionCoprocessorEnvironment> c,
boolean abortRequested) {
+ // A failed open runs postClose, possibly without postOpen; only undo what
postOpen counted.
+ if (!this.countedOpen) {
+ return;
+ }
+ this.countedOpen = false;
+ ReplicationLogGroup.regionClosed(c.getEnvironment().getServerName(),
+ this.abortable instanceof Server ? (Server) this.abortable : null);
+ }
+
/**
* We use an Increment to serialize the ON DUPLICATE KEY clause so that the
HBase plumbing sets up
* the necessary locks and mvcc to allow an atomic update. The Increment is
not a real increment,
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 a738066e72..5d469f0ea9 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
@@ -46,6 +46,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.EnumSet;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -65,6 +66,7 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Abortable;
import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.Server;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverConstants;
@@ -241,6 +243,9 @@ public class ReplicationLogGroup {
protected static final ConcurrentHashMap<String, ReplicationLogGroup>
INSTANCES =
new ConcurrentHashMap<>();
+ /** Open IndexRegionObserver regions per server, keyed by serverIdentity
(see regionClosed). */
+ private static final Map<String, Integer> OPEN_REGIONS = new HashMap<>();
+
/**
* What a write to this group should do with the replication log.
*/
@@ -1216,10 +1221,57 @@ public class ReplicationLogGroup {
LOG.info("HAGroup {} closed", this);
}
+ /** Records that an IndexRegionObserver region opened on the given server. */
+ public static void regionOpened(ServerName serverName) {
+ synchronized (OPEN_REGIONS) {
+ OPEN_REGIONS.merge(serverIdentity(serverName), 1, Integer::sum);
+ }
+ }
+
/**
- * Close all ReplicationLogGroup instances belonging to the given server.
Called during
- * RegionServer shutdown.
+ * Records that an IndexRegionObserver region closed on the given server,
and closes the server's
+ * groups when that was its last such region and the server is stopped,
stopping or aborted. A
+ * graceful stop sets isStopped before the regions close; cluster shutdown
sets isStopping. HBase
+ * calls postClose after the close write lock is taken, so every batch on
the region, including
+ * its replication sync, has finished. Reaching zero in normal operation
(regions moved away)
+ * leaves the groups open; {@link #closeAllIfNoOpenRegions} closes them on
stop.
*/
+ public static void regionClosed(ServerName serverName, Server server) {
+ String identity = serverIdentity(serverName);
+ synchronized (OPEN_REGIONS) {
+ Integer count = OPEN_REGIONS.get(identity);
+ if (count == null) {
+ return;
+ }
+ if (count > 1) {
+ OPEN_REGIONS.put(identity, count - 1);
+ return;
+ }
+ OPEN_REGIONS.remove(identity);
+ if (server != null && (server.isStopped() || server.isStopping() ||
server.isAborted())) {
+ closeAll(serverName);
+ }
+ }
+ }
+
+ /**
+ * Closes the server's groups now if none of its IndexRegionObserver regions
is open, for example
+ * after the regions were moved away before a graceful stop. Otherwise the
close is left to the
+ * last region's {@link #regionClosed}.
+ */
+ public static void closeAllIfNoOpenRegions(ServerName serverName) {
+ synchronized (OPEN_REGIONS) {
+ Integer count = OPEN_REGIONS.get(serverIdentity(serverName));
+ if (count == null) {
+ closeAll(serverName);
+ } else {
+ LOG.info("Deferring ReplicationLogGroup close for {} until its {} open
regions close",
+ serverName, count);
+ }
+ }
+ }
+
+ /** Closes all ReplicationLogGroup instances belonging to the given server.
*/
public static void closeAll(ServerName serverName) {
LOG.info("Closing all ReplicationLogGroup instances for server {}, total
count={}", serverName,
INSTANCES.size());
diff --git
a/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupIT.java
b/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupIT.java
index 727af83a00..d354c9f797 100644
---
a/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupIT.java
+++
b/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupIT.java
@@ -36,26 +36,39 @@ import java.sql.ResultSet;
import java.sql.Statement;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.Properties;
import java.util.Random;
+import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseTestingUtility;
import org.apache.hadoop.hbase.MiniHBaseCluster;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
+import org.apache.hadoop.hbase.coprocessor.ObserverContext;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessor;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.hadoop.hbase.coprocessor.RegionObserver;
+import org.apache.hadoop.hbase.regionserver.MiniBatchOperationInProgress;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.JVMClusterUtil;
import org.apache.hadoop.hbase.util.Threads;
+import org.apache.hadoop.test.GenericTestUtils;
import org.apache.phoenix.end2end.NeedsOwnMiniClusterTest;
import org.apache.phoenix.hbase.index.IndexRegionObserver;
import org.apache.phoenix.hbase.index.metrics.MetricsHaBypassSourceFactory;
@@ -65,6 +78,8 @@ import org.apache.phoenix.jdbc.FailoverPhoenixConnection;
import org.apache.phoenix.jdbc.PhoenixResultSet;
import org.apache.phoenix.query.QueryConstants;
import org.apache.phoenix.replication.reader.ReplicationLogProcessor;
+import org.apache.phoenix.replication.tool.LogFileAnalyzer;
+import org.apache.phoenix.schema.types.PInteger;
import org.apache.phoenix.util.CDCUtil;
import org.apache.phoenix.util.MetaDataUtil;
import org.apache.phoenix.util.PhoenixRuntime;
@@ -1003,6 +1018,10 @@ public class ReplicationLogGroupIT extends
ReplicationLogGroupBaseIT {
}
// Kill the RS
cluster.killRegionServer(rs2.getRegionServer().getServerName());
+ // A kill skips the region close, so no postClose closes the killed RS's
groups. In a real kill
+ // the process is gone; here the group lives on in the test JVM with an
open writer, so close
+ // it.
+ ReplicationLogGroup.closeAll(sn2);
Threads.sleep(20000); // just to be sure that the kill has fully started.
// Regions will be re-opened and the WAL will be replayed
util.waitUntilAllRegionsAssigned(table);
@@ -1026,6 +1045,218 @@ public class ReplicationLogGroupIT extends
ReplicationLogGroupBaseIT {
indexName);
}
+ /**
+ * Holds one write after its WAL commit (in postBatchMutate, before Phoenix
replicates it in
+ * postBatchMutateIndispensably) until the test releases it.
+ */
+ public static class HoldAfterWalCommitObserver implements RegionCoprocessor,
RegionObserver {
+ static volatile byte[] holdRow;
+ static volatile CountDownLatch held = new CountDownLatch(1);
+ static volatile CountDownLatch release = new CountDownLatch(1);
+
+ @Override
+ public Optional<RegionObserver> getRegionObserver() {
+ return Optional.of(this);
+ }
+
+ @Override
+ public void postBatchMutate(ObserverContext<RegionCoprocessorEnvironment>
c,
+ MiniBatchOperationInProgress<Mutation> miniBatchOp) {
+ byte[] row = holdRow;
+ if (row == null) {
+ return;
+ }
+ for (int i = 0; i < miniBatchOp.size(); i++) {
+ if (Bytes.equals(row, miniBatchOp.getOperation(i).getRow())) {
+ LOG.info("Holding write after WAL commit on {} region {}",
+ c.getEnvironment().getServerName(),
+ c.getEnvironment().getRegionInfo().getEncodedName());
+ held.countDown();
+ try {
+ release.await(120, TimeUnit.SECONDS);
+ LOG.info("Released held write on {}",
c.getEnvironment().getServerName());
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ return;
+ }
+ }
+ }
+ }
+
+ /**
+ * A graceful RegionServer stop with a write already committed to the WAL
but not yet replicated.
+ * The write must still be shipped: if the log groups were closed in preStop
(before the regions),
+ * it would fail replication with "Closed" and never be shipped. Closing
rs2's groups after its
+ * last region closes also writes their log trailers, so the log check
passes without closing them
+ * by hand.
+ */
+ @Test
+ public void testGracefulRegionServerStopClosesLogGroups() throws Exception {
+ HBaseTestingUtility util = CLUSTERS.getHBaseCluster1();
+ MiniHBaseCluster cluster = util.getHBaseCluster();
+ final String tableName = "T_" + generateUniqueName();
+ TableName table = TableName.valueOf(tableName);
+ try (FailoverPhoenixConnection conn = (FailoverPhoenixConnection)
DriverManager
+ .getConnection(CLUSTERS.getJdbcHAUrl(), clientProps)) {
+ conn.createStatement().execute(String.format("create table if not exists
%s (id1 integer "
+ + "not null, val1 varchar constraint pk primary key (id1))",
tableName));
+ conn.commit();
+ }
+ try (Admin admin = util.getConnection().getAdmin()) {
+
admin.modifyTable(TableDescriptorBuilder.newBuilder(admin.getDescriptor(table))
+ .setCoprocessor(HoldAfterWalCommitObserver.class.getName()).build());
+ }
+ JVMClusterUtil.RegionServerThread rs2 = cluster.startRegionServer();
+ ServerName sn2 = rs2.getRegionServer().getServerName();
+ moveRegionToServer(table, sn2);
+
+ final int heldId = 1000;
+ try (FailoverPhoenixConnection conn = (FailoverPhoenixConnection)
DriverManager
+ .getConnection(CLUSTERS.getJdbcHAUrl(), clientProps)) {
+ PreparedStatement stmt = conn.prepareStatement("upsert into " +
tableName + " VALUES(?, ?)");
+ for (int i = 0; i < 10; ++i) {
+ stmt.setInt(1, i);
+ stmt.setString(2, "v");
+ stmt.executeUpdate();
+ conn.commit();
+ }
+ }
+ assertTrue("rs2 should have a replication log group before the stop",
+ ReplicationLogGroup.INSTANCES.values().stream().anyMatch(g ->
g.getServerName().equals(sn2)));
+
+ HoldAfterWalCommitObserver.held = new CountDownLatch(1);
+ HoldAfterWalCommitObserver.release = new CountDownLatch(1);
+ HoldAfterWalCommitObserver.holdRow = PInteger.INSTANCE.toBytes(heldId);
+ Thread writer = new Thread(() -> {
+ try (FailoverPhoenixConnection conn = (FailoverPhoenixConnection)
DriverManager
+ .getConnection(CLUSTERS.getJdbcHAUrl(), clientProps)) {
+ conn.createStatement()
+ .execute("upsert into " + tableName + " VALUES(" + heldId + ",
'held')");
+ conn.commit();
+ } catch (Exception e) {
+ LOG.info("Held upsert failed: {}", e.toString());
+ }
+ }, "held-writer");
+ writer.start();
+ try {
+ assertTrue("The write should be held after its WAL commit",
+ HoldAfterWalCommitObserver.held.await(60, TimeUnit.SECONDS));
+ // stop() runs the RS coprocessors' stop in preStop, then marks the
server stopped; the held
+ // write keeps its region open, so the region close waits for it.
+ cluster.stopRegionServer(sn2);
+ GenericTestUtils.waitFor(() -> rs2.getRegionServer().isStopped(), 100,
60000);
+ } finally {
+ HoldAfterWalCommitObserver.holdRow = null;
+ HoldAfterWalCommitObserver.release.countDown();
+ }
+ writer.join(120000);
+ rs2.join(120000);
+ assertFalse("rs2 should have stopped", rs2.isAlive());
+ assertFalse("rs2's replication log groups should be closed on a graceful
stop",
+ ReplicationLogGroup.INSTANCES.values().stream().anyMatch(g ->
g.getServerName().equals(sn2)));
+ util.waitUntilAllRegionsAssigned(table);
+
+ // Every row committed on cluster-a, acknowledged or not, must be in the
shipped logs.
+ Set<String> committed = new HashSet<>();
+ try (Connection conn =
DriverManager.getConnection(CLUSTERS.getJdbcUrl1(haGroup));
+ ResultSet rs = conn.createStatement().executeQuery("select id1 from " +
tableName)) {
+ while (rs.next()) {
+
committed.add(Bytes.toStringBinary(PInteger.INSTANCE.toBytes(rs.getInt(1))));
+ }
+ }
+ assertTrue("The held write was committed on cluster-a",
+
committed.contains(Bytes.toStringBinary(PInteger.INSTANCE.toBytes(heldId))));
+ logGroup.close();
+ List<Mutation> shippedMutations = groupLogsByTable().get(tableName);
+ Set<String> shipped = new HashSet<>();
+ if (shippedMutations != null) {
+ for (Mutation m : shippedMutations) {
+ shipped.add(Bytes.toStringBinary(m.getRow()));
+ }
+ }
+ Set<String> missing = new HashSet<>(committed);
+ missing.removeAll(shipped);
+ assertTrue("Rows committed on the active but never shipped: " + missing,
missing.isEmpty());
+
+ // rs2 committed the held write, so rs2's own log must carry it. A client
retry through the
+ // reopened region would also ship the row, but from another server's log,
hiding a lost first
+ // attempt.
+ Path root = logGroup.getOrCreatePeerShardManager().getRootDirectoryPath();
+ FileSystem fs = root.getFileSystem(conf2);
+ LogFileAnalyzer analyzer = new LogFileAnalyzer();
+ analyzer.setConf(conf2);
+ Set<String> shippedByRs2 = new HashSet<>();
+ RemoteIterator<LocatedFileStatus> files = fs.listFiles(root, true);
+ while (files.hasNext()) {
+ Path file = files.next().getPath();
+ if (!file.getName().contains(sn2.getServerName())) {
+ continue;
+ }
+ List<Mutation> mutations =
analyzer.groupLogsByTable(file.toString()).get(tableName);
+ if (mutations != null) {
+ for (Mutation m : mutations) {
+ shippedByRs2.add(Bytes.toStringBinary(m.getRow()));
+ }
+ }
+ }
+ assertTrue("rs2 committed the held write but its log does not carry it",
+
shippedByRs2.contains(Bytes.toStringBinary(PInteger.INSTANCE.toBytes(heldId))));
+ }
+
+ /**
+ * A stop after the server's regions were moved away, as a
graceful_stop.sh-style unload does. No
+ * IndexRegionObserver region close happens during the stop, so the stop
itself must close the
+ * server's groups.
+ */
+ @Test
+ public void testRegionServerStopAfterUnloadClosesLogGroups() throws
Exception {
+ HBaseTestingUtility util = CLUSTERS.getHBaseCluster1();
+ MiniHBaseCluster cluster = util.getHBaseCluster();
+ final String tableName = "T_" + generateUniqueName();
+ TableName table = TableName.valueOf(tableName);
+ try (FailoverPhoenixConnection conn = (FailoverPhoenixConnection)
DriverManager
+ .getConnection(CLUSTERS.getJdbcHAUrl(), clientProps)) {
+ conn.createStatement().execute(String.format("create table if not exists
%s (id1 integer "
+ + "not null, val1 varchar constraint pk primary key (id1))",
tableName));
+ conn.commit();
+ }
+ ServerName sn1 = cluster.getRegionServer(0).getServerName();
+ // Keep the balancer from moving another IndexRegionObserver region (e.g.
SYSTEM.CATALOG) onto
+ // rs2, which would make its stop close the groups through regionClosed
instead.
+ boolean balancerWasOn = util.getAdmin().balancerSwitch(false, true);
+ try {
+ JVMClusterUtil.RegionServerThread rs2 = cluster.startRegionServer();
+ ServerName sn2 = rs2.getRegionServer().getServerName();
+ moveRegionToServer(table, sn2);
+ try (FailoverPhoenixConnection conn = (FailoverPhoenixConnection)
DriverManager
+ .getConnection(CLUSTERS.getJdbcHAUrl(), clientProps)) {
+ conn.createStatement().execute("upsert into " + tableName + "
VALUES(1, 'v')");
+ conn.commit();
+ }
+ assertTrue("rs2 should have a replication log group before the unload",
+ ReplicationLogGroup.INSTANCES.values().stream()
+ .anyMatch(g -> g.getServerName().equals(sn2)));
+
+ moveRegionToServer(table, sn1);
+ assertTrue("rs2 must host no IndexRegionObserver region before the stop",
+ rs2.getRegionServer().getRegions().stream().noneMatch(
+ r ->
r.getCoprocessorHost().findCoprocessor(IndexRegionObserver.class) != null));
+ assertTrue("Regions moving away while running must leave rs2's groups
open",
+ ReplicationLogGroup.INSTANCES.values().stream()
+ .anyMatch(g -> g.getServerName().equals(sn2)));
+
+ cluster.stopRegionServer(sn2);
+ rs2.join(120000);
+ assertFalse("rs2 should have stopped", rs2.isAlive());
+ assertFalse("rs2's replication log groups should be closed by the stop",
+ ReplicationLogGroup.INSTANCES.values().stream()
+ .anyMatch(g -> g.getServerName().equals(sn2)));
+ } finally {
+ util.getAdmin().balancerSwitch(balancerWasOn, true);
+ }
+ }
+
/**
* End-to-end replication of the tenant-owned SYSTEM.CATALOG and
SYSTEM.CHILD_LINK rows that back
* a tenant view. Models the intended HA topology: the base table and the
global view are global
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 1e0854f2f3..20aae6b679 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
@@ -67,6 +67,7 @@ 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.Server;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.phoenix.jdbc.ClusterType;
@@ -992,6 +993,140 @@ public class ReplicationLogGroupTest extends
ReplicationLogBaseTest {
verify(logGroup, never()).abort(anyString(), any());
}
+ /** Registers an initialized group for a distinct server so closeAll can
reach it. */
+ private ReplicationLogGroup registerGroupForServer(ServerName sn, String
group) throws Exception {
+ ReplicationLogGroup g =
+ new TestableLogGroup(conf, sn, group, haGroupStoreManager,
useAlignedRotation());
+ g.init();
+ ReplicationLogGroup.cacheForTesting(g);
+ return g;
+ }
+
+ private static Server server(boolean stopped, boolean stopping, boolean
aborted) {
+ Server server = mock(Server.class);
+ doReturn(stopped).when(server).isStopped();
+ doReturn(stopping).when(server).isStopping();
+ doReturn(aborted).when(server).isAborted();
+ return server;
+ }
+
+ private static ServerName uniqueServer(int port) {
+ return ServerName.valueOf("rc-test", port, System.nanoTime());
+ }
+
+ @Test
+ public void testGroupsCloseOnLastRegionCloseWhileStopping() throws Exception
{
+ ServerName sn = uniqueServer(61001);
+ ReplicationLogGroup g = registerGroupForServer(sn, haGroupName + "-rc1");
+ try {
+ Server stopping = server(true, false, false);
+ ReplicationLogGroup.regionOpened(sn);
+ ReplicationLogGroup.regionOpened(sn);
+ ReplicationLogGroup.regionClosed(sn, stopping);
+ assertFalse("Groups must stay open while a region is still open",
g.isClosed());
+ ReplicationLogGroup.regionClosed(sn, stopping);
+ assertTrue("Groups must close when the last region closes while
stopping", g.isClosed());
+ } finally {
+ g.close();
+ }
+ }
+
+ @Test
+ public void testGroupsStayOpenWhenLastRegionClosesWhileRunning() throws
Exception {
+ ServerName sn = uniqueServer(61002);
+ ReplicationLogGroup g = registerGroupForServer(sn, haGroupName + "-rc2");
+ try {
+ ReplicationLogGroup.regionOpened(sn);
+ ReplicationLogGroup.regionClosed(sn, server(false, false, false));
+ assertFalse("Regions moving away while running must not close the
groups", g.isClosed());
+ } finally {
+ g.close();
+ }
+ }
+
+ @Test
+ public void testGroupsCloseOnLastRegionCloseWhileAbortingOrClusterStopping()
throws Exception {
+ for (Server s : new Server[] { server(false, false, true), server(false,
true, false) }) {
+ ServerName sn = uniqueServer(61003);
+ ReplicationLogGroup g = registerGroupForServer(sn, haGroupName + "-rc3");
+ try {
+ ReplicationLogGroup.regionOpened(sn);
+ ReplicationLogGroup.regionClosed(sn, s);
+ assertTrue(g.isClosed());
+ } finally {
+ g.close();
+ }
+ }
+ }
+
+ @Test
+ public void testRegionCloseWithoutOpenDoesNotGoNegative() throws Exception {
+ ServerName sn = uniqueServer(61004);
+ ReplicationLogGroup g = registerGroupForServer(sn, haGroupName + "-rc4");
+ try {
+ Server stopping = server(true, false, false);
+ ReplicationLogGroup.regionClosed(sn, stopping);
+ assertFalse("A close with no open region must not close the groups",
g.isClosed());
+ ReplicationLogGroup.regionOpened(sn);
+ ReplicationLogGroup.regionOpened(sn);
+ ReplicationLogGroup.regionClosed(sn, stopping);
+ assertFalse("The stray close must not have consumed a real region's
count", g.isClosed());
+ ReplicationLogGroup.regionClosed(sn, stopping);
+ assertTrue(g.isClosed());
+ } finally {
+ g.close();
+ }
+ }
+
+ @Test
+ public void testRegionCountsAreSeparatePerServer() throws Exception {
+ ServerName sn1 = uniqueServer(61005);
+ ServerName sn2 = uniqueServer(61006);
+ ReplicationLogGroup g1 = registerGroupForServer(sn1, haGroupName +
"-rc5a");
+ ReplicationLogGroup g2 = registerGroupForServer(sn2, haGroupName +
"-rc5b");
+ try {
+ ReplicationLogGroup.regionOpened(sn1);
+ ReplicationLogGroup.regionOpened(sn2);
+ ReplicationLogGroup.regionClosed(sn1, server(true, false, false));
+ assertTrue(g1.isClosed());
+ assertFalse("Another server's groups must stay open", g2.isClosed());
+ } finally {
+ g1.close();
+ g2.close();
+ }
+ }
+
+ @Test
+ public void testStopClosesGroupsWhenNoRegionIsOpen() throws Exception {
+ ServerName sn = uniqueServer(61007);
+ ReplicationLogGroup g = registerGroupForServer(sn, haGroupName + "-rc6");
+ try {
+ // Regions moved away while running (e.g. unloaded before a graceful
stop).
+ ReplicationLogGroup.regionOpened(sn);
+ ReplicationLogGroup.regionClosed(sn, server(false, false, false));
+ assertFalse(g.isClosed());
+ ReplicationLogGroup.closeAllIfNoOpenRegions(sn);
+ assertTrue("Stop must close the groups when no region is left to do it",
g.isClosed());
+ } finally {
+ g.close();
+ }
+ }
+
+ @Test
+ public void testStopDefersCloseWhileARegionIsOpen() throws Exception {
+ ServerName sn = uniqueServer(61008);
+ ReplicationLogGroup g = registerGroupForServer(sn, haGroupName + "-rc7");
+ try {
+ ReplicationLogGroup.regionOpened(sn);
+ ReplicationLogGroup.closeAllIfNoOpenRegions(sn);
+ assertFalse("Stop must not close the groups while a region can still
write", g.isClosed());
+ ReplicationLogGroup.regionClosed(sn, server(true, false, false));
+ assertTrue(g.isClosed());
+ } finally {
+ g.close();
+ }
+ }
+
/**
* Tests rotation during an in-flight batch. When a pending writer is staged
by the rotation task
* while appends are in flight, the sync's apply() drains the pending
writer, replays the 5
