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