1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.replication;
20
21 import static org.mockito.Mockito.doNothing;
22 import static org.mockito.Mockito.mock;
23 import static org.mockito.Mockito.spy;
24 import static org.mockito.Mockito.verify;
25 import static org.mockito.Mockito.when;
26
27 import java.io.IOException;
28 import java.util.ArrayList;
29 import java.util.HashMap;
30 import java.util.List;
31 import java.util.Map;
32 import java.util.UUID;
33 import java.util.concurrent.atomic.AtomicBoolean;
34 import java.util.concurrent.atomic.AtomicInteger;
35 import java.util.concurrent.atomic.AtomicReference;
36
37 import org.apache.commons.logging.Log;
38 import org.apache.commons.logging.LogFactory;
39 import org.apache.hadoop.hbase.Cell;
40 import org.apache.hadoop.hbase.testclassification.MediumTests;
41 import org.apache.hadoop.hbase.HRegionInfo;
42 import org.apache.hadoop.hbase.Waiter;
43 import org.apache.hadoop.hbase.client.Connection;
44 import org.apache.hadoop.hbase.client.ConnectionFactory;
45 import org.apache.hadoop.hbase.client.Put;
46 import org.apache.hadoop.hbase.client.Table;
47 import org.apache.hadoop.hbase.wal.WAL.Entry;
48 import org.apache.hadoop.hbase.regionserver.HRegion;
49 import org.apache.hadoop.hbase.replication.regionserver.HBaseInterClusterReplicationEndpoint;
50 import org.apache.hadoop.hbase.replication.regionserver.MetricsReplicationGlobalSourceSource;
51 import org.apache.hadoop.hbase.replication.regionserver.MetricsReplicationSourceImpl;
52 import org.apache.hadoop.hbase.replication.regionserver.MetricsReplicationSourceSource;
53 import org.apache.hadoop.hbase.replication.regionserver.MetricsReplicationSourceSourceImpl;
54 import org.apache.hadoop.hbase.replication.regionserver.MetricsSource;
55 import org.apache.hadoop.hbase.util.Bytes;
56 import org.apache.hadoop.hbase.util.JVMClusterUtil.RegionServerThread;
57 import org.apache.hadoop.hbase.util.Threads;
58 import org.apache.hadoop.hbase.zookeeper.ZKConfig;
59 import org.apache.hadoop.metrics2.lib.DynamicMetricsRegistry;
60 import org.junit.AfterClass;
61 import org.junit.Assert;
62 import org.junit.Before;
63 import org.junit.BeforeClass;
64 import org.junit.Test;
65 import org.junit.experimental.categories.Category;
66
67
68
69
70 @Category(MediumTests.class)
71 public class TestReplicationEndpoint extends TestReplicationBase {
72 private static final Log LOG = LogFactory.getLog(TestReplicationEndpoint.class);
73
74 static int numRegionServers;
75
76 @BeforeClass
77 public static void setUpBeforeClass() throws Exception {
78 TestReplicationBase.setUpBeforeClass();
79 admin.removePeer("2");
80 numRegionServers = utility1.getHBaseCluster().getRegionServerThreads().size();
81 }
82
83 @AfterClass
84 public static void tearDownAfterClass() throws Exception {
85 TestReplicationBase.tearDownAfterClass();
86
87 Assert.assertTrue(ReplicationEndpointForTest.stoppedCount.get() > 0);
88 }
89
90 @Before
91 public void setup() throws Exception {
92 ReplicationEndpointForTest.contructedCount.set(0);
93 ReplicationEndpointForTest.startedCount.set(0);
94 ReplicationEndpointForTest.replicateCount.set(0);
95 ReplicationEndpointReturningFalse.replicated.set(false);
96 ReplicationEndpointForTest.lastEntries = null;
97 final List<RegionServerThread> rsThreads =
98 utility1.getMiniHBaseCluster().getRegionServerThreads();
99 for (RegionServerThread rs : rsThreads) {
100 utility1.getHBaseAdmin().rollWALWriter(rs.getRegionServer().getServerName());
101 }
102
103 utility1.waitFor(3000, new Waiter.ExplainingPredicate<Exception>() {
104 @Override
105 public boolean evaluate() throws Exception {
106 for (RegionServerThread rs : rsThreads) {
107 if (!rs.getRegionServer().walRollRequestFinished()) {
108 return false;
109 }
110 }
111 return true;
112 }
113
114 @Override
115 public String explainFailure() throws Exception {
116 List<String> logRollInProgressRsList = new ArrayList<String>();
117 for (RegionServerThread rs : rsThreads) {
118 if (!rs.getRegionServer().walRollRequestFinished()) {
119 logRollInProgressRsList.add(rs.getRegionServer().toString());
120 }
121 }
122 return "Still waiting for log roll on regionservers: " + logRollInProgressRsList;
123 }
124 });
125 }
126
127 @Test (timeout=120000)
128 public void testCustomReplicationEndpoint() throws Exception {
129
130 admin.addPeer("testCustomReplicationEndpoint",
131 new ReplicationPeerConfig().setClusterKey(ZKConfig.getZooKeeperClusterKey(conf1))
132 .setReplicationEndpointImpl(ReplicationEndpointForTest.class.getName()), null);
133
134
135 Waiter.waitFor(conf1, 60000, new Waiter.Predicate<Exception>() {
136 @Override
137 public boolean evaluate() throws Exception {
138 return ReplicationEndpointForTest.contructedCount.get() >= numRegionServers;
139 }
140 });
141
142 Waiter.waitFor(conf1, 60000, new Waiter.Predicate<Exception>() {
143 @Override
144 public boolean evaluate() throws Exception {
145 return ReplicationEndpointForTest.startedCount.get() >= numRegionServers;
146 }
147 });
148
149 Assert.assertEquals(0, ReplicationEndpointForTest.replicateCount.get());
150
151
152 doPut(Bytes.toBytes("row42"));
153
154 Waiter.waitFor(conf1, 60000, new Waiter.Predicate<Exception>() {
155 @Override
156 public boolean evaluate() throws Exception {
157 return ReplicationEndpointForTest.replicateCount.get() >= 1;
158 }
159 });
160
161 doAssert(Bytes.toBytes("row42"));
162
163 admin.removePeer("testCustomReplicationEndpoint");
164 }
165
166 @Test (timeout=120000)
167 public void testReplicationEndpointReturnsFalseOnReplicate() throws Exception {
168 Assert.assertEquals(0, ReplicationEndpointForTest.replicateCount.get());
169 Assert.assertTrue(!ReplicationEndpointReturningFalse.replicated.get());
170 int peerCount = admin.getPeersCount();
171 final String id = "testReplicationEndpointReturnsFalseOnReplicate";
172 admin.addPeer(id,
173 new ReplicationPeerConfig().setClusterKey(ZKConfig.getZooKeeperClusterKey(conf1))
174 .setReplicationEndpointImpl(ReplicationEndpointReturningFalse.class.getName()), null);
175
176
177
178 if (admin.getPeersCount() <= peerCount) {
179 LOG.info("Waiting on peercount to go up from " + peerCount);
180 Threads.sleep(100);
181 }
182
183 doPut(row);
184
185 Waiter.waitFor(conf1, 60000, new Waiter.Predicate<Exception>() {
186 @Override
187 public boolean evaluate() throws Exception {
188
189
190 int count = ReplicationEndpointForTest.replicateCount.get();
191 LOG.info("count=" + count);
192 return ReplicationEndpointReturningFalse.replicated.get();
193 }
194 });
195 if (ReplicationEndpointReturningFalse.ex.get() != null) {
196 throw ReplicationEndpointReturningFalse.ex.get();
197 }
198
199 admin.removePeer("testReplicationEndpointReturnsFalseOnReplicate");
200 }
201
202 @Test (timeout=120000)
203 public void testInterClusterReplication() throws Exception {
204 final String id = "testInterClusterReplication";
205
206 List<HRegion> regions = utility1.getHBaseCluster().getRegions(tableName);
207 int totEdits = 0;
208
209
210
211 for(HRegion region: regions) {
212 HRegionInfo hri = region.getRegionInfo();
213 byte[] row = hri.getStartKey();
214 for (int i = 0; i < 100; i++) {
215 if (row.length > 0) {
216 Put put = new Put(row);
217 put.addColumn(famName, row, row);
218 region.put(put);
219 totEdits++;
220 }
221 }
222 }
223
224 admin.addPeer(id,
225 new ReplicationPeerConfig().setClusterKey(ZKConfig.getZooKeeperClusterKey(conf2))
226 .setReplicationEndpointImpl(InterClusterReplicationEndpointForTest.class.getName()),
227 null);
228
229 final int numEdits = totEdits;
230 Waiter.waitFor(conf1, 30000, new Waiter.ExplainingPredicate<Exception>() {
231 @Override
232 public boolean evaluate() throws Exception {
233 return InterClusterReplicationEndpointForTest.replicateCount.get() == numEdits;
234 }
235 @Override
236 public String explainFailure() throws Exception {
237 String failure = "Failed to replicate all edits, expected = " + numEdits
238 + " replicated = " + InterClusterReplicationEndpointForTest.replicateCount.get();
239 return failure;
240 }
241 });
242
243 admin.removePeer("testInterClusterReplication");
244 utility1.deleteTableData(tableName);
245 }
246
247 @Test (timeout=120000)
248 public void testWALEntryFilterFromReplicationEndpoint() throws Exception {
249 admin.addPeer("testWALEntryFilterFromReplicationEndpoint",
250 new ReplicationPeerConfig().setClusterKey(ZKConfig.getZooKeeperClusterKey(conf1))
251 .setReplicationEndpointImpl(ReplicationEndpointWithWALEntryFilter.class.getName()), null);
252
253 try (Connection connection = ConnectionFactory.createConnection(conf1)) {
254 doPut(connection, Bytes.toBytes("row1"));
255 doPut(connection, row);
256 doPut(connection, Bytes.toBytes("row2"));
257 }
258
259 Waiter.waitFor(conf1, 60000, new Waiter.Predicate<Exception>() {
260 @Override
261 public boolean evaluate() throws Exception {
262 return ReplicationEndpointForTest.replicateCount.get() >= 1;
263 }
264 });
265
266 Assert.assertNull(ReplicationEndpointWithWALEntryFilter.ex.get());
267 admin.removePeer("testWALEntryFilterFromReplicationEndpoint");
268 }
269
270 @Test
271 public void testMetricsSourceBaseSourcePassthrough(){
272
273
274
275
276
277
278
279
280 String id = "id";
281 DynamicMetricsRegistry mockRegistry = new DynamicMetricsRegistry(id);
282 MetricsReplicationSourceImpl singleRms = mock(MetricsReplicationSourceImpl.class);
283 when(singleRms.getMetricsRegistry()).thenReturn(mockRegistry);
284 MetricsReplicationSourceImpl globalRms = mock(MetricsReplicationSourceImpl.class);
285 when(globalRms.getMetricsRegistry()).thenReturn(mockRegistry);
286
287 MetricsReplicationSourceSource singleSourceSource = new MetricsReplicationSourceSourceImpl(singleRms, id);
288 MetricsReplicationSourceSource globalSourceSource = new MetricsReplicationGlobalSourceSource(globalRms);
289 MetricsReplicationSourceSource spyglobalSourceSource = spy(globalSourceSource);
290 doNothing().when(spyglobalSourceSource).incrFailedRecoveryQueue();
291 Map<String, MetricsReplicationSourceSource> singleSourceSourceByTable = new HashMap<>();
292 MetricsSource source = new MetricsSource(id, singleSourceSource, spyglobalSourceSource,
293 singleSourceSourceByTable);
294
295
296
297
298 boolean containsRandomNewTable = source.getSingleSourceSourceByTable()
299 .containsKey("RandomNewTable");
300 Assert.assertEquals(false, containsRandomNewTable);
301 source.setAgeOfLastShippedOpByTable(123L, "RandomNewTable");
302 containsRandomNewTable = source.getSingleSourceSourceByTable()
303 .containsKey("RandomNewTable");
304 Assert.assertEquals(true, containsRandomNewTable);
305 MetricsReplicationSourceSource msr = source.getSingleSourceSourceByTable()
306 .get("RandomNewTable");
307
308
309 Assert.assertTrue(msr.getLastShippedAge() > 0);
310
311 source.incrFailedRecoveryQueue();
312 verify(spyglobalSourceSource).incrFailedRecoveryQueue();
313 }
314
315 private void doPut(byte[] row) throws IOException {
316 try (Connection connection = ConnectionFactory.createConnection(conf1)) {
317 doPut(connection, row);
318 }
319 }
320
321 private void doPut(final Connection connection, final byte [] row) throws IOException {
322 try (Table t = connection.getTable(tableName)) {
323 Put put = new Put(row);
324 put.add(famName, row, row);
325 t.put(put);
326 }
327 }
328
329 private static void doAssert(byte[] row) throws Exception {
330 if (ReplicationEndpointForTest.lastEntries == null) {
331 return;
332 }
333 Assert.assertEquals(1, ReplicationEndpointForTest.lastEntries.size());
334 List<Cell> cells = ReplicationEndpointForTest.lastEntries.get(0).getEdit().getCells();
335 Assert.assertEquals(1, cells.size());
336 Assert.assertTrue(Bytes.equals(cells.get(0).getRowArray(), cells.get(0).getRowOffset(),
337 cells.get(0).getRowLength(), row, 0, row.length));
338 }
339
340 public static class ReplicationEndpointForTest extends BaseReplicationEndpoint {
341 static UUID uuid = UUID.randomUUID();
342 static AtomicInteger contructedCount = new AtomicInteger();
343 static AtomicInteger startedCount = new AtomicInteger();
344 static AtomicInteger stoppedCount = new AtomicInteger();
345 static AtomicInteger replicateCount = new AtomicInteger();
346 static volatile List<Entry> lastEntries = null;
347
348 public ReplicationEndpointForTest() {
349 contructedCount.incrementAndGet();
350 }
351
352 @Override
353 public UUID getPeerUUID() {
354 return uuid;
355 }
356
357 @Override
358 public boolean replicate(ReplicateContext replicateContext) {
359 replicateCount.incrementAndGet();
360 lastEntries = replicateContext.entries;
361 return true;
362 }
363
364 @Override
365 protected void doStart() {
366 startedCount.incrementAndGet();
367 notifyStarted();
368 }
369
370 @Override
371 protected void doStop() {
372 stoppedCount.incrementAndGet();
373 notifyStopped();
374 }
375 }
376
377 public static class InterClusterReplicationEndpointForTest
378 extends HBaseInterClusterReplicationEndpoint {
379
380 static AtomicInteger replicateCount = new AtomicInteger();
381 static boolean failedOnce;
382
383 @Override
384 public boolean replicate(ReplicateContext replicateContext) {
385 boolean success = super.replicate(replicateContext);
386 if (success) {
387 replicateCount.addAndGet(replicateContext.entries.size());
388 }
389 return success;
390 }
391
392 @Override
393 protected Replicator createReplicator(List<Entry> entries, int ordinal) {
394
395 if (failedOnce) {
396 return new DummyReplicator(entries, ordinal);
397 } else {
398 failedOnce = true;
399 return new FailingDummyReplicator(entries, ordinal);
400 }
401 }
402
403 protected class DummyReplicator extends Replicator {
404
405 private int ordinal;
406
407 public DummyReplicator(List<Entry> entries, int ordinal) {
408 super(entries, ordinal);
409 this.ordinal = ordinal;
410 }
411
412 @Override
413 public Integer call() throws IOException {
414 return ordinal;
415 }
416 }
417
418 protected class FailingDummyReplicator extends DummyReplicator {
419
420 public FailingDummyReplicator(List<Entry> entries, int ordinal) {
421 super(entries, ordinal);
422 }
423
424 @Override
425 public Integer call() throws IOException {
426 throw new IOException("Sample Exception: Failed to replicate.");
427 }
428 }
429 }
430
431 public static class ReplicationEndpointReturningFalse extends ReplicationEndpointForTest {
432 static int COUNT = 10;
433 static AtomicReference<Exception> ex = new AtomicReference<Exception>(null);
434 static AtomicBoolean replicated = new AtomicBoolean(false);
435 @Override
436 public boolean replicate(ReplicateContext replicateContext) {
437 try {
438
439 doAssert(row);
440 } catch (Exception e) {
441 ex.set(e);
442 }
443
444 super.replicate(replicateContext);
445 LOG.info("Replicated " + row + ", count=" + replicateCount.get());
446
447 replicated.set(replicateCount.get() > COUNT);
448 return replicated.get();
449 }
450 }
451
452
453 public static class ReplicationEndpointWithWALEntryFilter extends ReplicationEndpointForTest {
454 static AtomicReference<Exception> ex = new AtomicReference<Exception>(null);
455
456 @Override
457 public boolean replicate(ReplicateContext replicateContext) {
458 try {
459 super.replicate(replicateContext);
460 doAssert(row);
461 } catch (Exception e) {
462 ex.set(e);
463 }
464 return true;
465 }
466
467 @Override
468 public WALEntryFilter getWALEntryfilter() {
469 return new ChainWALEntryFilter(super.getWALEntryfilter(), new WALEntryFilter() {
470 @Override
471 public Entry filter(Entry entry) {
472 ArrayList<Cell> cells = entry.getEdit().getCells();
473 int size = cells.size();
474 for (int i = size-1; i >= 0; i--) {
475 Cell cell = cells.get(i);
476 if (!Bytes.equals(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength(),
477 row, 0, row.length)) {
478 cells.remove(i);
479 }
480 }
481 return entry;
482 }
483 });
484 }
485 }
486 }