Apache/hbase
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HBASE-16499 slow replication for small HBase clusters

Browse Source 
Signed-off-by: Ashish Singhi <ashishsinghi@apache.org>
This commit is contained in:
Ashish Singhi 2018-04-04 09:59:50 +05:30
parent ed21f26171
commit 9a34880724
2 changed files with 21 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSinkManager.java
View File

@ -58,7 +58,7 @@ public class ReplicationSinkManager {
* Default ratio of the total number of peer cluster region servers to consider * Default ratio of the total number of peer cluster region servers to consider
* replicating to. * replicating to.
*/ */
static final float DEFAULT_REPLICATION_SOURCE_RATIO = 0.1f; static final float DEFAULT_REPLICATION_SOURCE_RATIO = 0.5f;
private final Connection conn; private final Connection conn;

36
hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSinkManager.java
View File

@ -27,7 +27,6 @@ import org.apache.hadoop.hbase.HBaseClassTestRule;
import org.apache.hadoop.hbase.ServerName; import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.client.ClusterConnection; import org.apache.hadoop.hbase.client.ClusterConnection;
import org.apache.hadoop.hbase.replication.HBaseReplicationEndpoint; import org.apache.hadoop.hbase.replication.HBaseReplicationEndpoint;
import org.apache.hadoop.hbase.replication.ReplicationPeers;
import org.apache.hadoop.hbase.replication.regionserver.ReplicationSinkManager.SinkPeer; import org.apache.hadoop.hbase.replication.regionserver.ReplicationSinkManager.SinkPeer;
import org.apache.hadoop.hbase.testclassification.ReplicationTests; import org.apache.hadoop.hbase.testclassification.ReplicationTests;
import org.apache.hadoop.hbase.testclassification.SmallTests; import org.apache.hadoop.hbase.testclassification.SmallTests;
@ -49,13 +48,11 @@ public class TestReplicationSinkManager {
private static final String PEER_CLUSTER_ID = "PEER_CLUSTER_ID"; private static final String PEER_CLUSTER_ID = "PEER_CLUSTER_ID";
private ReplicationPeers replicationPeers;
private HBaseReplicationEndpoint replicationEndpoint; private HBaseReplicationEndpoint replicationEndpoint;
private ReplicationSinkManager sinkManager; private ReplicationSinkManager sinkManager;
@Before @Before
public void setUp() { public void setUp() {
replicationPeers = mock(ReplicationPeers.class);
replicationEndpoint = mock(HBaseReplicationEndpoint.class); replicationEndpoint = mock(HBaseReplicationEndpoint.class);
sinkManager = new ReplicationSinkManager(mock(ClusterConnection.class), sinkManager = new ReplicationSinkManager(mock(ClusterConnection.class),
PEER_CLUSTER_ID, replicationEndpoint, new Configuration()); PEER_CLUSTER_ID, replicationEndpoint, new Configuration());
@ -64,7 +61,8 @@ public class TestReplicationSinkManager {
@Test @Test
public void testChooseSinks() { public void testChooseSinks() {
List<ServerName> serverNames = Lists.newArrayList(); List<ServerName> serverNames = Lists.newArrayList();
for (int i = 0; i < 20; i++) { int totalServers = 20;
for (int i = 0; i < totalServers; i++) {
serverNames.add(mock(ServerName.class)); serverNames.add(mock(ServerName.class));
} }
@ -73,7 +71,8 @@ public class TestReplicationSinkManager {
sinkManager.chooseSinks(); sinkManager.chooseSinks();
assertEquals(2, sinkManager.getNumSinks()); int expected = (int) (totalServers * ReplicationSinkManager.DEFAULT_REPLICATION_SOURCE_RATIO);
assertEquals(expected, sinkManager.getNumSinks());
} }
@ -117,7 +116,8 @@ public class TestReplicationSinkManager {
@Test @Test
public void testReportBadSink_PastThreshold() { public void testReportBadSink_PastThreshold() {
List<ServerName> serverNames = Lists.newArrayList(); List<ServerName> serverNames = Lists.newArrayList();
for (int i = 0; i < 30; i++) { int totalServers = 30;
for (int i = 0; i < totalServers; i++) {
serverNames.add(mock(ServerName.class)); serverNames.add(mock(ServerName.class));
} }
when(replicationEndpoint.getRegionServers()) when(replicationEndpoint.getRegionServers())
@ -126,7 +126,8 @@ public class TestReplicationSinkManager {
sinkManager.chooseSinks(); sinkManager.chooseSinks();
// Sanity check // Sanity check
assertEquals(3, sinkManager.getNumSinks()); int expected = (int) (totalServers * ReplicationSinkManager.DEFAULT_REPLICATION_SOURCE_RATIO);
assertEquals(expected, sinkManager.getNumSinks());
ServerName serverName = sinkManager.getSinksForTesting().get(0); ServerName serverName = sinkManager.getSinksForTesting().get(0);
@ -139,7 +140,7 @@ public class TestReplicationSinkManager {
// Reporting a bad sink more than the threshold count should remove it // Reporting a bad sink more than the threshold count should remove it
// from the list of potential sinks // from the list of potential sinks
assertEquals(2, sinkManager.getNumSinks()); assertEquals(expected - 1, sinkManager.getNumSinks());
// //
// now try a sink that has some successes // now try a sink that has some successes
@ -154,23 +155,24 @@ public class TestReplicationSinkManager {
sinkManager.reportBadSink(sinkPeer); sinkManager.reportBadSink(sinkPeer);
// did not remove the sink, since we had one successful try // did not remove the sink, since we had one successful try
assertEquals(2, sinkManager.getNumSinks()); assertEquals(expected - 1, sinkManager.getNumSinks());
for (int i = 0; i <= ReplicationSinkManager.DEFAULT_BAD_SINK_THRESHOLD-2; i++) { for (int i = 0; i <= ReplicationSinkManager.DEFAULT_BAD_SINK_THRESHOLD-2; i++) {
sinkManager.reportBadSink(sinkPeer); sinkManager.reportBadSink(sinkPeer);
} }
// still not remove, since the success reset the counter // still not remove, since the success reset the counter
assertEquals(2, sinkManager.getNumSinks()); assertEquals(expected - 1, sinkManager.getNumSinks());
sinkManager.reportBadSink(sinkPeer); sinkManager.reportBadSink(sinkPeer);
// but we exhausted the tries // but we exhausted the tries
assertEquals(1, sinkManager.getNumSinks()); assertEquals(expected - 2, sinkManager.getNumSinks());
} }
@Test @Test
public void testReportBadSink_DownToZeroSinks() { public void testReportBadSink_DownToZeroSinks() {
List<ServerName> serverNames = Lists.newArrayList(); List<ServerName> serverNames = Lists.newArrayList();
for (int i = 0; i < 20; i++) { int totalServers = 4;
for (int i = 0; i < totalServers; i++) {
serverNames.add(mock(ServerName.class)); serverNames.add(mock(ServerName.class));
} }
when(replicationEndpoint.getRegionServers()) when(replicationEndpoint.getRegionServers())
@ -179,9 +181,9 @@ public class TestReplicationSinkManager {
sinkManager.chooseSinks(); sinkManager.chooseSinks();
// Sanity check // Sanity check
List<ServerName> sinkList = sinkManager.getSinksForTesting(); List<ServerName> sinkList = sinkManager.getSinksForTesting();
assertEquals(2, sinkList.size()); int expected = (int) (totalServers * ReplicationSinkManager.DEFAULT_REPLICATION_SOURCE_RATIO);
assertEquals(expected, sinkList.size());
ServerName serverNameA = sinkList.get(0); ServerName serverNameA = sinkList.get(0);
ServerName serverNameB = sinkList.get(1); ServerName serverNameB = sinkList.get(1);
@ -195,8 +197,10 @@ public class TestReplicationSinkManager {
} }
// We've gone down to 0 good sinks, so the replication sinks // We've gone down to 0 good sinks, so the replication sinks
// should have been refreshed now // should have been refreshed now, so out of 4 servers, 2 are not considered as they are
assertEquals(2, sinkManager.getNumSinks()); // reported as bad.
expected = (int) ((totalServers - 2) * ReplicationSinkManager.DEFAULT_REPLICATION_SOURCE_RATIO);
assertEquals(expected, sinkManager.getNumSinks());
} }
} }