1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.hbase.client;
19
20 import static org.junit.Assert.assertTrue;
21
22 import java.io.IOException;
23
24 import org.apache.commons.logging.Log;
25 import org.apache.commons.logging.LogFactory;
26 import org.apache.commons.logging.impl.Log4JLogger;
27 import org.apache.hadoop.conf.Configuration;
28 import org.apache.hadoop.hbase.HBaseTestingUtility;
29 import org.apache.hadoop.hbase.HConstants;
30 import org.apache.hadoop.hbase.MediumTests;
31 import org.apache.hadoop.hbase.MiniHBaseCluster.MiniHBaseClusterRegionServer;
32 import org.apache.hadoop.hbase.TableName;
33 import org.apache.hadoop.hbase.ipc.RpcClient;
34 import org.apache.hadoop.hbase.ipc.RpcServer;
35 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ScanRequest;
36 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ScanResponse;
37 import org.apache.hadoop.hbase.util.Bytes;
38 import org.apache.log4j.Level;
39 import org.junit.AfterClass;
40 import org.junit.BeforeClass;
41 import org.junit.Test;
42 import org.junit.experimental.categories.Category;
43
44 import com.google.protobuf.RpcController;
45 import com.google.protobuf.ServiceException;
46
47
48
49
50
51 @Category(MediumTests.class)
52 public class TestClientScannerRPCTimeout {
53 final Log LOG = LogFactory.getLog(getClass());
54 private final static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
55 private static final byte[] FAMILY = Bytes.toBytes("testFamily");
56 private static final byte[] QUALIFIER = Bytes.toBytes("testQualifier");
57 private static final byte[] VALUE = Bytes.toBytes("testValue");
58 private static final int rpcTimeout = 2 * 1000;
59 private static final int CLIENT_RETRIES_NUMBER = 3;
60
61 @BeforeClass
62 public static void setUpBeforeClass() throws Exception {
63 ((Log4JLogger)RpcServer.LOG).getLogger().setLevel(Level.ALL);
64 ((Log4JLogger)RpcClient.LOG).getLogger().setLevel(Level.ALL);
65 ((Log4JLogger)ScannerCallable.LOG).getLogger().setLevel(Level.ALL);
66 Configuration conf = TEST_UTIL.getConfiguration();
67
68 conf.setInt("hbase.regionserver.msginterval", 3 * 10000);
69 conf.setInt(HConstants.HBASE_RPC_TIMEOUT_KEY, rpcTimeout);
70 conf.setStrings(HConstants.REGION_SERVER_IMPL, RegionServerWithScanTimeout.class.getName());
71 conf.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, CLIENT_RETRIES_NUMBER);
72 conf.setInt(HConstants.HBASE_CLIENT_PAUSE, 1000);
73 TEST_UTIL.startMiniCluster(1);
74 }
75
76 @AfterClass
77 public static void tearDownAfterClass() throws Exception {
78 TEST_UTIL.shutdownMiniCluster();
79 }
80
81 @Test
82 public void testScannerNextRPCTimesout() throws Exception {
83 final byte[] TABLE_NAME = Bytes.toBytes("testScannerNextRPCTimesout");
84 HTable ht = TEST_UTIL.createTable(TABLE_NAME, FAMILY);
85 byte[] r1 = Bytes.toBytes("row-1");
86 byte[] r2 = Bytes.toBytes("row-2");
87 byte[] r3 = Bytes.toBytes("row-3");
88 putToTable(ht, r1);
89 putToTable(ht, r2);
90 putToTable(ht, r3);
91 LOG.info("Wrote our three values");
92 RegionServerWithScanTimeout.seqNoToSleepOn = 1;
93 Scan scan = new Scan();
94 scan.setCaching(1);
95 ResultScanner scanner = ht.getScanner(scan);
96 Result result = scanner.next();
97 assertTrue("Expected row: row-1", Bytes.equals(r1, result.getRow()));
98 LOG.info("Got expected first row");
99 long t1 = System.currentTimeMillis();
100 result = scanner.next();
101 assertTrue((System.currentTimeMillis() - t1) > rpcTimeout);
102 assertTrue("Expected row: row-2", Bytes.equals(r2, result.getRow()));
103 RegionServerWithScanTimeout.seqNoToSleepOn = -1;
104 result = scanner.next();
105 assertTrue("Expected row: row-3", Bytes.equals(r3, result.getRow()));
106 scanner.close();
107
108
109 scanner = ht.getScanner(scan);
110 RegionServerWithScanTimeout.sleepAlways = true;
111 RegionServerWithScanTimeout.tryNumber = 0;
112 try {
113 result = scanner.next();
114 } catch (IOException ioe) {
115
116 LOG.info("Failed after maximal attempts=" + CLIENT_RETRIES_NUMBER, ioe);
117 }
118 assertTrue("Expected maximal try number=" + CLIENT_RETRIES_NUMBER
119 + ", actual =" + RegionServerWithScanTimeout.tryNumber,
120 RegionServerWithScanTimeout.tryNumber <= CLIENT_RETRIES_NUMBER);
121 }
122
123 private void putToTable(HTable ht, byte[] rowkey) throws IOException {
124 Put put = new Put(rowkey);
125 put.add(FAMILY, QUALIFIER, VALUE);
126 ht.put(put);
127 }
128
129 private static class RegionServerWithScanTimeout extends MiniHBaseClusterRegionServer {
130 private long tableScannerId;
131 private boolean slept;
132 private static long seqNoToSleepOn = -1;
133 private static boolean sleepAlways = false;
134 private static int tryNumber = 0;
135
136 public RegionServerWithScanTimeout(Configuration conf)
137 throws IOException, InterruptedException {
138 super(conf);
139 }
140
141 @Override
142 public ScanResponse scan(final RpcController controller, final ScanRequest request)
143 throws ServiceException {
144 if (request.hasScannerId()) {
145 ScanResponse scanResponse = super.scan(controller, request);
146 if (this.tableScannerId == request.getScannerId() &&
147 (sleepAlways || (!slept && seqNoToSleepOn == request.getNextCallSeq()))) {
148 try {
149 LOG.info("SLEEPING " + (rpcTimeout + 500));
150 Thread.sleep(rpcTimeout + 500);
151 } catch (InterruptedException e) {
152 }
153 slept = true;
154 tryNumber++;
155 if (tryNumber > 2 * CLIENT_RETRIES_NUMBER) {
156 sleepAlways = false;
157 }
158 }
159 return scanResponse;
160 } else {
161 ScanResponse scanRes = super.scan(controller, request);
162 String regionName = Bytes.toString(request.getRegion().getValue().toByteArray());
163 if (!regionName.contains(TableName.META_TABLE_NAME.getNameAsString())) {
164 tableScannerId = scanRes.getScannerId();
165 }
166 return scanRes;
167 }
168 }
169 }
170 }