1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20 package org.apache.hadoop.hbase.client;
21
22 import java.io.IOException;
23 import static org.junit.Assert.assertEquals;
24 import static org.junit.Assert.assertNull;
25 import static org.junit.Assert.assertTrue;
26 import static org.junit.Assert.fail;
27
28 import java.util.ArrayList;
29 import java.util.Arrays;
30 import java.util.List;
31 import java.util.Random;
32 import java.util.concurrent.CountDownLatch;
33 import java.util.concurrent.ExecutorService;
34 import java.util.concurrent.Executors;
35 import java.util.concurrent.TimeUnit;
36 import static junit.framework.Assert.assertFalse;
37 import org.apache.commons.logging.Log;
38 import org.apache.commons.logging.LogFactory;
39 import org.apache.hadoop.conf.Configuration;
40 import org.apache.hadoop.hbase.Cell;
41 import org.apache.hadoop.hbase.CellUtil;
42 import org.apache.hadoop.hbase.HBaseTestingUtility;
43 import org.apache.hadoop.hbase.HColumnDescriptor;
44 import org.apache.hadoop.hbase.HConstants;
45 import org.apache.hadoop.hbase.HRegionLocation;
46 import org.apache.hadoop.hbase.HTableDescriptor;
47 import org.apache.hadoop.hbase.TableName;
48 import org.apache.hadoop.hbase.client.coprocessor.Batch;
49 import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
50 import org.apache.hadoop.hbase.coprocessor.MultiRowMutationEndpoint;
51 import org.apache.hadoop.hbase.coprocessor.ObserverContext;
52 import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
53 import org.apache.hadoop.hbase.coprocessor.RegionObserver;
54 import org.apache.hadoop.hbase.Coprocessor;
55 import org.apache.hadoop.hbase.ipc.BlockingRpcCallback;
56 import org.apache.hadoop.hbase.ipc.ServerRpcController;
57 import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
58 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos;
59 import org.apache.hadoop.hbase.protobuf.generated.MultiRowMutationProtos;
60 import org.apache.hadoop.hbase.regionserver.HRegion;
61 import org.apache.hadoop.hbase.regionserver.HRegionServer;
62 import org.apache.hadoop.hbase.regionserver.MiniBatchOperationInProgress;
63 import org.apache.hadoop.hbase.regionserver.Region;
64 import org.apache.hadoop.hbase.testclassification.LargeTests;
65 import org.apache.hadoop.hbase.util.Bytes;
66 import org.apache.hadoop.hbase.util.Pair;
67 import org.junit.After;
68 import org.junit.AfterClass;
69 import org.junit.Before;
70 import org.junit.BeforeClass;
71 import org.junit.Test;
72 import org.junit.experimental.categories.Category;
73
74 @Category(LargeTests.class)
75 public class TestFromClientSide3 {
76 private static final Log LOG = LogFactory.getLog(TestFromClientSide3.class);
77 private final static HBaseTestingUtility TEST_UTIL
78 = new HBaseTestingUtility();
79 private static byte[] FAMILY = Bytes.toBytes("testFamily");
80 private static Random random = new Random();
81 private static int SLAVES = 3;
82 private static byte [] ROW = Bytes.toBytes("testRow");
83 private static final byte[] ANOTHERROW = Bytes.toBytes("anotherrow");
84 private static byte [] QUALIFIER = Bytes.toBytes("testQualifier");
85 private static byte [] VALUE = Bytes.toBytes("testValue");
86 private final static byte[] COL_QUAL = Bytes.toBytes("f1");
87 private final static byte[] VAL_BYTES = Bytes.toBytes("v1");
88 private final static byte[] ROW_BYTES = Bytes.toBytes("r1");
89
90
91
92
93 @BeforeClass
94 public static void setUpBeforeClass() throws Exception {
95 TEST_UTIL.getConfiguration().setBoolean(
96 "hbase.online.schema.update.enable", true);
97 TEST_UTIL.startMiniCluster(SLAVES);
98 }
99
100
101
102
103 @AfterClass
104 public static void tearDownAfterClass() throws Exception {
105 TEST_UTIL.shutdownMiniCluster();
106 }
107
108
109
110
111 @Before
112 public void setUp() throws Exception {
113
114 }
115
116
117
118
119 @After
120 public void tearDown() throws Exception {
121 for (HTableDescriptor htd: TEST_UTIL.getHBaseAdmin().listTables()) {
122 LOG.info("Tear down, remove table=" + htd.getTableName());
123 TEST_UTIL.deleteTable(htd.getTableName());
124 }
125 }
126
127 private void randomCFPuts(Table table, byte[] row, byte[] family, int nPuts)
128 throws Exception {
129 Put put = new Put(row);
130 for (int i = 0; i < nPuts; i++) {
131 byte[] qualifier = Bytes.toBytes(random.nextInt());
132 byte[] value = Bytes.toBytes(random.nextInt());
133 put.add(family, qualifier, value);
134 }
135 table.put(put);
136 }
137
138 private void performMultiplePutAndFlush(HBaseAdmin admin, HTable table,
139 byte[] row, byte[] family, int nFlushes, int nPuts)
140 throws Exception {
141
142
143 HRegionLocation loc = table.getRegionLocation(row, true);
144 AdminProtos.AdminService.BlockingInterface server =
145 admin.getConnection().getAdmin(loc.getServerName());
146 byte[] regName = loc.getRegionInfo().getRegionName();
147
148 for (int i = 0; i < nFlushes; i++) {
149 randomCFPuts(table, row, family, nPuts);
150 List<String> sf = ProtobufUtil.getStoreFiles(server, regName, FAMILY);
151 int sfCount = sf.size();
152
153
154 admin.flush(table.getTableName());
155
156
157 while (ProtobufUtil.getStoreFiles(
158 server, regName, FAMILY).size() == sfCount) {
159 Thread.sleep(40);
160 }
161 }
162 }
163
164 private static List<Cell> toList(ResultScanner scanner) {
165 try {
166 List<Cell> cells = new ArrayList<>();
167 for (Result r : scanner) {
168 cells.addAll(r.listCells());
169 }
170 return cells;
171 } finally {
172 scanner.close();
173 }
174 }
175
176 @Test
177 public void testScanAfterDeletingSpecifiedRow() throws IOException {
178 TableName tableName = TableName.valueOf("testScanAfterDeletingSpecifiedRow");
179 HTableDescriptor desc = new HTableDescriptor(tableName);
180 desc.addFamily(new HColumnDescriptor(FAMILY));
181 TEST_UTIL.getHBaseAdmin().createTable(desc);
182 byte[] row = Bytes.toBytes("SpecifiedRow");
183 byte[] value0 = Bytes.toBytes("value_0");
184 byte[] value1 = Bytes.toBytes("value_1");
185 try (Table t = TEST_UTIL.getConnection().getTable(tableName)) {
186 Put put = new Put(row);
187 put.addColumn(FAMILY, QUALIFIER, VALUE);
188 t.put(put);
189 Delete d = new Delete(row);
190 t.delete(d);
191 put = new Put(row);
192 put.addColumn(FAMILY, null, value0);
193 t.put(put);
194 put = new Put(row);
195 put.addColumn(FAMILY, null, value1);
196 t.put(put);
197 List<Cell> cells = toList(t.getScanner(new Scan()));
198 assertEquals(1, cells.size());
199 assertEquals("value_1", Bytes.toString(CellUtil.cloneValue(cells.get(0))));
200
201 cells = toList(t.getScanner(new Scan().addFamily(FAMILY)));
202 assertEquals(1, cells.size());
203 assertEquals("value_1", Bytes.toString(CellUtil.cloneValue(cells.get(0))));
204
205 cells = toList(t.getScanner(new Scan().addColumn(FAMILY, QUALIFIER)));
206 assertEquals(0, cells.size());
207
208 TEST_UTIL.getHBaseAdmin().flush(tableName);
209 cells = toList(t.getScanner(new Scan()));
210 assertEquals(1, cells.size());
211 assertEquals("value_1", Bytes.toString(CellUtil.cloneValue(cells.get(0))));
212
213 cells = toList(t.getScanner(new Scan().addFamily(FAMILY)));
214 assertEquals(1, cells.size());
215 assertEquals("value_1", Bytes.toString(CellUtil.cloneValue(cells.get(0))));
216
217 cells = toList(t.getScanner(new Scan().addColumn(FAMILY, QUALIFIER)));
218 assertEquals(0, cells.size());
219 }
220 }
221
222 @Test
223 public void testScanAfterDeletingSpecifiedRowV2() throws IOException {
224 TableName tableName = TableName.valueOf("testScanAfterDeletingSpecifiedRowV2");
225 HTableDescriptor desc = new HTableDescriptor(tableName);
226 desc.addFamily(new HColumnDescriptor(FAMILY));
227 TEST_UTIL.getHBaseAdmin().createTable(desc);
228 byte[] row = Bytes.toBytes("SpecifiedRow");
229 byte[] qual0 = Bytes.toBytes("qual0");
230 byte[] qual1 = Bytes.toBytes("qual1");
231 try (Table t = TEST_UTIL.getConnection().getTable(tableName)) {
232 Delete d = new Delete(row);
233 t.delete(d);
234
235 Put put = new Put(row);
236 put.addColumn(FAMILY, null, VALUE);
237 t.put(put);
238
239 put = new Put(row);
240 put.addColumn(FAMILY, qual1, qual1);
241 t.put(put);
242
243 put = new Put(row);
244 put.addColumn(FAMILY, qual0, qual0);
245 t.put(put);
246
247 Result r = t.get(new Get(row));
248 assertEquals(3, r.size());
249 assertEquals("testValue", Bytes.toString(CellUtil.cloneValue(r.rawCells()[0])));
250 assertEquals("qual0", Bytes.toString(CellUtil.cloneValue(r.rawCells()[1])));
251 assertEquals("qual1", Bytes.toString(CellUtil.cloneValue(r.rawCells()[2])));
252
253 TEST_UTIL.getHBaseAdmin().flush(tableName);
254 r = t.get(new Get(row));
255 assertEquals(3, r.size());
256 assertEquals("testValue", Bytes.toString(CellUtil.cloneValue(r.rawCells()[0])));
257 assertEquals("qual0", Bytes.toString(CellUtil.cloneValue(r.rawCells()[1])));
258 assertEquals("qual1", Bytes.toString(CellUtil.cloneValue(r.rawCells()[2])));
259 }
260 }
261
262
263 @Test(timeout = 60000)
264 public void testAdvancedConfigOverride() throws Exception {
265
266
267
268
269
270
271
272
273
274 TEST_UTIL.getConfiguration().setInt("hbase.hstore.compaction.min", 3);
275
276 String tableName = "testAdvancedConfigOverride";
277 TableName TABLE = TableName.valueOf(tableName);
278 HTable hTable = TEST_UTIL.createTable(TABLE, FAMILY, 10);
279 HBaseAdmin admin = TEST_UTIL.getHBaseAdmin();
280 ClusterConnection connection = (ClusterConnection)TEST_UTIL.getConnection();
281
282
283 byte[] row = Bytes.toBytes(random.nextInt());
284 performMultiplePutAndFlush(admin, hTable, row, FAMILY, 3, 100);
285
286
287 HRegionLocation loc = hTable.getRegionLocation(row, true);
288 byte[] regionName = loc.getRegionInfo().getRegionName();
289 AdminProtos.AdminService.BlockingInterface server =
290 connection.getAdmin(loc.getServerName());
291 assertTrue(ProtobufUtil.getStoreFiles(
292 server, regionName, FAMILY).size() > 1);
293
294
295 admin.compact(TABLE.getName());
296
297
298 for (int i = 0; i < 10 * 1000 / 40; ++i) {
299
300 loc = hTable.getRegionLocation(row, true);
301 if (!loc.getRegionInfo().isOffline()) {
302 regionName = loc.getRegionInfo().getRegionName();
303 server = connection.getAdmin(loc.getServerName());
304 if (ProtobufUtil.getStoreFiles(
305 server, regionName, FAMILY).size() <= 1) {
306 break;
307 }
308 }
309 Thread.sleep(40);
310 }
311
312 assertTrue(ProtobufUtil.getStoreFiles(
313 server, regionName, FAMILY).size() <= 1);
314
315
316 LOG.info("hbase.hstore.compaction.min should now be 5");
317 HTableDescriptor htd = new HTableDescriptor(hTable.getTableDescriptor());
318 htd.setValue("hbase.hstore.compaction.min", String.valueOf(5));
319 admin.modifyTable(TABLE, htd);
320 Pair<Integer, Integer> st;
321 while (null != (st = admin.getAlterStatus(TABLE)) && st.getFirst() > 0) {
322 LOG.debug(st.getFirst() + " regions left to update");
323 Thread.sleep(40);
324 }
325 LOG.info("alter status finished");
326
327
328 performMultiplePutAndFlush(admin, hTable, row, FAMILY, 3, 10);
329
330
331 admin.compact(TABLE.getName());
332
333
334 Thread.sleep(10 * 1000);
335 loc = hTable.getRegionLocation(row, true);
336 regionName = loc.getRegionInfo().getRegionName();
337 server = connection.getAdmin(loc.getServerName());
338 int sfCount = ProtobufUtil.getStoreFiles(
339 server, regionName, FAMILY).size();
340 assertTrue(sfCount > 1);
341
342
343 LOG.info("hbase.hstore.compaction.min should now be 2");
344 HColumnDescriptor hcd = new HColumnDescriptor(htd.getFamily(FAMILY));
345 hcd.setValue("hbase.hstore.compaction.min", String.valueOf(2));
346 htd.modifyFamily(hcd);
347 admin.modifyTable(TABLE, htd);
348 while (null != (st = admin.getAlterStatus(TABLE)) && st.getFirst() > 0) {
349 LOG.debug(st.getFirst() + " regions left to update");
350 Thread.sleep(40);
351 }
352 LOG.info("alter status finished");
353
354
355 admin.compact(TABLE.getName());
356
357
358 for (int i = 0; i < 10 * 1000 / 40; ++i) {
359 loc = hTable.getRegionLocation(row, true);
360 regionName = loc.getRegionInfo().getRegionName();
361 try {
362 server = connection.getAdmin(loc.getServerName());
363 if (ProtobufUtil.getStoreFiles(
364 server, regionName, FAMILY).size() < sfCount) {
365 break;
366 }
367 } catch (Exception e) {
368 LOG.debug("Waiting for region to come online: " + regionName);
369 }
370 Thread.sleep(40);
371 }
372
373 assertTrue(ProtobufUtil.getStoreFiles(
374 server, regionName, FAMILY).size() < sfCount);
375
376
377 LOG.info("Removing CF config value");
378 LOG.info("hbase.hstore.compaction.min should now be 5");
379 hcd = new HColumnDescriptor(htd.getFamily(FAMILY));
380 hcd.setValue("hbase.hstore.compaction.min", null);
381 htd.modifyFamily(hcd);
382 admin.modifyTable(TABLE, htd);
383 while (null != (st = admin.getAlterStatus(TABLE)) && st.getFirst() > 0) {
384 LOG.debug(st.getFirst() + " regions left to update");
385 Thread.sleep(40);
386 }
387 LOG.info("alter status finished");
388 assertNull(hTable.getTableDescriptor().getFamily(FAMILY).getValue(
389 "hbase.hstore.compaction.min"));
390 }
391
392 @Test
393 public void testHTableBatchWithEmptyPut() throws Exception {
394 Table table = TEST_UTIL.createTable(
395 Bytes.toBytes("testHTableBatchWithEmptyPut"), new byte[][] { FAMILY });
396 try {
397 List actions = (List) new ArrayList();
398 Object[] results = new Object[2];
399
400 Put put1 = new Put(ROW);
401 actions.add(put1);
402
403 Put put2 = new Put(ANOTHERROW);
404 put2.add(FAMILY, QUALIFIER, VALUE);
405 actions.add(put2);
406
407 table.batch(actions, results);
408 fail("Empty Put should have failed the batch call");
409 } catch (IllegalArgumentException iae) {
410
411 } finally {
412 table.close();
413 }
414 }
415
416
417
418 @Test
419 public void testHTableWithLargeBatch() throws Exception {
420 Table table = TEST_UTIL.createTable(TableName.valueOf("testHTableWithLargeBatch"),
421 new byte[][] { FAMILY });
422 int sixtyFourK = 64 * 1024;
423 try {
424 List actions = new ArrayList();
425 Object[] results = new Object[(sixtyFourK + 1) * 2];
426
427 for (int i = 0; i < sixtyFourK + 1; i ++) {
428 Put put1 = new Put(ROW);
429 put1.addColumn(FAMILY, QUALIFIER, VALUE);
430 actions.add(put1);
431
432 Put put2 = new Put(ANOTHERROW);
433 put2.addColumn(FAMILY, QUALIFIER, VALUE);
434 actions.add(put2);
435 }
436
437 table.batch(actions, results);
438 } finally {
439 table.close();
440 }
441 }
442
443 @Test
444 public void testHTableExistsMethodSingleRegionSingleGet() throws Exception {
445
446
447
448 Table table = TEST_UTIL.createTable(
449 Bytes.toBytes("testHTableExistsMethodSingleRegionSingleGet"), new byte[][] { FAMILY });
450
451 Put put = new Put(ROW);
452 put.add(FAMILY, QUALIFIER, VALUE);
453
454 Get get = new Get(ROW);
455
456 boolean exist = table.exists(get);
457 assertEquals(exist, false);
458
459 table.put(put);
460
461 exist = table.exists(get);
462 assertEquals(exist, true);
463 }
464
465 public void testHTableExistsMethodSingleRegionMultipleGets() throws Exception {
466
467 HTable table = TEST_UTIL.createTable(
468 Bytes.toBytes("testHTableExistsMethodSingleRegionMultipleGets"), new byte[][] { FAMILY });
469
470 Put put = new Put(ROW);
471 put.add(FAMILY, QUALIFIER, VALUE);
472 table.put(put);
473
474 List<Get> gets = new ArrayList<Get>();
475 gets.add(new Get(ROW));
476 gets.add(null);
477 gets.add(new Get(ANOTHERROW));
478
479 Boolean[] results = table.exists(gets);
480 assertEquals(results[0], true);
481 assertEquals(results[1], false);
482 assertEquals(results[2], false);
483 }
484
485 @Test
486 public void testHTableExistsBeforeGet() throws Exception {
487 Table table = TEST_UTIL.createTable(
488 Bytes.toBytes("testHTableExistsBeforeGet"), new byte[][] { FAMILY });
489 try {
490 Put put = new Put(ROW);
491 put.add(FAMILY, QUALIFIER, VALUE);
492 table.put(put);
493
494 Get get = new Get(ROW);
495
496 boolean exist = table.exists(get);
497 assertEquals(true, exist);
498
499 Result result = table.get(get);
500 assertEquals(false, result.isEmpty());
501 assertTrue(Bytes.equals(VALUE, result.getValue(FAMILY, QUALIFIER)));
502 } finally {
503 table.close();
504 }
505 }
506
507 @Test
508 public void testHTableExistsAllBeforeGet() throws Exception {
509 final byte[] ROW2 = Bytes.add(ROW, Bytes.toBytes("2"));
510 Table table = TEST_UTIL.createTable(
511 Bytes.toBytes("testHTableExistsAllBeforeGet"), new byte[][] { FAMILY });
512 try {
513 Put put = new Put(ROW);
514 put.add(FAMILY, QUALIFIER, VALUE);
515 table.put(put);
516 put = new Put(ROW2);
517 put.add(FAMILY, QUALIFIER, VALUE);
518 table.put(put);
519
520 Get get = new Get(ROW);
521 Get get2 = new Get(ROW2);
522 ArrayList<Get> getList = new ArrayList(2);
523 getList.add(get);
524 getList.add(get2);
525
526 boolean[] exists = table.existsAll(getList);
527 assertEquals(true, exists[0]);
528 assertEquals(true, exists[1]);
529
530 Result[] result = table.get(getList);
531 assertEquals(false, result[0].isEmpty());
532 assertTrue(Bytes.equals(VALUE, result[0].getValue(FAMILY, QUALIFIER)));
533 assertEquals(false, result[1].isEmpty());
534 assertTrue(Bytes.equals(VALUE, result[1].getValue(FAMILY, QUALIFIER)));
535 } finally {
536 table.close();
537 }
538 }
539
540 @Test
541 public void testHTableExistsMethodMultipleRegionsSingleGet() throws Exception {
542
543 Table table = TEST_UTIL.createTable(
544 TableName.valueOf("testHTableExistsMethodMultipleRegionsSingleGet"), new byte[][] { FAMILY },
545 1, new byte[] { 0x00 }, new byte[] { (byte) 0xff }, 255);
546 Put put = new Put(ROW);
547 put.add(FAMILY, QUALIFIER, VALUE);
548
549 Get get = new Get(ROW);
550
551 boolean exist = table.exists(get);
552 assertEquals(exist, false);
553
554 table.put(put);
555
556 exist = table.exists(get);
557 assertEquals(exist, true);
558 }
559
560 @Test
561 public void testHTableExistsMethodMultipleRegionsMultipleGets() throws Exception {
562 HTable table = TEST_UTIL.createTable(
563 TableName.valueOf("testHTableExistsMethodMultipleRegionsMultipleGets"),
564 new byte[][] { FAMILY }, 1, new byte[] { 0x00 }, new byte[] { (byte) 0xff }, 255);
565 Put put = new Put(ROW);
566 put.add(FAMILY, QUALIFIER, VALUE);
567 table.put (put);
568
569 List<Get> gets = new ArrayList<Get>();
570 gets.add(new Get(ANOTHERROW));
571 gets.add(new Get(Bytes.add(ROW, new byte[] { 0x00 })));
572 gets.add(new Get(ROW));
573 gets.add(new Get(Bytes.add(ANOTHERROW, new byte[] { 0x00 })));
574
575 LOG.info("Calling exists");
576 Boolean[] results = table.exists(gets);
577 assertEquals(results[0], false);
578 assertEquals(results[1], false);
579 assertEquals(results[2], true);
580 assertEquals(results[3], false);
581
582
583 put = new Put(new byte[] { 0x00 });
584 put.add(FAMILY, QUALIFIER, VALUE);
585 table.put(put);
586
587 gets = new ArrayList<Get>();
588 gets.add(new Get(new byte[] { 0x00 }));
589 gets.add(new Get(new byte[] { 0x00, 0x00 }));
590 results = table.exists(gets);
591 assertEquals(results[0], true);
592 assertEquals(results[1], false);
593
594
595 put = new Put(new byte[] { (byte) 0xff, (byte) 0xff });
596 put.add(FAMILY, QUALIFIER, VALUE);
597 table.put(put);
598
599 gets = new ArrayList<Get>();
600 gets.add(new Get(new byte[] { (byte) 0xff }));
601 gets.add(new Get(new byte[] { (byte) 0xff, (byte) 0xff }));
602 gets.add(new Get(new byte[] { (byte) 0xff, (byte) 0xff, (byte) 0xff }));
603 results = table.exists(gets);
604 assertEquals(results[0], false);
605 assertEquals(results[1], true);
606 assertEquals(results[2], false);
607 }
608
609 @Test
610 public void testGetEmptyRow() throws Exception {
611
612 Admin admin = TEST_UTIL.getHBaseAdmin();
613 HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(Bytes.toBytes("test")));
614 desc.addFamily(new HColumnDescriptor(FAMILY));
615 admin.createTable(desc);
616 Table table = new HTable(TEST_UTIL.getConfiguration(), desc.getTableName());
617
618 Put put = new Put(ROW_BYTES);
619 put.add(FAMILY, COL_QUAL, VAL_BYTES);
620 table.put(put);
621
622
623 Result res = null;
624 try {
625 res = table.get(new Get(new byte[0]));
626 fail();
627 } catch (IllegalArgumentException e) {
628
629 }
630 assertTrue(res == null);
631 res = table.get(new Get(Bytes.toBytes("r1-not-exist")));
632 assertTrue(res.isEmpty() == true);
633 res = table.get(new Get(ROW_BYTES));
634 assertTrue(Arrays.equals(res.getValue(FAMILY, COL_QUAL), VAL_BYTES));
635 table.close();
636 }
637
638 @Test(timeout = 30000)
639 public void testMultiRowMutations() throws Exception, Throwable {
640 final TableName tableName = TableName.valueOf("testMultiRowMutations");
641 HTableDescriptor desc = new HTableDescriptor(tableName);
642 desc.addCoprocessor(MultiRowMutationEndpoint.class.getName());
643 desc.addCoprocessor(WatiingForMultiMutationsObserver.class.getName());
644 desc.setConfiguration("hbase.rowlock.wait.duration", String.valueOf(5000));
645 desc.addFamily(new HColumnDescriptor(FAMILY));
646 TEST_UTIL.getHBaseAdmin().createTable(desc);
647
648 Configuration copy = new Configuration(TEST_UTIL.getConfiguration());
649 copy.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, 2);
650 try (Connection con = ConnectionFactory.createConnection(copy)) {
651 final byte[] row = Bytes.toBytes("ROW-0");
652 final byte[] rowLocked= Bytes.toBytes("ROW-1");
653 final byte[] value0 = Bytes.toBytes("VALUE-0");
654 final byte[] value1 = Bytes.toBytes("VALUE-1");
655 final byte[] value2 = Bytes.toBytes("VALUE-2");
656 assertNoLocks(tableName);
657 ExecutorService putService = Executors.newSingleThreadExecutor();
658 putService.execute(new Runnable() {
659 @Override
660 public void run() {
661 try (Table table = con.getTable(tableName)) {
662 Put put0 = new Put(rowLocked);
663 put0.addColumn(FAMILY, QUALIFIER, value0);
664
665 table.put(put0);
666 } catch (IOException ex) {
667 throw new RuntimeException(ex);
668 }
669 }
670 });
671 ExecutorService cpService = Executors.newSingleThreadExecutor();
672 cpService.execute(new Runnable() {
673 @Override
674 public void run() {
675 Put put1 = new Put(row);
676 Put put2 = new Put(rowLocked);
677 put1.addColumn(FAMILY, QUALIFIER, value1);
678 put2.addColumn(FAMILY, QUALIFIER, value2);
679 try (Table table = con.getTable(tableName)) {
680 final MultiRowMutationProtos.MutateRowsRequest request
681 = MultiRowMutationProtos.MutateRowsRequest.newBuilder()
682 .addMutationRequest(org.apache.hadoop.hbase.protobuf.ProtobufUtil.toMutation(
683 org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutationProto.MutationType.PUT, put1))
684 .addMutationRequest(org.apache.hadoop.hbase.protobuf.ProtobufUtil.toMutation(
685 org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutationProto.MutationType.PUT, put2))
686 .build();
687 table.coprocessorService(MultiRowMutationProtos.MultiRowMutationService.class, ROW, ROW,
688 new Batch.Call<MultiRowMutationProtos.MultiRowMutationService, MultiRowMutationProtos.MutateRowsResponse>() {
689 public MultiRowMutationProtos.MutateRowsResponse call(MultiRowMutationProtos.MultiRowMutationService instance) throws IOException {
690 ServerRpcController controller = new ServerRpcController();
691 BlockingRpcCallback<MultiRowMutationProtos.MutateRowsResponse> rpcCallback = new BlockingRpcCallback<>();
692 instance.mutateRows(controller, request, rpcCallback);
693 return rpcCallback.get();
694 }
695 });
696 fail("This cp should fail because the target lock is blocked by previous put");
697 } catch (Throwable ex) {
698 }
699 }
700 });
701 cpService.shutdown();
702 cpService.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);
703 WatiingForMultiMutationsObserver observer = find(tableName, WatiingForMultiMutationsObserver.class);
704 observer.latch.countDown();
705 putService.shutdown();
706 putService.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);
707 try (Table table = con.getTable(tableName)) {
708 Get g0 = new Get(row);
709 Get g1 = new Get(rowLocked);
710 Result r0 = table.get(g0);
711 Result r1 = table.get(g1);
712 assertTrue(r0.isEmpty());
713 assertFalse(r1.isEmpty());
714 assertTrue(Bytes.equals(r1.getValue(FAMILY, QUALIFIER), value0));
715 }
716 assertNoLocks(tableName);
717 }
718 }
719
720 private static void assertNoLocks(final TableName tableName) throws IOException, InterruptedException {
721 HRegion region = (HRegion) find(tableName);
722 assertEquals(0, region.getLockedRows().size());
723 }
724 private static Region find(final TableName tableName)
725 throws IOException, InterruptedException {
726 HRegionServer rs = TEST_UTIL.getRSForFirstRegionInTable(tableName);
727 List<Region> regions = rs.getOnlineRegions(tableName);
728 assertEquals(1, regions.size());
729 return regions.get(0);
730 }
731
732 private static <T extends RegionObserver> T find(final TableName tableName,
733 Class<T> clz) throws IOException, InterruptedException {
734 Region region = find(tableName);
735 Coprocessor cp = region.getCoprocessorHost().findCoprocessor(clz.getName());
736 assertTrue("The cp instance should be " + clz.getName()
737 + ", current instance is " + cp.getClass().getName(), clz.isInstance(cp));
738 return clz.cast(cp);
739 }
740
741 public static class WatiingForMultiMutationsObserver extends BaseRegionObserver {
742 final CountDownLatch latch = new CountDownLatch(1);
743 @Override
744 public void preBatchMutate(final ObserverContext<RegionCoprocessorEnvironment> c,
745 final MiniBatchOperationInProgress<Mutation> miniBatchOp) throws IOException {
746 try {
747 latch.await();
748 } catch (InterruptedException ex) {
749 throw new IOException(ex);
750 }
751 }
752 }
753 }