From 7e9baa2eb2795b4ef14d7995614b8371c816fa72 Mon Sep 17 00:00:00 2001 From: XinSun Date: Sun, 20 Sep 2020 10:54:43 +0800 Subject: [PATCH] =?UTF-8?q?HBASE-24684=20Fetch=20ReplicationSink=20servers?= =?UTF-8?q?=20list=20from=20HMaster=20instead=20o=E2=80=A6=20(#2077)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Wellington Chevreuil --- .../main/protobuf/server/master/Master.proto | 12 +- .../hbase/coprocessor/MasterObserver.java | 16 ++ .../apache/hadoop/hbase/master/HMaster.java | 5 + .../hbase/master/MasterCoprocessorHost.java | 18 ++ .../hbase/master/MasterRpcServices.java | 23 +++ .../hadoop/hbase/master/MasterServices.java | 6 + .../replication/HBaseReplicationEndpoint.java | 170 ++++++++++++++++-- .../hbase/master/MockNoopMasterServices.java | 5 + .../TestReplicationFetchServers.java | 106 +++++++++++ .../TestGlobalReplicationThrottler.java | 4 + ...ionReplicaReplicationEndpointNoMaster.java | 2 + 11 files changed, 347 insertions(+), 20 deletions(-) create mode 100644 hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationFetchServers.java diff --git a/hbase-protocol-shaded/src/main/protobuf/server/master/Master.proto b/hbase-protocol-shaded/src/main/protobuf/server/master/Master.proto index 58ab01d980e6..48180274689e 100644 --- a/hbase-protocol-shaded/src/main/protobuf/server/master/Master.proto +++ b/hbase-protocol-shaded/src/main/protobuf/server/master/Master.proto @@ -693,6 +693,13 @@ message SwitchExceedThrottleQuotaResponse { required bool previous_exceed_throttle_quota_enabled = 1; } +message ListReplicationSinkServersRequest { +} + +message ListReplicationSinkServersResponse { + repeated ServerName server_name = 1; +} + service MasterService { /** Used by the client to get the number of regions that have received the updated schema */ rpc GetSchemaAlterStatus(GetSchemaAlterStatusRequest) @@ -1122,7 +1129,10 @@ service MasterService { returns (RenameRSGroupResponse); rpc UpdateRSGroupConfig(UpdateRSGroupConfigRequest) - returns (UpdateRSGroupConfigResponse); + returns (UpdateRSGroupConfigResponse); + + rpc ListReplicationSinkServers(ListReplicationSinkServersRequest) + returns (ListReplicationSinkServersResponse); } // HBCK Service definitions. diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/MasterObserver.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/MasterObserver.java index ffe4f53ac66f..0c5bb0d49411 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/MasterObserver.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/MasterObserver.java @@ -1784,4 +1784,20 @@ default void preHasUserPermissions(ObserverContext default void postHasUserPermissions(ObserverContext ctx, String userName, List permissions) throws IOException { } + + /** + * Called before getting servers for replication sink. + * @param ctx the coprocessor instance's environment + */ + default void preListReplicationSinkServers(ObserverContext ctx) + throws IOException { + } + + /** + * Called after getting servers for replication sink. + * @param ctx the coprocessor instance's environment + */ + default void postListReplicationSinkServers(ObserverContext ctx) + throws IOException { + } } diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java index 520d9d52cb07..4ad685772a0a 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java @@ -3907,4 +3907,9 @@ public MetaRegionLocationCache getMetaRegionLocationCache() { public RSGroupInfoManager getRSGroupInfoManager() { return rsGroupInfoManager; } + + @Override + public List listReplicationSinkServers() throws IOException { + return this.serverManager.getOnlineServersList(); + } } diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/MasterCoprocessorHost.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/MasterCoprocessorHost.java index 28c9c3090911..3de4cac82f2b 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/MasterCoprocessorHost.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/MasterCoprocessorHost.java @@ -2039,4 +2039,22 @@ public void call(MasterObserver observer) throws IOException { } }); } + + public void preListReplicationSinkServers() throws IOException { + execOperation(coprocEnvironments.isEmpty() ? null : new MasterObserverOperation() { + @Override + public void call(MasterObserver observer) throws IOException { + observer.preListReplicationSinkServers(this); + } + }); + } + + public void postListReplicationSinkServers() throws IOException { + execOperation(coprocEnvironments.isEmpty() ? null : new MasterObserverOperation() { + @Override + public void call(MasterObserver observer) throws IOException { + observer.postListReplicationSinkServers(this); + } + }); + } } diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/MasterRpcServices.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/MasterRpcServices.java index 46c887b24ffc..fd388843766d 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/MasterRpcServices.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/MasterRpcServices.java @@ -255,6 +255,8 @@ import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.ListNamespaceDescriptorsResponse; import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.ListNamespacesRequest; import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.ListNamespacesResponse; +import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.ListReplicationSinkServersRequest; +import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.ListReplicationSinkServersResponse; import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.ListTableDescriptorsByNamespaceRequest; import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.ListTableDescriptorsByNamespaceResponse; import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.ListTableNamesByNamespaceRequest; @@ -3321,4 +3323,25 @@ public UpdateRSGroupConfigResponse updateRSGroupConfig(RpcController controller, } return builder.build(); } + + @Override + public ListReplicationSinkServersResponse listReplicationSinkServers( + RpcController controller, ListReplicationSinkServersRequest request) + throws ServiceException { + ListReplicationSinkServersResponse.Builder builder = + ListReplicationSinkServersResponse.newBuilder(); + try { + if (master.getMasterCoprocessorHost() != null) { + master.getMasterCoprocessorHost().preListReplicationSinkServers(); + } + builder.addAllServerName(master.listReplicationSinkServers().stream() + .map(ProtobufUtil::toServerName).collect(Collectors.toList())); + if (master.getMasterCoprocessorHost() != null) { + master.getMasterCoprocessorHost().postListReplicationSinkServers(); + } + } catch (IOException e) { + throw new ServiceException(e); + } + return builder.build(); + } } diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/MasterServices.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/MasterServices.java index 908d21270c6e..7b212892e288 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/MasterServices.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/MasterServices.java @@ -553,4 +553,10 @@ default SplitWALManager getSplitWALManager(){ * @return The state of the load balancer, or false if the load balancer isn't defined. */ boolean isBalancerOn(); + + /** + * Get a list of servers' addresses for replication sink. + * @return a list of servers' address + */ + List listReplicationSinkServers() throws IOException; } diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/HBaseReplicationEndpoint.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/HBaseReplicationEndpoint.java index 3cde0d5113a0..85a1eee262d5 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/HBaseReplicationEndpoint.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/HBaseReplicationEndpoint.java @@ -18,19 +18,31 @@ package org.apache.hadoop.hbase.replication; +import static org.apache.hadoop.hbase.HConstants.DEFAULT_HBASE_RPC_SHORTOPERATION_TIMEOUT; +import static org.apache.hadoop.hbase.HConstants.HBASE_RPC_SHORTOPERATION_TIMEOUT_KEY; + import java.io.IOException; import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.UUID; -import org.apache.hadoop.hbase.zookeeper.ZKListener; -import org.apache.yetus.audience.InterfaceAudience; +import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.Abortable; +import org.apache.hadoop.hbase.ChoreService; +import org.apache.hadoop.hbase.ScheduledChore; +import org.apache.hadoop.hbase.Server; import org.apache.hadoop.hbase.ServerName; +import org.apache.hadoop.hbase.client.AsyncClusterConnection; +import org.apache.hadoop.hbase.client.ClusterConnectionFactory; +import org.apache.hadoop.hbase.security.User; +import org.apache.hadoop.hbase.security.UserProvider; +import org.apache.hadoop.hbase.util.FutureUtils; import org.apache.hadoop.hbase.zookeeper.ZKClusterId; +import org.apache.hadoop.hbase.zookeeper.ZKListener; import org.apache.hadoop.hbase.zookeeper.ZKUtil; import org.apache.hadoop.hbase.zookeeper.ZKWatcher; +import org.apache.yetus.audience.InterfaceAudience; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.KeeperException.AuthFailedException; import org.apache.zookeeper.KeeperException.ConnectionLossException; @@ -38,6 +50,13 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.apache.hbase.thirdparty.com.google.protobuf.ServiceException; + +import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil; +import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.ListReplicationSinkServersRequest; +import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.ListReplicationSinkServersResponse; +import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.MasterService; + /** * A {@link BaseReplicationEndpoint} for replication endpoints whose * target cluster is an HBase cluster. @@ -48,15 +67,39 @@ public abstract class HBaseReplicationEndpoint extends BaseReplicationEndpoint private static final Logger LOG = LoggerFactory.getLogger(HBaseReplicationEndpoint.class); + public static final String FETCH_SERVERS_USE_ZK_CONF_KEY = + "hbase.replication.fetch.servers.usezk"; + + public static final String FETCH_SERVERS_INTERVAL_CONF_KEY = + "hbase.replication.fetch.servers.interval"; + public static final int DEFAULT_FETCH_SERVERS_INTERVAL = 10 * 60 * 1000; // 10 mins + private ZKWatcher zkw = null; private List regionServers = new ArrayList<>(0); private long lastRegionServerUpdate; + private AsyncClusterConnection peerConnection; + private boolean fetchServersUseZk = false; + private FetchServersChore fetchServersChore; + private int shortOperationTimeout; protected synchronized void disconnect() { if (zkw != null) { zkw.close(); } + if (fetchServersChore != null) { + ChoreService choreService = ctx.getServer().getChoreService(); + if (null != choreService) { + choreService.cancelChore(fetchServersChore); + } + } + if (peerConnection != null) { + try { + peerConnection.close(); + } catch (IOException e) { + LOG.warn("Attempt to close peerConnection failed.", e); + } + } } /** @@ -87,8 +130,27 @@ public void stop() { } @Override - protected void doStart() { + protected synchronized void doStart() { + this.shortOperationTimeout = ctx.getLocalConfiguration().getInt( + HBASE_RPC_SHORTOPERATION_TIMEOUT_KEY, DEFAULT_HBASE_RPC_SHORTOPERATION_TIMEOUT); try { + if (ctx.getLocalConfiguration().getBoolean(FETCH_SERVERS_USE_ZK_CONF_KEY, false)) { + fetchServersUseZk = true; + } else { + try { + if (ReplicationUtils.isPeerClusterSupportReplicationOffload(getPeerConnection())) { + fetchServersChore = new FetchServersChore(ctx.getServer(), this); + ctx.getServer().getChoreService().scheduleChore(fetchServersChore); + fetchServersUseZk = false; + } else { + fetchServersUseZk = true; + } + } catch (Throwable t) { + fetchServersUseZk = true; + LOG.warn("Peer {} try to fetch servers by admin failed. Using zk impl.", + ctx.getPeerId(), t); + } + } reloadZkWatcher(); notifyStarted(); } catch (IOException e) { @@ -130,10 +192,14 @@ protected synchronized ZKWatcher getZkw() { * @throws IOException If anything goes wrong connecting */ synchronized void reloadZkWatcher() throws IOException { - if (zkw != null) zkw.close(); + if (zkw != null) { + zkw.close(); + } zkw = new ZKWatcher(ctx.getConfiguration(), "connection to cluster: " + ctx.getPeerId(), this); - getZkw().registerListener(new PeerRegionServerListener(this)); + if (fetchServersUseZk) { + getZkw().registerListener(new PeerRegionServerListener(this)); + } } @Override @@ -148,15 +214,48 @@ public boolean isAborted() { return false; } + /** + * Get the connection to peer cluster + * @return connection to peer cluster + * @throws IOException If anything goes wrong connecting + */ + protected synchronized AsyncClusterConnection getPeerConnection() throws IOException { + if (peerConnection == null) { + Configuration conf = ctx.getConfiguration(); + peerConnection = ClusterConnectionFactory.createAsyncClusterConnection(conf, null, + UserProvider.instantiate(conf).getCurrent()); + } + return peerConnection; + } + + /** + * Get the list of all the servers that are responsible for replication sink + * from the specified peer master + * @return list of server addresses or an empty list if the slave is unavailable + */ + protected List fetchSlavesAddresses() throws IOException { + AsyncClusterConnection peerConn = getPeerConnection(); + ServerName master = FutureUtils.get(peerConn.getAdmin().getMaster()); + MasterService.BlockingInterface masterStub = MasterService.newBlockingStub( + peerConn.getRpcClient() + .createBlockingRpcChannel(master, User.getCurrent(), shortOperationTimeout)); + try { + ListReplicationSinkServersResponse resp = masterStub.listReplicationSinkServers(null, + ListReplicationSinkServersRequest.newBuilder().build()); + return ProtobufUtil.toServerNameList(resp.getServerNameList()); + } catch (ServiceException se) { + throw ProtobufUtil.getRemoteException(se); + } + } + /** * Get the list of all the region servers from the specified peer - * @param zkw zk connection to use * @return list of region server addresses or an empty list if the slave is unavailable */ - protected static List fetchSlavesAddresses(ZKWatcher zkw) - throws KeeperException { - List children = ZKUtil.listChildrenAndWatchForNewChildren(zkw, - zkw.getZNodePaths().rsZNode); + protected List fetchSlavesAddressesByZK() throws KeeperException { + ZKWatcher zk = getZkw(); + List children = ZKUtil.listChildrenAndWatchForNewChildren(zk, + zk.getZNodePaths().rsZNode); if (children == null) { return Collections.emptyList(); } @@ -168,8 +267,9 @@ protected static List fetchSlavesAddresses(ZKWatcher zkw) } /** - * Get a list of all the addresses of all the available region servers - * for this peer cluster, or an empty list if no region servers available at peer cluster. + * Get a list of all the addresses of all the available servers that are responsible for + * replication sink for this peer cluster, or an empty list if no servers available at peer + * cluster. * @return list of addresses */ // Synchronize peer cluster connection attempts to avoid races and rate @@ -177,13 +277,21 @@ protected static List fetchSlavesAddresses(ZKWatcher zkw) // the peer cluster. If the peer cluster is down we can get out of control // over time. public synchronized List getRegionServers() { - try { - setRegionServers(fetchSlavesAddresses(this.getZkw())); - } catch (KeeperException ke) { - if (LOG.isDebugEnabled()) { - LOG.debug("Fetch slaves addresses failed", ke); + if (fetchServersUseZk) { + try { + setRegionServers(fetchSlavesAddressesByZK()); + } catch (KeeperException ke) { + if (LOG.isDebugEnabled()) { + LOG.debug("Fetch slaves addresses failed", ke); + } + reconnect(ke); + } + } else { + try { + setRegionServers(fetchSlavesAddresses()); + } catch (IOException e) { + LOG.warn("Fetch slaves addresses failed", e); } - reconnect(ke); } return regionServers; } @@ -225,11 +333,35 @@ public synchronized void nodeChildrenChanged(String path) { if (path.equals(regionServerListNode)) { try { LOG.info("Detected change to peer region servers, fetching updated list"); - replicationEndpoint.setRegionServers(fetchSlavesAddresses(replicationEndpoint.getZkw())); + replicationEndpoint.setRegionServers(replicationEndpoint.fetchSlavesAddressesByZK()); } catch (KeeperException e) { LOG.error("Error reading slave addresses", e); } } } } + + /** + * Chore that will fetch the list of servers from peer master. + */ + public static class FetchServersChore extends ScheduledChore { + + private HBaseReplicationEndpoint endpoint; + + public FetchServersChore(Server server, HBaseReplicationEndpoint endpoint) { + super("Peer-" + endpoint.ctx.getPeerId() + "-FetchServersChore", server, + server.getConfiguration().getInt(FETCH_SERVERS_INTERVAL_CONF_KEY, + DEFAULT_FETCH_SERVERS_INTERVAL)); + this.endpoint = endpoint; + } + + @Override + protected void chore() { + try { + endpoint.setRegionServers(endpoint.fetchSlavesAddresses()); + } catch (Throwable t) { + LOG.error("Peer {} fetches servers failed", endpoint.ctx.getPeerId(), t); + } + } + } } diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/MockNoopMasterServices.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/MockNoopMasterServices.java index 7c65005de55d..947ba5d7281a 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/MockNoopMasterServices.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/MockNoopMasterServices.java @@ -507,4 +507,9 @@ public RSGroupInfoManager getRSGroupInfoManager() { public boolean isBalancerOn() { return false; } + + @Override + public List listReplicationSinkServers() throws IOException { + return null; + } } diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationFetchServers.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationFetchServers.java new file mode 100644 index 000000000000..9ceaceec6c19 --- /dev/null +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationFetchServers.java @@ -0,0 +1,106 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.hadoop.hbase.replication; + +import static org.apache.hadoop.hbase.coprocessor.CoprocessorHost.MASTER_COPROCESSOR_CONF_KEY; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import java.io.IOException; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.apache.hadoop.hbase.HBaseClassTestRule; +import org.apache.hadoop.hbase.ServerName; +import org.apache.hadoop.hbase.client.AsyncClusterConnection; +import org.apache.hadoop.hbase.client.Get; +import org.apache.hadoop.hbase.client.Put; +import org.apache.hadoop.hbase.coprocessor.MasterCoprocessor; +import org.apache.hadoop.hbase.coprocessor.MasterCoprocessorEnvironment; +import org.apache.hadoop.hbase.coprocessor.MasterObserver; +import org.apache.hadoop.hbase.coprocessor.ObserverContext; +import org.apache.hadoop.hbase.security.User; +import org.apache.hadoop.hbase.testclassification.MediumTests; +import org.apache.hadoop.hbase.testclassification.ReplicationTests; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.experimental.categories.Category; + +import org.apache.hbase.thirdparty.com.google.protobuf.ServiceException; + +import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil; +import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.ListReplicationSinkServersRequest; +import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.ListReplicationSinkServersResponse; +import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.MasterService; + +@Category({ ReplicationTests.class, MediumTests.class }) +public class TestReplicationFetchServers extends TestReplicationBase { + + @ClassRule + public static final HBaseClassTestRule CLASS_RULE = + HBaseClassTestRule.forClass(TestReplicationFetchServers.class); + + private static AtomicBoolean fetchFlag = new AtomicBoolean(false); + + public static class MyObserver implements MasterCoprocessor, MasterObserver { + + @Override + public Optional getMasterObserver() { + return Optional.of(this); + } + + @Override + public void postListReplicationSinkServers(ObserverContext ctx) { + fetchFlag.set(true); + } + } + + @BeforeClass + public static void setUpBeforeClass() throws Exception { + CONF2.set(MASTER_COPROCESSOR_CONF_KEY, MyObserver.class.getName()); + TestReplicationBase.setUpBeforeClass(); + } + + @Before + public void beforeMethod() { + fetchFlag.set(false); + } + + @Test + public void testMasterListReplicationPeerServers() throws IOException, ServiceException { + AsyncClusterConnection conn = UTIL2.getAsyncConnection(); + ServerName master = UTIL2.getAdmin().getMaster(); + MasterService.BlockingInterface masterStub = MasterService.newBlockingStub( + conn.getRpcClient().createBlockingRpcChannel(master, User.getCurrent(), 1000)); + ListReplicationSinkServersResponse resp = masterStub.listReplicationSinkServers( + null, ListReplicationSinkServersRequest.newBuilder().build()); + List servers = ProtobufUtil.toServerNameList(resp.getServerNameList()); + assertFalse(servers.isEmpty()); + assertTrue(fetchFlag.get()); + } + + @Test + public void testPutData() throws IOException { + htable1.put(new Put(row).addColumn(famName, famName, row)); + UTIL2.waitFor(30000L, () -> !htable2.get(new Get(row)).isEmpty()); + assertTrue(fetchFlag.get()); + } +} diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestGlobalReplicationThrottler.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestGlobalReplicationThrottler.java index 1538fa360093..cfc9fa3652e7 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestGlobalReplicationThrottler.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestGlobalReplicationThrottler.java @@ -118,6 +118,10 @@ public static void setUpBeforeClass() throws Exception { @AfterClass public static void tearDownAfterClass() throws Exception { + Admin admin1 = utility1.getAdmin(); + admin1.removeReplicationPeer("peer1"); + admin1.removeReplicationPeer("peer2"); + admin1.removeReplicationPeer("peer3"); utility2.shutdownMiniCluster(); utility1.shutdownMiniCluster(); } diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestRegionReplicaReplicationEndpointNoMaster.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestRegionReplicaReplicationEndpointNoMaster.java index ee1ae5f380d5..c676e3096105 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestRegionReplicaReplicationEndpointNoMaster.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestRegionReplicaReplicationEndpointNoMaster.java @@ -256,11 +256,13 @@ public void testRegionReplicaReplicationEndpointReplicate() throws Exception { ReplicationEndpoint.Context context = mock(ReplicationEndpoint.Context.class); when(context.getConfiguration()).thenReturn(HTU.getConfiguration()); + when(context.getLocalConfiguration()).thenReturn(HTU.getConfiguration()); when(context.getMetrics()).thenReturn(mock(MetricsSource.class)); when(context.getServer()).thenReturn(rs0); when(context.getTableDescriptors()).thenReturn(rs0.getTableDescriptors()); replicator.init(context); replicator.startAsync(); + HTU.waitFor(30000, replicator::isRunning); //load some data to primary HTU.loadNumericRows(table, f, 0, 1000);