001/** 002 * Licensed to the Apache Software Foundation (ASF) under one 003 * or more contributor license agreements. See the NOTICE file 004 * distributed with this work for additional information 005 * regarding copyright ownership. The ASF licenses this file 006 * to you under the Apache License, Version 2.0 (the 007 * "License"); you may not use this file except in compliance 008 * with the License. You may obtain a copy of the License at 009 * 010 * http://www.apache.org/licenses/LICENSE-2.0 011 * 012 * Unless required by applicable law or agreed to in writing, software 013 * distributed under the License is distributed on an "AS IS" BASIS, 014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 015 * See the License for the specific language governing permissions and 016 * limitations under the License. 017 */ 018package org.apache.hadoop.hbase.client; 019 020import static org.junit.Assert.assertArrayEquals; 021import static org.junit.Assert.assertEquals; 022import static org.junit.Assert.assertNull; 023 024import java.io.IOException; 025import org.apache.hadoop.hbase.HBaseClassTestRule; 026import org.apache.hadoop.hbase.HBaseTestingUtility; 027import org.apache.hadoop.hbase.TableName; 028import org.apache.hadoop.hbase.regionserver.HRegionServer; 029import org.apache.hadoop.hbase.testclassification.ClientTests; 030import org.apache.hadoop.hbase.testclassification.LargeTests; 031import org.apache.hadoop.hbase.util.Bytes; 032import org.junit.AfterClass; 033import org.junit.Before; 034import org.junit.BeforeClass; 035import org.junit.ClassRule; 036import org.junit.Rule; 037import org.junit.Test; 038import org.junit.experimental.categories.Category; 039import org.junit.rules.TestName; 040 041@Category({ LargeTests.class, ClientTests.class }) 042public class TestMvccConsistentScanner { 043 044 @ClassRule 045 public static final HBaseClassTestRule CLASS_RULE = 046 HBaseClassTestRule.forClass(TestMvccConsistentScanner.class); 047 048 private static final HBaseTestingUtility UTIL = new HBaseTestingUtility(); 049 050 private static Connection CONN; 051 052 private static final byte[] CF = Bytes.toBytes("cf"); 053 054 private static final byte[] CQ1 = Bytes.toBytes("cq1"); 055 056 private static final byte[] CQ2 = Bytes.toBytes("cq2"); 057 058 private static final byte[] CQ3 = Bytes.toBytes("cq3"); 059 @Rule 060 public TestName testName = new TestName(); 061 062 private TableName tableName; 063 064 @BeforeClass 065 public static void setUpBeforeClass() throws Exception { 066 UTIL.startMiniCluster(2); 067 CONN = ConnectionFactory.createConnection(UTIL.getConfiguration()); 068 } 069 070 @AfterClass 071 public static void tearDownAfterClass() throws Exception { 072 CONN.close(); 073 UTIL.shutdownMiniCluster(); 074 } 075 076 @Before 077 public void setUp() throws IOException, InterruptedException { 078 tableName = TableName.valueOf(testName.getMethodName().replaceAll("[^0-9a-zA-Z]", "_")); 079 UTIL.createTable(tableName, CF); 080 UTIL.waitTableAvailable(tableName); 081 } 082 083 private void put(byte[] row, byte[] cq, byte[] value) throws IOException { 084 try (Table table = CONN.getTable(tableName)) { 085 table.put(new Put(row).addColumn(CF, cq, value)); 086 } 087 } 088 089 private void move() throws IOException, InterruptedException { 090 RegionInfo region = 091 UTIL.getHBaseCluster().getRegions(tableName).stream().findAny().get().getRegionInfo(); 092 HRegionServer rs = 093 UTIL.getHBaseCluster().getRegionServerThreads().stream().map(t -> t.getRegionServer()) 094 .filter(r -> !r.getOnlineTables().contains(tableName)).findAny().get(); 095 UTIL.getAdmin().move(region.getEncodedNameAsBytes(), 096 Bytes.toBytes(rs.getServerName().getServerName())); 097 while (UTIL.getRSForFirstRegionInTable(tableName) != rs) { 098 Thread.sleep(100); 099 } 100 } 101 102 @Test 103 public void testRowAtomic() throws IOException, InterruptedException { 104 byte[] row = Bytes.toBytes("row"); 105 put(row, CQ1, Bytes.toBytes(1)); 106 put(row, CQ2, Bytes.toBytes(2)); 107 try (Table table = CONN.getTable(tableName); 108 ResultScanner scanner = table.getScanner(new Scan().setBatch(1).setCaching(1))) { 109 Result result = scanner.next(); 110 assertEquals(1, result.rawCells().length); 111 assertEquals(1, Bytes.toInt(result.getValue(CF, CQ1))); 112 move(); 113 put(row, CQ3, Bytes.toBytes(3)); 114 result = scanner.next(); 115 assertEquals(1, result.rawCells().length); 116 assertEquals(2, Bytes.toInt(result.getValue(CF, CQ2))); 117 assertNull(scanner.next()); 118 } 119 } 120 121 @Test 122 public void testCrossRowAtomicInRegion() throws IOException, InterruptedException { 123 put(Bytes.toBytes("row1"), CQ1, Bytes.toBytes(1)); 124 put(Bytes.toBytes("row2"), CQ1, Bytes.toBytes(2)); 125 try (Table table = CONN.getTable(tableName); 126 ResultScanner scanner = table.getScanner(new Scan().setCaching(1))) { 127 Result result = scanner.next(); 128 assertArrayEquals(Bytes.toBytes("row1"), result.getRow()); 129 assertEquals(1, Bytes.toInt(result.getValue(CF, CQ1))); 130 move(); 131 put(Bytes.toBytes("row3"), CQ1, Bytes.toBytes(3)); 132 result = scanner.next(); 133 assertArrayEquals(Bytes.toBytes("row2"), result.getRow()); 134 assertEquals(2, Bytes.toInt(result.getValue(CF, CQ1))); 135 assertNull(scanner.next()); 136 } 137 } 138}