1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.hbase.replication.regionserver;
19
20 import static org.junit.Assert.assertEquals;
21
22 import java.io.IOException;
23 import java.util.List;
24 import java.util.concurrent.atomic.AtomicBoolean;
25 import java.util.concurrent.atomic.AtomicInteger;
26
27 import org.apache.commons.logging.Log;
28 import org.apache.commons.logging.LogFactory;
29 import org.apache.hadoop.fs.Path;
30 import org.apache.hadoop.hbase.HBaseTestingUtility;
31 import org.apache.hadoop.hbase.TableName;
32 import org.apache.hadoop.hbase.Waiter;
33 import org.apache.hadoop.hbase.client.HBaseAdmin;
34 import org.apache.hadoop.hbase.client.Put;
35 import org.apache.hadoop.hbase.ipc.RpcServer;
36 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.AdminService.BlockingInterface;
37 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.*;
38 import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
39 import org.apache.hadoop.hbase.replication.TestReplicationBase;
40 import org.apache.hadoop.hbase.replication.regionserver.HBaseInterClusterReplicationEndpoint;
41 import org.apache.hadoop.hbase.testclassification.MediumTests;
42 import org.apache.hadoop.hbase.wal.WAL.Entry;
43
44 import org.junit.AfterClass;
45 import org.junit.BeforeClass;
46 import org.junit.Test;
47 import org.junit.experimental.categories.Category;
48
49 import com.google.protobuf.RpcController;
50 import com.google.protobuf.ServiceException;
51
52 @Category(MediumTests.class)
53 public class TestReplicator extends TestReplicationBase {
54
55 static final Log LOG = LogFactory.getLog(TestReplicator.class);
56 static final int NUM_ROWS = 10;
57
58 @BeforeClass
59 public static void setUpBeforeClass() throws Exception {
60
61 conf1.setInt(RpcServer.MAX_REQUEST_SIZE, 1024 * 10);
62 TestReplicationBase.setUpBeforeClass();
63 admin.removePeer("2");
64 }
65
66 @Test
67 public void testReplicatorBatching() throws Exception {
68
69 truncateTable(utility1, tableName);
70 truncateTable(utility2, tableName);
71
72
73 admin.addPeer("testReplicatorBatching",
74 new ReplicationPeerConfig().setClusterKey(utility2.getClusterKey())
75 .setReplicationEndpointImpl(ReplicationEndpointForTest.class.getName()), null);
76
77 ReplicationEndpointForTest.setBatchCount(0);
78 ReplicationEndpointForTest.setEntriesCount(0);
79 try {
80 ReplicationEndpointForTest.pause();
81 try {
82
83
84 final byte[] valueBytes = new byte[8 *1024];
85 for (int i = 0; i < NUM_ROWS; i++) {
86 htable1.put(new Put(("row"+Integer.toString(i)).getBytes())
87 .addColumn(famName, null, valueBytes)
88 );
89 }
90 } finally {
91 ReplicationEndpointForTest.resume();
92 }
93
94
95
96 Waiter.waitFor(conf1, 60000, new Waiter.ExplainingPredicate<Exception>() {
97 @Override
98 public boolean evaluate() throws Exception {
99 LOG.info("Count=" + ReplicationEndpointForTest.getBatchCount());
100 return ReplicationEndpointForTest.getBatchCount() >= NUM_ROWS;
101 }
102
103 @Override
104 public String explainFailure() throws Exception {
105 return "We waited too long for expected replication of " + NUM_ROWS + " entries";
106 }
107 });
108
109 assertEquals("We sent an incorrect number of batches", NUM_ROWS,
110 ReplicationEndpointForTest.getBatchCount());
111 assertEquals("We did not replicate enough rows", NUM_ROWS,
112 utility2.countRows(htable2));
113 } finally {
114 admin.removePeer("testReplicatorBatching");
115 }
116 }
117
118 @Test
119 public void testReplicatorWithErrors() throws Exception {
120
121 truncateTable(utility1, tableName);
122 truncateTable(utility2, tableName);
123
124
125 admin.addPeer("testReplicatorWithErrors",
126 new ReplicationPeerConfig().setClusterKey(utility2.getClusterKey())
127 .setReplicationEndpointImpl(FailureInjectingReplicationEndpointForTest.class.getName()),
128 null);
129
130 FailureInjectingReplicationEndpointForTest.setBatchCount(0);
131 FailureInjectingReplicationEndpointForTest.setEntriesCount(0);
132 try {
133 FailureInjectingReplicationEndpointForTest.pause();
134 try {
135
136
137 final byte[] valueBytes = new byte[8 *1024];
138 for (int i = 0; i < NUM_ROWS; i++) {
139 htable1.put(new Put(("row"+Integer.toString(i)).getBytes())
140 .addColumn(famName, null, valueBytes)
141 );
142 }
143 } finally {
144 FailureInjectingReplicationEndpointForTest.resume();
145 }
146
147
148
149 Waiter.waitFor(conf1, 60000, new Waiter.ExplainingPredicate<Exception>() {
150 @Override
151 public boolean evaluate() throws Exception {
152 return FailureInjectingReplicationEndpointForTest.getEntriesCount() >= NUM_ROWS;
153 }
154
155 @Override
156 public String explainFailure() throws Exception {
157 return "We waited too long for expected replication of " + NUM_ROWS + " entries";
158 }
159 });
160
161 assertEquals("We did not replicate enough rows", NUM_ROWS,
162 utility2.countRows(htable2));
163 } finally {
164 admin.removePeer("testReplicatorWithErrors");
165 }
166 }
167
168 @AfterClass
169 public static void tearDownAfterClass() throws Exception {
170 TestReplicationBase.tearDownAfterClass();
171 }
172
173 private void truncateTable(HBaseTestingUtility util, TableName tablename) throws IOException {
174 HBaseAdmin admin = util.getHBaseAdmin();
175 admin.disableTable(tableName);
176 admin.truncateTable(tablename, false);
177 }
178
179 public static class ReplicationEndpointForTest extends HBaseInterClusterReplicationEndpoint {
180
181 private static AtomicInteger batchCount = new AtomicInteger(0);
182 private static int entriesCount;
183 private static final Object latch = new Object();
184 private static AtomicBoolean useLatch = new AtomicBoolean(false);
185
186 public static void resume() {
187 useLatch.set(false);
188 synchronized (latch) {
189 latch.notifyAll();
190 }
191 }
192
193 public static void pause() {
194 useLatch.set(true);
195 }
196
197 public static void await() throws InterruptedException {
198 if (useLatch.get()) {
199 LOG.info("Waiting on latch");
200 synchronized(latch) {
201 latch.wait();
202 }
203 LOG.info("Waited on latch, now proceeding");
204 }
205 }
206
207 public static int getBatchCount() {
208 return batchCount.get();
209 }
210
211 public static void setBatchCount(int i) {
212 LOG.info("SetBatchCount=" + i + ", old=" + getBatchCount());
213 batchCount.set(i);
214 }
215
216 public static int getEntriesCount() {
217 return entriesCount;
218 }
219
220 public static void setEntriesCount(int i) {
221 LOG.info("SetEntriesCount=" + i);
222 entriesCount = i;
223 }
224
225 public class ReplicatorForTest extends Replicator {
226
227 public ReplicatorForTest(List<Entry> entries, int ordinal) {
228 super(entries, ordinal);
229 }
230
231 @Override
232 protected void replicateEntries(BlockingInterface rrs, final List<Entry> entries,
233 String replicationClusterId, Path baseNamespaceDir, Path hfileArchiveDir)
234 throws IOException {
235 try {
236 long size = 0;
237 for (Entry e: entries) {
238 size += e.getKey().estimatedSerializedSizeOf();
239 size += e.getEdit().estimatedSerializedSizeOf();
240 }
241 LOG.info("Replicating batch " + System.identityHashCode(entries) + " of " +
242 entries.size() + " entries with total size " + size + " bytes to " +
243 replicationClusterId);
244 super.replicateEntries(rrs, entries, replicationClusterId, baseNamespaceDir,
245 hfileArchiveDir);
246 entriesCount += entries.size();
247 int count = batchCount.incrementAndGet();
248 LOG.info("Completed replicating batch " + System.identityHashCode(entries) +
249 " count=" + count);
250 } catch (IOException e) {
251 LOG.info("Failed to replicate batch " + System.identityHashCode(entries), e);
252 throw e;
253 }
254 }
255 }
256
257 @Override
258 public boolean replicate(ReplicateContext replicateContext) {
259 try {
260 await();
261 } catch (InterruptedException e) {
262 LOG.warn("Interrupted waiting for latch", e);
263 }
264 return super.replicate(replicateContext);
265 }
266
267 @Override
268 protected Replicator createReplicator(List<Entry> entries, int ordinal) {
269 return new ReplicatorForTest(entries, ordinal);
270 }
271 }
272
273 public static class FailureInjectingReplicationEndpointForTest
274 extends ReplicationEndpointForTest {
275
276 static class FailureInjectingBlockingInterface implements BlockingInterface {
277
278 private final BlockingInterface delegate;
279 private volatile boolean failNext;
280
281 public FailureInjectingBlockingInterface(BlockingInterface delegate) {
282 this.delegate = delegate;
283 }
284
285 @Override
286 public ReplicateWALEntryResponse replicateWALEntry(RpcController controller,
287 ReplicateWALEntryRequest request) throws ServiceException {
288 if (!failNext) {
289 failNext = true;
290 return delegate.replicateWALEntry(controller, request);
291 } else {
292 failNext = false;
293 throw new ServiceException("Injected failure");
294 }
295 }
296
297 @Override
298 public GetRegionInfoResponse getRegionInfo(RpcController controller,
299 GetRegionInfoRequest request) throws ServiceException {
300 return delegate.getRegionInfo(controller, request);
301 }
302
303 @Override
304 public GetStoreFileResponse getStoreFile(RpcController controller,
305 GetStoreFileRequest request) throws ServiceException {
306 return delegate.getStoreFile(controller, request);
307 }
308
309 @Override
310 public GetOnlineRegionResponse getOnlineRegion(RpcController controller,
311 GetOnlineRegionRequest request) throws ServiceException {
312 return delegate.getOnlineRegion(controller, request);
313 }
314
315 @Override
316 public OpenRegionResponse openRegion(RpcController controller, OpenRegionRequest request)
317 throws ServiceException {
318 return delegate.openRegion(controller, request);
319 }
320
321 @Override
322 public WarmupRegionResponse warmupRegion(RpcController controller,
323 WarmupRegionRequest request) throws ServiceException {
324 return delegate.warmupRegion(controller, request);
325 }
326
327 @Override
328 public CloseRegionResponse closeRegion(RpcController controller, CloseRegionRequest request)
329 throws ServiceException {
330 return delegate.closeRegion(controller, request);
331 }
332
333 @Override
334 public FlushRegionResponse flushRegion(RpcController controller, FlushRegionRequest request)
335 throws ServiceException {
336 return delegate.flushRegion(controller, request);
337 }
338
339 @Override
340 public SplitRegionResponse splitRegion(RpcController controller, SplitRegionRequest request)
341 throws ServiceException {
342 return delegate.splitRegion(controller, request);
343 }
344
345 @Override
346 public CompactRegionResponse compactRegion(RpcController controller,
347 CompactRegionRequest request) throws ServiceException {
348 return delegate.compactRegion(controller, request);
349 }
350
351 @Override
352 public MergeRegionsResponse mergeRegions(RpcController controller,
353 MergeRegionsRequest request) throws ServiceException {
354 return delegate.mergeRegions(controller, request);
355 }
356
357 @Override
358 public ReplicateWALEntryResponse replay(RpcController controller,
359 ReplicateWALEntryRequest request) throws ServiceException {
360 return delegate.replay(controller, request);
361 }
362
363 @Override
364 public RollWALWriterResponse rollWALWriter(RpcController controller,
365 RollWALWriterRequest request) throws ServiceException {
366 return delegate.rollWALWriter(controller, request);
367 }
368
369 @Override
370 public GetServerInfoResponse getServerInfo(RpcController controller,
371 GetServerInfoRequest request) throws ServiceException {
372 return delegate.getServerInfo(controller, request);
373 }
374
375 @Override
376 public StopServerResponse stopServer(RpcController controller, StopServerRequest request)
377 throws ServiceException {
378 return delegate.stopServer(controller, request);
379 }
380
381 @Override
382 public UpdateFavoredNodesResponse updateFavoredNodes(RpcController controller,
383 UpdateFavoredNodesRequest request) throws ServiceException {
384 return delegate.updateFavoredNodes(controller, request);
385 }
386
387 @Override
388 public UpdateConfigurationResponse updateConfiguration(RpcController controller,
389 UpdateConfigurationRequest request) throws ServiceException {
390 return delegate.updateConfiguration(controller, request);
391 }
392 }
393
394 public class FailureInjectingReplicatorForTest extends ReplicatorForTest {
395
396 public FailureInjectingReplicatorForTest(List<Entry> entries, int ordinal) {
397 super(entries, ordinal);
398 }
399
400 @Override
401 protected void replicateEntries(BlockingInterface rrs, List<Entry> entries,
402 String replicationClusterId, Path baseNamespaceDir, Path hfileArchiveDir)
403 throws IOException {
404 super.replicateEntries(new FailureInjectingBlockingInterface(rrs), entries,
405 replicationClusterId, baseNamespaceDir, hfileArchiveDir);
406 }
407 }
408
409 @Override
410 protected Replicator createReplicator(List<Entry> entries, int ordinal) {
411 return new FailureInjectingReplicatorForTest(entries, ordinal);
412 }
413 }
414
415 }