View Javadoc

1   /*
2    * Licensed to the Apache Software Foundation (ASF) under one or more
3    * contributor license agreements. See the NOTICE file distributed with this
4    * work for additional information regarding copyright ownership. The ASF
5    * licenses this file to you under the Apache License, Version 2.0 (the
6    * "License"); you may not use this file except in compliance with the License.
7    * You may obtain a copy of the License at
8    *
9    * http://www.apache.org/licenses/LICENSE-2.0
10   *
11   * Unless required by applicable law or agreed to in writing, software
12   * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
13   * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
14   * License for the specific language governing permissions and limitations
15   * under the License.
16   */
17  package org.apache.hadoop.hbase.client;
18  
19  import static org.junit.Assert.assertEquals;
20  import static org.junit.Assert.assertTrue;
21  
22  import java.util.ArrayList;
23  import java.util.List;
24  
25  import org.apache.commons.logging.Log;
26  import org.apache.commons.logging.LogFactory;
27  import org.apache.hadoop.conf.Configuration;
28  import org.apache.hadoop.hbase.Cell;
29  import org.apache.hadoop.hbase.HBaseTestingUtility;
30  import org.apache.hadoop.hbase.HConstants;
31  import org.apache.hadoop.hbase.HRegionInfo;
32  import org.apache.hadoop.hbase.HRegionLocation;
33  import org.apache.hadoop.hbase.HTestConst;
34  import org.apache.hadoop.hbase.KeyValue;
35  import org.apache.hadoop.hbase.testclassification.MediumTests;
36  import org.apache.hadoop.hbase.MiniHBaseCluster;
37  import org.apache.hadoop.hbase.filter.ColumnPrefixFilter;
38  import org.apache.hadoop.hbase.filter.ColumnRangeFilter;
39  import org.apache.hadoop.hbase.master.HMaster;
40  import org.apache.hadoop.hbase.master.RegionState.State;
41  import org.apache.hadoop.hbase.master.RegionStates;
42  import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
43  import org.apache.hadoop.hbase.regionserver.HRegionServer;
44  import org.apache.hadoop.hbase.TableName;
45  import org.apache.hadoop.hbase.util.Bytes;
46  import org.apache.hadoop.hbase.util.ConfigUtil;
47  import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
48  import org.apache.hadoop.hbase.zookeeper.ZKAssign;
49  import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
50  import org.junit.After;
51  import org.junit.AfterClass;
52  import org.junit.Before;
53  import org.junit.BeforeClass;
54  import org.junit.Test;
55  import org.junit.experimental.categories.Category;
56  
57  /**
58   * A client-side test, mostly testing scanners with various parameters.
59   */
60  @Category(MediumTests.class)
61  public class TestScannersFromClientSide {
62    private static final Log LOG = LogFactory.getLog(TestScannersFromClientSide.class);
63  
64    private final static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
65    private static byte [] ROW = Bytes.toBytes("testRow");
66    private static byte [] FAMILY = Bytes.toBytes("testFamily");
67    private static byte [] QUALIFIER = Bytes.toBytes("testQualifier");
68    private static byte [] VALUE = Bytes.toBytes("testValue");
69  
70    /**
71     * @throws java.lang.Exception
72     */
73    @BeforeClass
74    public static void setUpBeforeClass() throws Exception {
75      Configuration conf = TEST_UTIL.getConfiguration();
76      conf.setLong(HConstants.HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE_KEY, 10 * 1024 * 1024);
77      TEST_UTIL.startMiniCluster(3);
78    }
79  
80    /**
81     * @throws java.lang.Exception
82     */
83    @AfterClass
84    public static void tearDownAfterClass() throws Exception {
85      TEST_UTIL.shutdownMiniCluster();
86    }
87  
88    /**
89     * @throws java.lang.Exception
90     */
91    @Before
92    public void setUp() throws Exception {
93      // Nothing to do.
94    }
95  
96    /**
97     * @throws java.lang.Exception
98     */
99    @After
100   public void tearDown() throws Exception {
101     // Nothing to do.
102   }
103 
104   /**
105    * Test from client side for batch of scan
106    *
107    * @throws Exception
108    */
109   @Test
110   public void testScanBatch() throws Exception {
111     byte [] TABLE = Bytes.toBytes("testScanBatch");
112     byte [][] QUALIFIERS = HTestConst.makeNAscii(QUALIFIER, 8);
113 
114     HTable ht = TEST_UTIL.createTable(TABLE, FAMILY);
115 
116     Put put;
117     Scan scan;
118     Delete delete;
119     Result result;
120     ResultScanner scanner;
121     boolean toLog = true;
122     List<Cell> kvListExp;
123 
124     // table: row, family, c0:0, c1:1, ... , c7:7
125     put = new Put(ROW);
126     for (int i=0; i < QUALIFIERS.length; i++) {
127       KeyValue kv = new KeyValue(ROW, FAMILY, QUALIFIERS[i], i, VALUE);
128       put.add(kv);
129     }
130     ht.put(put);
131 
132     // table: row, family, c0:0, c1:1, ..., c6:2, c6:6 , c7:7
133     put = new Put(ROW);
134     KeyValue kv = new KeyValue(ROW, FAMILY, QUALIFIERS[6], 2, VALUE);
135     put.add(kv);
136     ht.put(put);
137 
138     // delete upto ts: 3
139     delete = new Delete(ROW);
140     delete.deleteFamily(FAMILY, 3);
141     ht.delete(delete);
142 
143     // without batch
144     scan = new Scan(ROW);
145     scan.setMaxVersions();
146     scanner = ht.getScanner(scan);
147 
148     // c4:4, c5:5, c6:6, c7:7
149     kvListExp = new ArrayList<Cell>();
150     kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[4], 4, VALUE));
151     kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[5], 5, VALUE));
152     kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[6], 6, VALUE));
153     kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[7], 7, VALUE));
154     result = scanner.next();
155     verifyResult(result, kvListExp, toLog, "Testing first batch of scan");
156 
157     // with batch
158     scan = new Scan(ROW);
159     scan.setMaxVersions();
160     scan.setBatch(2);
161     scanner = ht.getScanner(scan);
162 
163     // First batch: c4:4, c5:5
164     kvListExp = new ArrayList<Cell>();
165     kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[4], 4, VALUE));
166     kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[5], 5, VALUE));
167     result = scanner.next();
168     verifyResult(result, kvListExp, toLog, "Testing first batch of scan");
169 
170     // Second batch: c6:6, c7:7
171     kvListExp = new ArrayList<Cell>();
172     kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[6], 6, VALUE));
173     kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[7], 7, VALUE));
174     result = scanner.next();
175     verifyResult(result, kvListExp, toLog, "Testing second batch of scan");
176 
177   }
178 
179   @Test
180   public void testSmallScan() throws Exception {
181     TableName TABLE = TableName.valueOf("testSmallScan");
182 
183     int numRows = 10;
184     byte[][] ROWS = HTestConst.makeNAscii(ROW, numRows);
185 
186     int numQualifiers = 10;
187     byte[][] QUALIFIERS = HTestConst.makeNAscii(QUALIFIER, numQualifiers);
188 
189     HTable ht = TEST_UTIL.createTable(TABLE, FAMILY);
190 
191     Put put;
192     List<Put> puts = new ArrayList<Put>();
193     for (int row = 0; row < ROWS.length; row++) {
194       put = new Put(ROWS[row]);
195       for (int qual = 0; qual < QUALIFIERS.length; qual++) {
196         KeyValue kv = new KeyValue(ROWS[row], FAMILY, QUALIFIERS[qual], VALUE);
197         put.add(kv);
198       }
199       puts.add(put);
200     }
201     ht.put(puts);
202 
203     int expectedRows = numRows;
204     int expectedCols = numRows * numQualifiers;
205 
206     // Test normal and reversed
207     testSmallScan(ht, true, expectedRows, expectedCols);
208     testSmallScan(ht, false, expectedRows, expectedCols);
209   }
210 
211   /**
212    * Run through a variety of test configurations with a small scan
213    * @param table
214    * @param reversed
215    * @param rows
216    * @param columns
217    * @throws Exception
218    */
219   public void testSmallScan(HTable table, boolean reversed, int rows, int columns) throws Exception {
220     Scan baseScan = new Scan();
221     baseScan.setReversed(reversed);
222     baseScan.setSmall(true);
223 
224     Scan scan = new Scan(baseScan);
225     verifyExpectedCounts(table, scan, rows, columns);
226 
227     scan = new Scan(baseScan);
228     scan.setMaxResultSize(1);
229     verifyExpectedCounts(table, scan, rows, columns);
230 
231     scan = new Scan(baseScan);
232     scan.setMaxResultSize(1);
233     scan.setCaching(Integer.MAX_VALUE);
234     verifyExpectedCounts(table, scan, rows, columns);
235   }
236 
237   private void verifyExpectedCounts(HTable table, Scan scan, int expectedRowCount,
238       int expectedCellCount) throws Exception {
239     ResultScanner scanner = table.getScanner(scan);
240     
241     int rowCount = 0;
242     int cellCount = 0;
243     Result r = null;
244     while ((r = scanner.next()) != null) {
245       rowCount++;
246       for (Cell c : r.rawCells()) {
247         cellCount++;
248       }
249     }
250 
251     assertTrue("Expected row count: " + expectedRowCount + " Actual row count: " + rowCount,
252       expectedRowCount == rowCount);
253     assertTrue("Expected cell count: " + expectedCellCount + " Actual cell count: " + cellCount,
254       expectedCellCount == cellCount);
255     scanner.close();
256   }
257 
258   /**
259    * Test from client side for get with maxResultPerCF set
260    *
261    * @throws Exception
262    */
263   @Test
264   public void testGetMaxResults() throws Exception {
265     byte [] TABLE = Bytes.toBytes("testGetMaxResults");
266     byte [][] FAMILIES = HTestConst.makeNAscii(FAMILY, 3);
267     byte [][] QUALIFIERS = HTestConst.makeNAscii(QUALIFIER, 20);
268 
269     HTable ht = TEST_UTIL.createTable(TABLE, FAMILIES);
270 
271     Get get;
272     Put put;
273     Result result;
274     boolean toLog = true;
275     List<Cell> kvListExp;
276 
277     kvListExp = new ArrayList<Cell>();
278     // Insert one CF for row[0]
279     put = new Put(ROW);
280     for (int i=0; i < 10; i++) {
281       KeyValue kv = new KeyValue(ROW, FAMILIES[0], QUALIFIERS[i], 1, VALUE);
282       put.add(kv);
283       kvListExp.add(kv);
284     }
285     ht.put(put);
286 
287     get = new Get(ROW);
288     result = ht.get(get);
289     verifyResult(result, kvListExp, toLog, "Testing without setting maxResults");
290 
291     get = new Get(ROW);
292     get.setMaxResultsPerColumnFamily(2);
293     result = ht.get(get);
294     kvListExp = new ArrayList<Cell>();
295     kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[0], 1, VALUE));
296     kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[1], 1, VALUE));
297     verifyResult(result, kvListExp, toLog, "Testing basic setMaxResults");
298 
299     // Filters: ColumnRangeFilter
300     get = new Get(ROW);
301     get.setMaxResultsPerColumnFamily(5);
302     get.setFilter(new ColumnRangeFilter(QUALIFIERS[2], true, QUALIFIERS[5],
303                                         true));
304     result = ht.get(get);
305     kvListExp = new ArrayList<Cell>();
306     kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[2], 1, VALUE));
307     kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[3], 1, VALUE));
308     kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[4], 1, VALUE));
309     kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[5], 1, VALUE));
310     verifyResult(result, kvListExp, toLog, "Testing single CF with CRF");
311 
312     // Insert two more CF for row[0]
313     // 20 columns for CF2, 10 columns for CF1
314     put = new Put(ROW);
315     for (int i=0; i < QUALIFIERS.length; i++) {
316       KeyValue kv = new KeyValue(ROW, FAMILIES[2], QUALIFIERS[i], 1, VALUE);
317       put.add(kv);
318     }
319     ht.put(put);
320 
321     put = new Put(ROW);
322     for (int i=0; i < 10; i++) {
323       KeyValue kv = new KeyValue(ROW, FAMILIES[1], QUALIFIERS[i], 1, VALUE);
324       put.add(kv);
325     }
326     ht.put(put);
327 
328     get = new Get(ROW);
329     get.setMaxResultsPerColumnFamily(12);
330     get.addFamily(FAMILIES[1]);
331     get.addFamily(FAMILIES[2]);
332     result = ht.get(get);
333     kvListExp = new ArrayList<Cell>();
334     //Exp: CF1:q0, ..., q9, CF2: q0, q1, q10, q11, ..., q19
335     for (int i=0; i < 10; i++) {
336       kvListExp.add(new KeyValue(ROW, FAMILIES[1], QUALIFIERS[i], 1, VALUE));
337     }
338     for (int i=0; i < 2; i++) {
339         kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[i], 1, VALUE));
340       }
341     for (int i=10; i < 20; i++) {
342       kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[i], 1, VALUE));
343     }
344     verifyResult(result, kvListExp, toLog, "Testing multiple CFs");
345 
346     // Filters: ColumnRangeFilter and ColumnPrefixFilter
347     get = new Get(ROW);
348     get.setMaxResultsPerColumnFamily(3);
349     get.setFilter(new ColumnRangeFilter(QUALIFIERS[2], true, null, true));
350     result = ht.get(get);
351     kvListExp = new ArrayList<Cell>();
352     for (int i=2; i < 5; i++) {
353       kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[i], 1, VALUE));
354     }
355     for (int i=2; i < 5; i++) {
356       kvListExp.add(new KeyValue(ROW, FAMILIES[1], QUALIFIERS[i], 1, VALUE));
357     }
358     for (int i=2; i < 5; i++) {
359       kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[i], 1, VALUE));
360     }
361     verifyResult(result, kvListExp, toLog, "Testing multiple CFs + CRF");
362 
363     get = new Get(ROW);
364     get.setMaxResultsPerColumnFamily(7);
365     get.setFilter(new ColumnPrefixFilter(QUALIFIERS[1]));
366     result = ht.get(get);
367     kvListExp = new ArrayList<Cell>();
368     kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[1], 1, VALUE));
369     kvListExp.add(new KeyValue(ROW, FAMILIES[1], QUALIFIERS[1], 1, VALUE));
370     kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[1], 1, VALUE));
371     for (int i=10; i < 16; i++) {
372       kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[i], 1, VALUE));
373     }
374     verifyResult(result, kvListExp, toLog, "Testing multiple CFs + PFF");
375 
376   }
377 
378   /**
379    * Test from client side for scan with maxResultPerCF set
380    *
381    * @throws Exception
382    */
383   @Test
384   public void testScanMaxResults() throws Exception {
385     byte [] TABLE = Bytes.toBytes("testScanLimit");
386     byte [][] ROWS = HTestConst.makeNAscii(ROW, 2);
387     byte [][] FAMILIES = HTestConst.makeNAscii(FAMILY, 3);
388     byte [][] QUALIFIERS = HTestConst.makeNAscii(QUALIFIER, 10);
389 
390     HTable ht = TEST_UTIL.createTable(TABLE, FAMILIES);
391 
392     Put put;
393     Scan scan;
394     Result result;
395     boolean toLog = true;
396     List<Cell> kvListExp, kvListScan;
397 
398     kvListExp = new ArrayList<Cell>();
399 
400     for (int r=0; r < ROWS.length; r++) {
401       put = new Put(ROWS[r]);
402       for (int c=0; c < FAMILIES.length; c++) {
403         for (int q=0; q < QUALIFIERS.length; q++) {
404           KeyValue kv = new KeyValue(ROWS[r], FAMILIES[c], QUALIFIERS[q], 1, VALUE);
405           put.add(kv);
406           if (q < 4) {
407             kvListExp.add(kv);
408           }
409         }
410       }
411       ht.put(put);
412     }
413 
414     scan = new Scan();
415     scan.setMaxResultsPerColumnFamily(4);
416     ResultScanner scanner = ht.getScanner(scan);
417     kvListScan = new ArrayList<Cell>();
418     while ((result = scanner.next()) != null) {
419       for (Cell kv : result.listCells()) {
420         kvListScan.add(kv);
421       }
422     }
423     result = Result.create(kvListScan);
424     verifyResult(result, kvListExp, toLog, "Testing scan with maxResults");
425 
426   }
427 
428   /**
429    * Test from client side for get with rowOffset
430    *
431    * @throws Exception
432    */
433   @Test
434   public void testGetRowOffset() throws Exception {
435     byte [] TABLE = Bytes.toBytes("testGetRowOffset");
436     byte [][] FAMILIES = HTestConst.makeNAscii(FAMILY, 3);
437     byte [][] QUALIFIERS = HTestConst.makeNAscii(QUALIFIER, 20);
438 
439     HTable ht = TEST_UTIL.createTable(TABLE, FAMILIES);
440 
441     Get get;
442     Put put;
443     Result result;
444     boolean toLog = true;
445     List<Cell> kvListExp;
446 
447     // Insert one CF for row
448     kvListExp = new ArrayList<Cell>();
449     put = new Put(ROW);
450     for (int i=0; i < 10; i++) {
451       KeyValue kv = new KeyValue(ROW, FAMILIES[0], QUALIFIERS[i], 1, VALUE);
452       put.add(kv);
453       // skipping first two kvs
454       if (i < 2) continue;
455       kvListExp.add(kv);
456     }
457     ht.put(put);
458 
459     //setting offset to 2
460     get = new Get(ROW);
461     get.setRowOffsetPerColumnFamily(2);
462     result = ht.get(get);
463     verifyResult(result, kvListExp, toLog, "Testing basic setRowOffset");
464 
465     //setting offset to 20
466     get = new Get(ROW);
467     get.setRowOffsetPerColumnFamily(20);
468     result = ht.get(get);
469     kvListExp = new ArrayList<Cell>();
470     verifyResult(result, kvListExp, toLog, "Testing offset > #kvs");
471 
472     //offset + maxResultPerCF
473     get = new Get(ROW);
474     get.setRowOffsetPerColumnFamily(4);
475     get.setMaxResultsPerColumnFamily(5);
476     result = ht.get(get);
477     kvListExp = new ArrayList<Cell>();
478     for (int i=4; i < 9; i++) {
479       kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[i], 1, VALUE));
480     }
481     verifyResult(result, kvListExp, toLog,
482       "Testing offset + setMaxResultsPerCF");
483 
484     // Filters: ColumnRangeFilter
485     get = new Get(ROW);
486     get.setRowOffsetPerColumnFamily(1);
487     get.setFilter(new ColumnRangeFilter(QUALIFIERS[2], true, QUALIFIERS[5],
488                                         true));
489     result = ht.get(get);
490     kvListExp = new ArrayList<Cell>();
491     kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[3], 1, VALUE));
492     kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[4], 1, VALUE));
493     kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[5], 1, VALUE));
494     verifyResult(result, kvListExp, toLog, "Testing offset with CRF");
495 
496     // Insert into two more CFs for row
497     // 10 columns for CF2, 10 columns for CF1
498     for(int j=2; j > 0; j--) {
499       put = new Put(ROW);
500       for (int i=0; i < 10; i++) {
501         KeyValue kv = new KeyValue(ROW, FAMILIES[j], QUALIFIERS[i], 1, VALUE);
502         put.add(kv);
503       }
504       ht.put(put);
505     }
506 
507     get = new Get(ROW);
508     get.setRowOffsetPerColumnFamily(4);
509     get.setMaxResultsPerColumnFamily(2);
510     get.addFamily(FAMILIES[1]);
511     get.addFamily(FAMILIES[2]);
512     result = ht.get(get);
513     kvListExp = new ArrayList<Cell>();
514     //Exp: CF1:q4, q5, CF2: q4, q5
515     kvListExp.add(new KeyValue(ROW, FAMILIES[1], QUALIFIERS[4], 1, VALUE));
516     kvListExp.add(new KeyValue(ROW, FAMILIES[1], QUALIFIERS[5], 1, VALUE));
517     kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[4], 1, VALUE));
518     kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[5], 1, VALUE));
519     verifyResult(result, kvListExp, toLog,
520        "Testing offset + multiple CFs + maxResults");
521   }
522 
523   /**
524    * Test from client side for scan while the region is reopened
525    * on the same region server.
526    *
527    * @throws Exception
528    */
529   @Test
530   public void testScanOnReopenedRegion() throws Exception {
531     byte [] TABLE = Bytes.toBytes("testScanOnReopenedRegion");
532     byte [][] QUALIFIERS = HTestConst.makeNAscii(QUALIFIER, 2);
533 
534     HTable ht = TEST_UTIL.createTable(TABLE, FAMILY);
535 
536     Put put;
537     Scan scan;
538     Result result;
539     ResultScanner scanner;
540     boolean toLog = false;
541     List<Cell> kvListExp;
542 
543     // table: row, family, c0:0, c1:1
544     put = new Put(ROW);
545     for (int i=0; i < QUALIFIERS.length; i++) {
546       KeyValue kv = new KeyValue(ROW, FAMILY, QUALIFIERS[i], i, VALUE);
547       put.add(kv);
548     }
549     ht.put(put);
550 
551     scan = new Scan(ROW);
552     scanner = ht.getScanner(scan);
553 
554     HRegionLocation loc = ht.getRegionLocation(ROW);
555     HRegionInfo hri = loc.getRegionInfo();
556     MiniHBaseCluster cluster = TEST_UTIL.getMiniHBaseCluster();
557     byte[] regionName = hri.getRegionName();
558     int i = cluster.getServerWith(regionName);
559     HRegionServer rs = cluster.getRegionServer(i);
560     ProtobufUtil.closeRegion(rs, rs.getServerName(), regionName, false);
561     long startTime = EnvironmentEdgeManager.currentTimeMillis();
562     long timeOut = 300000;
563     while (true) {
564       if (rs.getOnlineRegion(regionName) == null) {
565         break;
566       }
567       assertTrue("Timed out in closing the testing region",
568         EnvironmentEdgeManager.currentTimeMillis() < startTime + timeOut);
569       Thread.sleep(500);
570     }
571 
572     // Now open the region again.
573     ZooKeeperWatcher zkw = TEST_UTIL.getZooKeeperWatcher();
574     try {
575       HMaster master = cluster.getMaster();
576       RegionStates states = master.getAssignmentManager().getRegionStates();
577       states.regionOffline(hri);
578       states.updateRegionState(hri, State.OPENING);
579       if (ConfigUtil.useZKForAssignment(TEST_UTIL.getConfiguration())) {
580         ZKAssign.createNodeOffline(zkw, hri, loc.getServerName());
581       }
582       ProtobufUtil.openRegion(rs, rs.getServerName(), hri);
583       startTime = EnvironmentEdgeManager.currentTimeMillis();
584       while (true) {
585         if (rs.getOnlineRegion(regionName) != null) {
586           break;
587         }
588         assertTrue("Timed out in open the testing region",
589           EnvironmentEdgeManager.currentTimeMillis() < startTime + timeOut);
590         Thread.sleep(500);
591       }
592     } finally {
593       ZKAssign.deleteNodeFailSilent(zkw, hri);
594     }
595 
596     // c0:0, c1:1
597     kvListExp = new ArrayList<Cell>();
598     kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[0], 0, VALUE));
599     kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[1], 1, VALUE));
600     result = scanner.next();
601     verifyResult(result, kvListExp, toLog, "Testing scan on re-opened region");
602   }
603 
604   static void verifyResult(Result result, List<Cell> expKvList, boolean toLog,
605       String msg) {
606 
607     LOG.info(msg);
608     LOG.info("Expected count: " + expKvList.size());
609     LOG.info("Actual count: " + result.size());
610     if (expKvList.size() == 0)
611       return;
612 
613     int i = 0;
614     for (Cell kv : result.rawCells()) {
615       if (i >= expKvList.size()) {
616         break;  // we will check the size later
617       }
618 
619       Cell kvExp = expKvList.get(i++);
620       if (toLog) {
621         LOG.info("get kv is: " + kv.toString());
622         LOG.info("exp kv is: " + kvExp.toString());
623       }
624       assertTrue("Not equal", kvExp.equals(kv));
625     }
626 
627     assertEquals(expKvList.size(), result.size());
628   }
629 
630 
631 }