Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

HBASE-26784 Use HIGH_QOS for ResultScanner.close requests #4146

Merged
merged 7 commits into from
Mar 8, 2022
Merged

HBASE-26784 Use HIGH_QOS for ResultScanner.close requests #4146

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
1635eb6
HBASE-26784 Use HIGH_QOS for ResultScanner.close requests
bbeaudreault Mar 2, 2022
a5ab3ce
Add test case to AbstractTestAsyncTableScan suite for testing closing…
bbeaudreault Mar 3, 2022
712abb6
fix tests and fix priority of openScanner call
bbeaudreault Mar 3, 2022
7469dbd
clean up tests
bbeaudreault Mar 4, 2022
5e1dc27
checkstyle
bbeaudreault Mar 4, 2022
f2f2008
fix imports
bbeaudreault Mar 5, 2022
9868c1f
single import
bbeaudreault Mar 5, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncClientScanner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -188,6 +188,7 @@ private void startScan(OpenScannerResponse resp) {
private CompletableFuture<OpenScannerResponse> openScanner(int replicaId) {
return conn.callerFactory.<OpenScannerResponse> single().table(tableName)
.row(scan.getStartRow()).replicaId(replicaId).locateType(getLocateType(scan))
.priority(scan.getPriority())
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This absence was a bug, no?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, this was a bug. I caught it in my investigation into these tests. Should I be filing a separate JIRA for that, or is it ok to include here?

.rpcTimeout(rpcTimeoutNs, TimeUnit.NANOSECONDS)
.operationTimeout(scanTimeoutNs, TimeUnit.NANOSECONDS).pause(pauseNs, TimeUnit.NANOSECONDS)
.pauseForCQTBE(pauseForCQTBENs, TimeUnit.NANOSECONDS).maxAttempts(maxAttempts)
Expand Down
3 changes: 2 additions & 1 deletion .../src/main/java/org/apache/hadoop/hbase/client/AsyncScanSingleRegionRpcRetryingCaller.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
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;
Expand Down Expand Up @@ -347,7 +348,7 @@ private long remainingTimeNs() {

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()) {
Expand Down
140 changes: 102 additions & 38 deletions hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableRpcPriority.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -34,6 +38,9 @@
import java.io.IOException;
import java.util.Arrays;
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;
Expand Down Expand Up @@ -91,41 +98,18 @@ public class TestAsyncTableRpcPriority {

private ClientService.Interface stub;

private ExecutorService threadPool;

private AsyncConnection conn;

@Rule
public TestName name = new TestName();

@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
Expand Down Expand Up @@ -218,6 +202,16 @@ public boolean matches(HBaseRpcController controller) {
});
}

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()))
Expand Down Expand Up @@ -478,53 +472,123 @@ public void testCheckAndMutateMetaTable() throws IOException {
any(ClientProtos.MultiRequest.class), any());
}

private void mockScan(int scanPriority) {
int scannerId = 1;
AtomicInteger scanNextCalled = new AtomicInteger(0);
doAnswer(new Answer<Void>() {

@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 {
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>() {

@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());
}

@Test
public void testScan() throws IOException, InterruptedException {
mockScan(19);
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());
// ensures the close thread has time to finish before asserting
bbeaudreault marked this conversation as resolved.
Show resolved Hide resolved
threadPool.shutdown();
threadPool.awaitTermination(5, TimeUnit.SECONDS);

// 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(ScanRequest.class), any());
verify(stub, times(1)).scan(any(), assertScannerCloseRequest(), any());
}

@Test
public void testScanNormalTable() throws IOException, InterruptedException {
mockScan(NORMAL_QOS);
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());
// 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 several renew lease
verify(stub, atLeast(3)).scan(any(), any(ScanRequest.class), any());
verify(stub, times(1)).scan(any(), assertScannerCloseRequest(), any());
}

@Test
public void testScanSystemTable() throws IOException, InterruptedException {
mockScan(SYSTEMTABLE_QOS);
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());
// 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 several renew lease
verify(stub, atLeast(3)).scan(any(), any(ScanRequest.class), any());
verify(stub, times(1)).scan(any(), assertScannerCloseRequest(), any());
}

@Test
public void testScanMetaTable() throws IOException, InterruptedException {
mockScan(SYSTEMTABLE_QOS);
try (ResultScanner scanner = conn.getTable(TableName.META_TABLE_NAME)
.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());
// 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 several renew lease
verify(stub, atLeast(3)).scan(any(), any(ScanRequest.class), any());
verify(stub, times(1)).scan(any(), assertScannerCloseRequest(), any());
}

@Test
Expand Down
38 changes: 29 additions & 9 deletions hbase-server/src/test/java/org/apache/hadoop/hbase/client/AbstractTestAsyncTableScan.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -129,7 +129,7 @@ protected static List<Object[]> getTableAndScanCreatorParams() {

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);
Expand All @@ -145,7 +145,7 @@ protected final List<Result> convertFromBatchResult(List<Result> results) {

@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(
Expand All @@ -169,7 +169,7 @@ private void assertResultEquals(Result result, int i) {

@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));
}
Expand All @@ -178,7 +178,7 @@ public void testReversedScanAll() throws Exception {
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));
}
Expand All @@ -187,36 +187,44 @@ public void testScanNoStopKey() throws Exception {
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));
}

@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);
}
}

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));
}
Expand All @@ -229,12 +237,15 @@ private void testReversedScan(int start, boolean startInclusive, int stop, boole
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;
if (limit > 0) {
count = Math.min(count, limit);
}
if (scan.getBatch() > 0) {
bbeaudreault marked this conversation as resolved.
Show resolved Hide resolved

}
assertEquals(count, results.size());
IntStream.range(0, count).forEach(i -> assertResultEquals(results.get(i), actualStart - i));
Expand Down Expand Up @@ -309,4 +320,13 @@ public void testReversedScanWithLimitGreaterThanActualCount() throws Exception {
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);
}
}
Loading