You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by bb...@apache.org on 2022/12/15 17:04:09 UTC
[hbase] 29/91: HubSpot Backport: HBASE-26784 Use HIGH_QOS for ResultScanner.close requests (#4163)
This is an automated email from the ASF dual-hosted git repository.
bbeaudreault pushed a commit to branch hubspot-2
in repository https://gitbox.apache.org/repos/asf/hbase.git
commit 8a2782158273b9b0fa0f0bb1d60bc010a65cf9fb
Author: Bryan Beaudreault <bb...@hubspot.com>
AuthorDate: Mon Mar 7 21:50:00 2022 -0500
HubSpot Backport: HBASE-26784 Use HIGH_QOS for ResultScanner.close requests (#4163)
Signed-off-by: Duo Zhang <zh...@apache.org>
Signed-off-by: Andrew Purtell <ap...@apache.org>
Signed-off-by: Xiaolin Ha <ha...@apache.org>
---
.../hadoop/hbase/client/AsyncClientScanner.java | 1 +
.../AsyncScanSingleRegionRpcRetryingCaller.java | 3 +-
.../hadoop/hbase/client/ScannerCallable.java | 12 +-
.../hbase/client/TestAsyncTableRpcPriority.java | 180 +++++++++++------
.../hadoop/hbase/client/TestTableRpcPriority.java | 222 +++++++++++++++++++++
.../hbase/client/AbstractTestAsyncTableScan.java | 35 +++-
.../hadoop/hbase/client/TestAsyncTableScan.java | 68 ++++++-
.../hadoop/hbase/client/TestAsyncTableScanAll.java | 7 +-
.../hadoop/hbase/client/TestAsyncTableScanner.java | 11 +-
.../hadoop/hbase/client/TestRawAsyncTableScan.java | 11 +-
10 files changed, 469 insertions(+), 81 deletions(-)
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncClientScanner.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncClientScanner.java
index 5fd00a5f6ed..48f004c0a29 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncClientScanner.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncClientScanner.java
@@ -188,6 +188,7 @@ class AsyncClientScanner {
private CompletableFuture<OpenScannerResponse> openScanner(int replicaId) {
return conn.callerFactory.<OpenScannerResponse> single().table(tableName)
.row(scan.getStartRow()).replicaId(replicaId).locateType(getLocateType(scan))
+ .priority(scan.getPriority())
.rpcTimeout(rpcTimeoutNs, TimeUnit.NANOSECONDS)
.operationTimeout(scanTimeoutNs, TimeUnit.NANOSECONDS).pause(pauseNs, TimeUnit.NANOSECONDS)
.pauseForCQTBE(pauseForCQTBENs, TimeUnit.NANOSECONDS).maxAttempts(maxAttempts)
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncScanSingleRegionRpcRetryingCaller.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncScanSingleRegionRpcRetryingCaller.java
index 1fa3c81e5d1..7f19180a0ab 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncScanSingleRegionRpcRetryingCaller.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncScanSingleRegionRpcRetryingCaller.java
@@ -36,6 +36,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import org.apache.hadoop.hbase.CallQueueTooBigException;
import org.apache.hadoop.hbase.DoNotRetryIOException;
+import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HRegionLocation;
import org.apache.hadoop.hbase.NotServingRegionException;
import org.apache.hadoop.hbase.UnknownScannerException;
@@ -347,7 +348,7 @@ class AsyncScanSingleRegionRpcRetryingCaller {
private void closeScanner() {
incRPCCallsMetrics(scanMetrics, regionServerRemote);
- resetController(controller, rpcTimeoutNs, priority);
+ resetController(controller, rpcTimeoutNs, HConstants.HIGH_QOS);
ScanRequest req = RequestConverter.buildScanRequest(this.scannerId, 0, true, false);
stub.scan(controller, req, resp -> {
if (controller.failed()) {
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ScannerCallable.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ScannerCallable.java
index 7ca5a800b0e..6e0a51f8795 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ScannerCallable.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ScannerCallable.java
@@ -29,6 +29,7 @@ import java.io.InterruptedIOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.DoNotRetryIOException;
import org.apache.hadoop.hbase.HBaseIOException;
+import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.HRegionLocation;
import org.apache.hadoop.hbase.NotServingRegionException;
@@ -38,6 +39,7 @@ import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.UnknownScannerException;
import org.apache.hadoop.hbase.client.metrics.ScanMetrics;
import org.apache.hadoop.hbase.exceptions.ScannerResetException;
+import org.apache.hadoop.hbase.ipc.HBaseRpcController;
import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
import org.apache.hadoop.hbase.regionserver.RegionServerStoppedException;
import org.apache.yetus.audience.InterfaceAudience;
@@ -308,8 +310,16 @@ public class ScannerCallable extends ClientServiceCallable<Result[]> {
incRPCCallsMetrics(scanMetrics, isRegionServerRemote);
ScanRequest request =
RequestConverter.buildScanRequest(this.scannerId, 0, true, this.scanMetrics != null);
+ HBaseRpcController controller = rpcControllerFactory.newController();
+ // pull in the original priority, but then try to set to HIGH.
+ // it will take whatever is highest.
+ controller.setPriority(controller.getPriority());
+ controller.setPriority(HConstants.HIGH_QOS);
+ if (controller.hasCallTimeout()) {
+ controller.setCallTimeout(controller.getCallTimeout());
+ }
try {
- getStub().scan(getRpcController(), request);
+ getStub().scan(controller, request);
} catch (Exception e) {
throw ProtobufUtil.handleRemoteException(e);
}
diff --git a/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableRpcPriority.java b/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableRpcPriority.java
index 8c22946649d..1e6a4b345e7 100644
--- a/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableRpcPriority.java
+++ b/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableRpcPriority.java
@@ -17,10 +17,14 @@
*/
package org.apache.hadoop.hbase.client;
+import static org.apache.hadoop.hbase.HConstants.HIGH_QOS;
import static org.apache.hadoop.hbase.HConstants.NORMAL_QOS;
import static org.apache.hadoop.hbase.HConstants.SYSTEMTABLE_QOS;
import static org.apache.hadoop.hbase.NamespaceDescriptor.SYSTEM_NAMESPACE_NAME_STR;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyLong;
@@ -33,8 +37,13 @@ import static org.mockito.Mockito.verify;
import java.io.IOException;
import java.util.Arrays;
+import java.util.Optional;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.Cell.Type;
@@ -59,9 +68,7 @@ import org.junit.rules.TestName;
import org.mockito.ArgumentMatcher;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
-
import org.apache.hbase.thirdparty.com.google.protobuf.RpcCallback;
-
import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos;
import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.ClientService;
@@ -77,6 +84,7 @@ import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.ResultOrEx
import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.ScanRequest;
import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.ScanResponse;
+
/**
* Confirm that we will set the priority in {@link HBaseRpcController} for several table operations.
*/
@@ -91,6 +99,8 @@ public class TestAsyncTableRpcPriority {
private ClientService.Interface stub;
+ private ExecutorService threadPool;
+
private AsyncConnection conn;
@Rule
@@ -98,34 +108,9 @@ public class TestAsyncTableRpcPriority {
@Before
public void setUp() throws IOException {
+ this.threadPool = Executors.newSingleThreadExecutor();
stub = mock(ClientService.Interface.class);
- AtomicInteger scanNextCalled = new AtomicInteger(0);
- doAnswer(new Answer<Void>() {
- @Override
- public Void answer(InvocationOnMock invocation) throws Throwable {
- ScanRequest req = invocation.getArgument(1);
- RpcCallback<ScanResponse> done = invocation.getArgument(2);
- if (!req.hasScannerId()) {
- done.run(ScanResponse.newBuilder().setScannerId(1).setTtl(800)
- .setMoreResultsInRegion(true).setMoreResults(true).build());
- } else {
- if (req.hasCloseScanner() && req.getCloseScanner()) {
- done.run(ScanResponse.getDefaultInstance());
- } else {
- Cell cell = CellBuilderFactory.create(CellBuilderType.SHALLOW_COPY).setType(Type.Put)
- .setRow(Bytes.toBytes(scanNextCalled.incrementAndGet()))
- .setFamily(Bytes.toBytes("cf")).setQualifier(Bytes.toBytes("cq"))
- .setValue(Bytes.toBytes("v")).build();
- Result result = Result.create(Arrays.asList(cell));
- done.run(
- ScanResponse.newBuilder().setScannerId(1).setTtl(800).setMoreResultsInRegion(true)
- .setMoreResults(true).addResults(ProtobufUtil.toResult(result)).build());
- }
- }
- return null;
- }
- }).when(stub).scan(any(HBaseRpcController.class), any(ScanRequest.class), any());
doAnswer(new Answer<Void>() {
@Override
@@ -218,6 +203,16 @@ public class TestAsyncTableRpcPriority {
});
}
+ private ScanRequest assertScannerCloseRequest() {
+ return argThat(new ArgumentMatcher<ScanRequest>() {
+
+ @Override
+ public boolean matches(ScanRequest request) {
+ return request.hasCloseScanner() && request.getCloseScanner();
+ }
+ });
+ }
+
@Test
public void testGet() {
conn.getTable(TableName.valueOf(name.getMethodName()))
@@ -478,53 +473,112 @@ public class TestAsyncTableRpcPriority {
any(ClientProtos.MultiRequest.class), any());
}
+ private CompletableFuture<Void> mockScanReturnRenewFuture(int scanPriority) {
+ int scannerId = 1;
+ CompletableFuture<Void> future = new CompletableFuture<>();
+ AtomicInteger scanNextCalled = new AtomicInteger(0);
+ doAnswer(new Answer<Void>() {
+
+ @SuppressWarnings("FutureReturnValueIgnored")
+ @Override
+ public Void answer(InvocationOnMock invocation) throws Throwable {
+ threadPool.submit(() -> {
+ ScanRequest req = invocation.getArgument(1);
+ RpcCallback<ScanResponse> done = invocation.getArgument(2);
+ if (!req.hasScannerId()) {
+ done.run(
+ ScanResponse.newBuilder().setScannerId(scannerId).setTtl(800)
+ .setMoreResultsInRegion(true).setMoreResults(true).build());
+ } else {
+ if (req.hasRenew() && req.getRenew()) {
+ future.complete(null);
+ }
+
+ assertFalse("close scanner should not come in with scan priority " + scanPriority,
+ req.hasCloseScanner() && req.getCloseScanner());
+
+ Cell cell = CellBuilderFactory.create(CellBuilderType.SHALLOW_COPY)
+ .setType(Type.Put).setRow(Bytes.toBytes(scanNextCalled.incrementAndGet()))
+ .setFamily(Bytes.toBytes("cf")).setQualifier(Bytes.toBytes("cq"))
+ .setValue(Bytes.toBytes("v")).build();
+ Result result = Result.create(Arrays.asList(cell));
+ done.run(
+ ScanResponse.newBuilder().setScannerId(scannerId).setTtl(800)
+ .setMoreResultsInRegion(true).setMoreResults(true)
+ .addResults(ProtobufUtil.toResult(result)).build());
+ }
+ });
+ return null;
+ }
+ }).when(stub).scan(assertPriority(scanPriority), any(ScanRequest.class), any());
+
+ doAnswer(new Answer<Void>() {
+
+ @SuppressWarnings("FutureReturnValueIgnored")
+ @Override
+ public Void answer(InvocationOnMock invocation) throws Throwable {
+ threadPool.submit(() -> {
+ ScanRequest req = invocation.getArgument(1);
+ RpcCallback<ScanResponse> done = invocation.getArgument(2);
+ assertTrue("close request should have scannerId", req.hasScannerId());
+ assertEquals("close request's scannerId should match", scannerId,
+ req.getScannerId());
+ assertTrue("close request should have closerScanner set",
+ req.hasCloseScanner() && req.getCloseScanner());
+
+ done.run(ScanResponse.getDefaultInstance());
+ });
+ return null;
+ }
+ }).when(stub).scan(assertPriority(HIGH_QOS), assertScannerCloseRequest(), any());
+ return future;
+ }
+
@Test
- public void testScan() throws IOException, InterruptedException {
- try (ResultScanner scanner = conn.getTable(TableName.valueOf(name.getMethodName()))
- .getScanner(new Scan().setCaching(1).setMaxResultSize(1).setPriority(19))) {
- assertNotNull(scanner.next());
- Thread.sleep(1000);
- }
- Thread.sleep(1000);
- // open, next, several renew lease, and then close
- verify(stub, atLeast(4)).scan(assertPriority(19), any(ScanRequest.class), any());
+ public void testScan() throws Exception {
+ CompletableFuture<Void> renewFuture = mockScanReturnRenewFuture(19);
+ testForTable(TableName.valueOf(name.getMethodName()), renewFuture, Optional.of(19));
}
@Test
- public void testScanNormalTable() throws IOException, InterruptedException {
- try (ResultScanner scanner = conn.getTable(TableName.valueOf(name.getMethodName()))
- .getScanner(new Scan().setCaching(1).setMaxResultSize(1))) {
- assertNotNull(scanner.next());
- Thread.sleep(1000);
- }
- Thread.sleep(1000);
- // open, next, several renew lease, and then close
- verify(stub, atLeast(4)).scan(assertPriority(NORMAL_QOS), any(ScanRequest.class), any());
+ public void testScanNormalTable() throws Exception {
+ CompletableFuture<Void> renewFuture = mockScanReturnRenewFuture(NORMAL_QOS);
+ testForTable(TableName.valueOf(name.getMethodName()), renewFuture, Optional.of(NORMAL_QOS));
}
@Test
- public void testScanSystemTable() throws IOException, InterruptedException {
- try (ResultScanner scanner =
- conn.getTable(TableName.valueOf(SYSTEM_NAMESPACE_NAME_STR, name.getMethodName()))
- .getScanner(new Scan().setCaching(1).setMaxResultSize(1))) {
- assertNotNull(scanner.next());
- Thread.sleep(1000);
- }
- Thread.sleep(1000);
- // open, next, several renew lease, and then close
- verify(stub, atLeast(4)).scan(assertPriority(SYSTEMTABLE_QOS), any(ScanRequest.class), any());
+ public void testScanSystemTable() throws Exception {
+ CompletableFuture<Void> renewFuture = mockScanReturnRenewFuture(SYSTEMTABLE_QOS);
+ testForTable(TableName.valueOf(SYSTEM_NAMESPACE_NAME_STR, name.getMethodName()),
+ renewFuture, Optional.empty());
}
@Test
- public void testScanMetaTable() throws IOException, InterruptedException {
- try (ResultScanner scanner = conn.getTable(TableName.META_TABLE_NAME)
- .getScanner(new Scan().setCaching(1).setMaxResultSize(1))) {
+ public void testScanMetaTable() throws Exception {
+ CompletableFuture<Void> renewFuture = mockScanReturnRenewFuture(SYSTEMTABLE_QOS);
+ testForTable(TableName.META_TABLE_NAME, renewFuture, Optional.empty());
+ }
+
+ private void testForTable(TableName tableName, CompletableFuture<Void> renewFuture,
+ Optional<Integer> priority) throws Exception {
+ Scan scan = new Scan().setCaching(1).setMaxResultSize(1);
+ priority.ifPresent(scan::setPriority);
+
+ try (ResultScanner scanner = conn.getTable(tableName).getScanner(scan)) {
assertNotNull(scanner.next());
- Thread.sleep(1000);
+ // wait for at least one renew to come in before closing
+ renewFuture.join();
}
- Thread.sleep(1000);
- // open, next, several renew lease, and then close
- verify(stub, atLeast(4)).scan(assertPriority(SYSTEMTABLE_QOS), any(ScanRequest.class), any());
+
+ // ensures the close thread has time to finish before asserting
+ threadPool.shutdown();
+ threadPool.awaitTermination(5, TimeUnit.SECONDS);
+
+ // just verify that the calls happened. verification of priority occurred in the mocking
+ // open, next, then one or more lease renewals, then close
+ verify(stub, atLeast(4)).scan(any(), any(ScanRequest.class), any());
+ // additionally, explicitly check for a close request
+ verify(stub, times(1)).scan(any(), assertScannerCloseRequest(), any());
}
@Test
diff --git a/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestTableRpcPriority.java b/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestTableRpcPriority.java
new file mode 100644
index 00000000000..1ca09f0940e
--- /dev/null
+++ b/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestTableRpcPriority.java
@@ -0,0 +1,222 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.client;
+
+import static org.apache.hadoop.hbase.HConstants.HIGH_QOS;
+import static org.apache.hadoop.hbase.HConstants.NORMAL_QOS;
+import static org.apache.hadoop.hbase.HConstants.SYSTEMTABLE_QOS;
+import static org.apache.hadoop.hbase.NamespaceDescriptor.SYSTEM_NAMESPACE_NAME_STR;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.argThat;
+import static org.mockito.Mockito.atLeast;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellBuilderFactory;
+import org.apache.hadoop.hbase.CellBuilderType;
+import org.apache.hadoop.hbase.HBaseClassTestRule;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HRegionLocation;
+import org.apache.hadoop.hbase.RegionLocations;
+import org.apache.hadoop.hbase.ServerName;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.ipc.HBaseRpcController;
+import org.apache.hadoop.hbase.security.UserProvider;
+import org.apache.hadoop.hbase.testclassification.ClientTests;
+import org.apache.hadoop.hbase.testclassification.MediumTests;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.rules.TestName;
+import org.mockito.ArgumentMatcher;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+import org.apache.hbase.thirdparty.com.google.protobuf.ServiceException;
+import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos;
+
+/**
+ * Test that correct rpc priority is sent to server from blocking Table calls. Currently
+ * only implements checks for scans, but more could be added here.
+ */
+@Category({ ClientTests.class, MediumTests.class })
+public class TestTableRpcPriority {
+ @ClassRule
+ public static final HBaseClassTestRule CLASS_RULE =
+ HBaseClassTestRule.forClass(TestTableRpcPriority.class);
+
+ @Rule
+ public TestName name = new TestName();
+
+ private ClientProtos.ClientService.BlockingInterface stub;
+ private Connection conn;
+
+ @Before
+ public void setUp() throws IOException, ServiceException {
+ stub = mock(ClientProtos.ClientService.BlockingInterface.class);
+
+ Configuration conf = HBaseConfiguration.create();
+ ExecutorService executorService = Executors.newCachedThreadPool();
+ conn = new ConnectionImplementation(conf, executorService,
+ UserProvider.instantiate(conf).getCurrent()) {
+
+ @Override
+ public ClientProtos.ClientService.BlockingInterface getClient(ServerName serverName)
+ throws IOException {
+ return stub;
+ }
+
+ @Override
+ public RegionLocations relocateRegion(final TableName tableName, final byte[] row,
+ int replicaId) throws IOException {
+ return locateRegion(tableName, row, true, false, replicaId);
+ }
+
+ @Override
+ public RegionLocations locateRegion(TableName tableName, byte[] row, boolean useCache,
+ boolean retry, int replicaId) throws IOException {
+ RegionInfo info = RegionInfoBuilder.newBuilder(tableName).build();
+ ServerName serverName = ServerName.valueOf("rs", 16010, 12345);
+ HRegionLocation loc = new HRegionLocation(info, serverName);
+ return new RegionLocations(loc);
+ }
+ };
+ }
+
+ @Test
+ public void testScan() throws Exception {
+ mockScan(19);
+ testForTable(TableName.valueOf(name.getMethodName()), Optional.of(19));
+ }
+
+ @Test
+ public void testScanNormalTable() throws Exception {
+ mockScan(NORMAL_QOS);
+ testForTable(TableName.valueOf(name.getMethodName()), Optional.of(NORMAL_QOS));
+ }
+
+ @Test
+ public void testScanSystemTable() throws Exception {
+ mockScan(SYSTEMTABLE_QOS);
+ testForTable(TableName.valueOf(SYSTEM_NAMESPACE_NAME_STR, name.getMethodName()),
+ Optional.empty());
+ }
+
+ @Test
+ public void testScanMetaTable() throws Exception {
+ mockScan(SYSTEMTABLE_QOS);
+ testForTable(TableName.META_TABLE_NAME, Optional.empty());
+ }
+
+ private void testForTable(TableName tableName, Optional<Integer> priority) throws Exception {
+ Scan scan = new Scan().setCaching(1);
+ priority.ifPresent(scan::setPriority);
+
+ try (ResultScanner scanner = conn.getTable(tableName).getScanner(scan)) {
+ assertNotNull(scanner.next());
+ assertNotNull(scanner.next());
+ }
+
+ // just verify that the calls happened. verification of priority occurred in the mocking
+ // open, next, then several renew lease
+ verify(stub, atLeast(3)).scan(any(), any(ClientProtos.ScanRequest.class));
+ verify(stub, times(1)).scan(any(), assertScannerCloseRequest());
+ }
+
+ private void mockScan(int scanPriority) throws ServiceException {
+ int scannerId = 1;
+ AtomicInteger scanNextCalled = new AtomicInteger(0);
+ doAnswer(new Answer<ClientProtos.ScanResponse>() {
+
+ @Override
+ public ClientProtos.ScanResponse answer(InvocationOnMock invocation)
+ throws Throwable {
+ ClientProtos.ScanRequest req = invocation.getArgument(1);
+ assertFalse("close scanner should not come in with scan priority " + scanPriority,
+ req.hasCloseScanner() && req.getCloseScanner());
+ ClientProtos.ScanResponse.Builder builder = ClientProtos.ScanResponse.newBuilder();
+
+ if (!req.hasScannerId()) {
+ builder.setScannerId(scannerId);
+ } else {
+ builder.setScannerId(req.getScannerId());
+ }
+
+ Cell cell = CellBuilderFactory.create(CellBuilderType.SHALLOW_COPY).setType(Cell.Type.Put)
+ .setRow(Bytes.toBytes(scanNextCalled.incrementAndGet())).setFamily(Bytes.toBytes("cf"))
+ .setQualifier(Bytes.toBytes("cq")).setValue(Bytes.toBytes("v")).build();
+ Result result = Result.create(Arrays.asList(cell));
+ return builder.setTtl(800).setMoreResultsInRegion(true).setMoreResults(true)
+ .addResults(ProtobufUtil.toResult(result)).build();
+ }
+ }).when(stub).scan(assertPriority(scanPriority), any(ClientProtos.ScanRequest.class));
+
+ doAnswer(new Answer<ClientProtos.ScanResponse>() {
+
+ @Override
+ public ClientProtos.ScanResponse answer(InvocationOnMock invocation)
+ throws Throwable {
+ ClientProtos.ScanRequest req = invocation.getArgument(1);
+ assertTrue("close request should have scannerId", req.hasScannerId());
+ assertEquals("close request's scannerId should match", scannerId, req.getScannerId());
+ assertTrue("close request should have closerScanner set",
+ req.hasCloseScanner() && req.getCloseScanner());
+
+ return ClientProtos.ScanResponse.getDefaultInstance();
+ }
+ }).when(stub).scan(assertPriority(HIGH_QOS), assertScannerCloseRequest());
+ }
+
+ private HBaseRpcController assertPriority(int priority) {
+ return argThat(new ArgumentMatcher<HBaseRpcController>() {
+
+ @Override
+ public boolean matches(HBaseRpcController controller) {
+ return controller.getPriority() == priority;
+ }
+ });
+ }
+
+ private ClientProtos.ScanRequest assertScannerCloseRequest() {
+ return argThat(new ArgumentMatcher<ClientProtos.ScanRequest>() {
+
+ @Override
+ public boolean matches(ClientProtos.ScanRequest request) {
+ return request.hasCloseScanner() && request.getCloseScanner();
+ }
+ });
+ }
+}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/AbstractTestAsyncTableScan.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/AbstractTestAsyncTableScan.java
index 5831bfcc60b..f971a5384d7 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/AbstractTestAsyncTableScan.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/AbstractTestAsyncTableScan.java
@@ -130,7 +130,7 @@ public abstract class AbstractTestAsyncTableScan {
protected abstract Scan createScan();
- protected abstract List<Result> doScan(Scan scan) throws Exception;
+ protected abstract List<Result> doScan(Scan scan, int closeAfter) throws Exception;
protected final List<Result> convertFromBatchResult(List<Result> results) {
assertTrue(results.size() % 2 == 0);
@@ -146,7 +146,7 @@ public abstract class AbstractTestAsyncTableScan {
@Test
public void testScanAll() throws Exception {
- List<Result> results = doScan(createScan());
+ List<Result> results = doScan(createScan(), -1);
// make sure all scanners are closed at RS side
TEST_UTIL.getHBaseCluster().getRegionServerThreads().stream().map(t -> t.getRegionServer())
.forEach(
@@ -170,7 +170,7 @@ public abstract class AbstractTestAsyncTableScan {
@Test
public void testReversedScanAll() throws Exception {
- List<Result> results = doScan(createScan().setReversed(true));
+ List<Result> results = doScan(createScan().setReversed(true), -1);
assertEquals(COUNT, results.size());
IntStream.range(0, COUNT).forEach(i -> assertResultEquals(results.get(i), COUNT - i - 1));
}
@@ -179,7 +179,7 @@ public abstract class AbstractTestAsyncTableScan {
public void testScanNoStopKey() throws Exception {
int start = 345;
List<Result> results =
- doScan(createScan().withStartRow(Bytes.toBytes(String.format("%03d", start))));
+ doScan(createScan().withStartRow(Bytes.toBytes(String.format("%03d", start))), -1);
assertEquals(COUNT - start, results.size());
IntStream.range(0, COUNT - start).forEach(i -> assertResultEquals(results.get(i), start + i));
}
@@ -188,7 +188,7 @@ public abstract class AbstractTestAsyncTableScan {
public void testReverseScanNoStopKey() throws Exception {
int start = 765;
List<Result> results = doScan(
- createScan().withStartRow(Bytes.toBytes(String.format("%03d", start))).setReversed(true));
+ createScan().withStartRow(Bytes.toBytes(String.format("%03d", start))).setReversed(true), -1);
assertEquals(start + 1, results.size());
IntStream.range(0, start + 1).forEach(i -> assertResultEquals(results.get(i), start - i));
}
@@ -196,7 +196,7 @@ public abstract class AbstractTestAsyncTableScan {
@Test
public void testScanWrongColumnFamily() throws Exception {
try {
- doScan(createScan().addFamily(Bytes.toBytes("WrongColumnFamily")));
+ doScan(createScan().addFamily(Bytes.toBytes("WrongColumnFamily")), -1);
} catch (Exception e) {
assertTrue(e instanceof NoSuchColumnFamilyException ||
e.getCause() instanceof NoSuchColumnFamilyException);
@@ -204,20 +204,28 @@ public abstract class AbstractTestAsyncTableScan {
}
private void testScan(int start, boolean startInclusive, int stop, boolean stopInclusive,
- int limit) throws Exception {
+ int limit) throws Exception {
+ testScan(start, startInclusive, stop, stopInclusive, limit, -1);
+ }
+
+ private void testScan(int start, boolean startInclusive, int stop, boolean stopInclusive,
+ int limit, int closeAfter) throws Exception {
Scan scan =
createScan().withStartRow(Bytes.toBytes(String.format("%03d", start)), startInclusive)
.withStopRow(Bytes.toBytes(String.format("%03d", stop)), stopInclusive);
if (limit > 0) {
scan.setLimit(limit);
}
- List<Result> results = doScan(scan);
+ List<Result> results = doScan(scan, closeAfter);
int actualStart = startInclusive ? start : start + 1;
int actualStop = stopInclusive ? stop + 1 : stop;
int count = actualStop - actualStart;
if (limit > 0) {
count = Math.min(count, limit);
}
+ if (closeAfter > 0) {
+ count = Math.min(count, closeAfter);
+ }
assertEquals(count, results.size());
IntStream.range(0, count).forEach(i -> assertResultEquals(results.get(i), actualStart + i));
}
@@ -230,7 +238,7 @@ public abstract class AbstractTestAsyncTableScan {
if (limit > 0) {
scan.setLimit(limit);
}
- List<Result> results = doScan(scan);
+ List<Result> results = doScan(scan, -1);
int actualStart = startInclusive ? start : start - 1;
int actualStop = stopInclusive ? stop - 1 : stop;
int count = actualStart - actualStop;
@@ -310,4 +318,13 @@ public abstract class AbstractTestAsyncTableScan {
testReversedScan(765, false, 543, true, 200);
testReversedScan(876, false, 654, false, 200);
}
+
+ @Test
+ public void testScanEndingEarly() throws Exception {
+ testScan(1, true, 998, false, 0, 900); // from first region to last region
+ testScan(123, true, 234, true, 0, 100);
+ testScan(234, true, 456, false, 0, 100);
+ testScan(345, false, 567, true, 0, 100);
+ testScan(456, false, 678, false, 0, 100);
+ }
}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableScan.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableScan.java
index 42d2c38376d..c1797f3833c 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableScan.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableScan.java
@@ -17,6 +17,7 @@
*/
package org.apache.hadoop.hbase.client;
+import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ForkJoinPool;
import java.util.function.Supplier;
@@ -29,6 +30,7 @@ import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.junit.runners.Parameterized.Parameter;
import org.junit.runners.Parameterized.Parameters;
+import org.apache.hbase.thirdparty.com.google.common.base.Throwables;
@RunWith(Parameterized.class)
@Category({ LargeTests.class, ClientTests.class })
@@ -55,16 +57,74 @@ public class TestAsyncTableScan extends AbstractTestAsyncTableScan {
}
@Override
- protected List<Result> doScan(Scan scan) throws Exception {
+ protected List<Result> doScan(Scan scan, int closeAfter) throws Exception {
AsyncTable<ScanResultConsumer> table =
ASYNC_CONN.getTable(TABLE_NAME, ForkJoinPool.commonPool());
- SimpleScanResultConsumer consumer = new SimpleScanResultConsumer();
- table.scan(scan, consumer);
- List<Result> results = consumer.getAll();
+ List<Result> results;
+ if (closeAfter > 0) {
+ // these tests batch settings with the sample data result in each result being
+ // split in two. so we must allow twice the expected results in order to reach
+ // our true limit. see convertFromBatchResult for details.
+ if (scan.getBatch() > 0) {
+ closeAfter = closeAfter * 2;
+ }
+ LimitedScanResultConsumer consumer = new LimitedScanResultConsumer(closeAfter);
+ table.scan(scan, consumer);
+ results = consumer.getAll();
+ } else {
+ SimpleScanResultConsumer consumer = new SimpleScanResultConsumer();
+ table.scan(scan, consumer);
+ results = consumer.getAll();
+ }
if (scan.getBatch() > 0) {
results = convertFromBatchResult(results);
}
return results;
}
+ private static class LimitedScanResultConsumer implements ScanResultConsumer {
+
+ private final int limit;
+
+ public LimitedScanResultConsumer(int limit) {
+ this.limit = limit;
+ }
+
+ private final List<Result> results = new ArrayList<>();
+
+ private Throwable error;
+
+ private boolean finished = false;
+
+ @Override
+ public synchronized boolean onNext(Result result) {
+ results.add(result);
+ return results.size() < limit;
+ }
+
+ @Override
+ public synchronized void onError(Throwable error) {
+ this.error = error;
+ finished = true;
+ notifyAll();
+ }
+
+ @Override
+ public synchronized void onComplete() {
+ finished = true;
+ notifyAll();
+ }
+
+ public synchronized List<Result> getAll() throws Exception {
+ while (!finished) {
+ wait();
+ }
+ if (error != null) {
+ Throwables.propagateIfPossible(error, Exception.class);
+ throw new Exception(error);
+ }
+ return results;
+ }
+ }
+
}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableScanAll.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableScanAll.java
index d9a53952ab8..96c2d40138c 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableScanAll.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableScanAll.java
@@ -60,11 +60,16 @@ public class TestAsyncTableScanAll extends AbstractTestAsyncTableScan {
}
@Override
- protected List<Result> doScan(Scan scan) throws Exception {
+ protected List<Result> doScan(Scan scan, int closeAfter) throws Exception {
List<Result> results = getTable.get().scanAll(scan).get();
if (scan.getBatch() > 0) {
results = convertFromBatchResult(results);
}
+ // we can't really close the scan early for scanAll, but to keep the assertions
+ // simple in AbstractTestAsyncTableScan we'll just sublist here instead.
+ if (closeAfter > 0 && closeAfter < results.size()) {
+ results = results.subList(0, closeAfter);
+ }
return results;
}
}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableScanner.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableScanner.java
index f832cfd759a..2e990f763da 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableScanner.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableScanner.java
@@ -62,12 +62,21 @@ public class TestAsyncTableScanner extends AbstractTestAsyncTableScan {
}
@Override
- protected List<Result> doScan(Scan scan) throws Exception {
+ protected List<Result> doScan(Scan scan, int closeAfter) throws Exception {
AsyncTable<?> table = ASYNC_CONN.getTable(TABLE_NAME, ForkJoinPool.commonPool());
List<Result> results = new ArrayList<>();
+ // these tests batch settings with the sample data result in each result being
+ // split in two. so we must allow twice the expected results in order to reach
+ // our true limit. see convertFromBatchResult for details.
+ if (closeAfter > 0 && scan.getBatch() > 0) {
+ closeAfter = closeAfter * 2;
+ }
try (ResultScanner scanner = table.getScanner(scan)) {
for (Result result; (result = scanner.next()) != null;) {
results.add(result);
+ if (closeAfter > 0 && results.size() >= closeAfter) {
+ break;
+ }
}
}
if (scan.getBatch() > 0) {
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestRawAsyncTableScan.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestRawAsyncTableScan.java
index 78a54edd24b..26c201e1986 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestRawAsyncTableScan.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestRawAsyncTableScan.java
@@ -55,12 +55,21 @@ public class TestRawAsyncTableScan extends AbstractTestAsyncTableScan {
}
@Override
- protected List<Result> doScan(Scan scan) throws Exception {
+ protected List<Result> doScan(Scan scan, int closeAfter) throws Exception {
BufferingScanResultConsumer scanConsumer = new BufferingScanResultConsumer();
ASYNC_CONN.getTable(TABLE_NAME).scan(scan, scanConsumer);
List<Result> results = new ArrayList<>();
+ // these tests batch settings with the sample data result in each result being
+ // split in two. so we must allow twice the expected results in order to reach
+ // our true limit. see convertFromBatchResult for details.
+ if (closeAfter > 0 && scan.getBatch() > 0) {
+ closeAfter = closeAfter * 2;
+ }
for (Result result; (result = scanConsumer.take()) != null;) {
results.add(result);
+ if (closeAfter > 0 && results.size() >= closeAfter) {
+ break;
+ }
}
if (scan.getBatch() > 0) {
results = convertFromBatchResult(results);