You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by ap...@apache.org on 2022/03/08 20:00:41 UTC
[hbase] 02/02: HBASE-26784 Addendum: Close scanner request should properly inherit original timeout and priority (#4183)
This is an automated email from the ASF dual-hosted git repository.
apurtell pushed a commit to branch branch-2.5
in repository https://gitbox.apache.org/repos/asf/hbase.git
commit b1a7c6210fac261ffbac93ea44f814f1d64a9038
Author: Bryan Beaudreault <bb...@hubspot.com>
AuthorDate: Tue Mar 8 14:40:39 2022 -0500
HBASE-26784 Addendum: Close scanner request should properly inherit original timeout and priority (#4183)
Signed-off-by: Andrew Purtell <ap...@apache.org>
---
.../hadoop/hbase/client/ScannerCallable.java | 18 ++++++----
.../hadoop/hbase/ipc/HBaseRpcControllerImpl.java | 7 ++++
.../hadoop/hbase/client/TestTableRpcPriority.java | 39 ++++++++++++++++++----
3 files changed, 51 insertions(+), 13 deletions(-)
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 e6c7583..8a80b6a 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
@@ -312,13 +312,19 @@ public class ScannerCallable extends ClientServiceCallable<Result[]> {
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());
+
+ // Set fields from the original controller onto the close-specific controller
+ // We set the timeout and the priority -- we will overwrite the priority to HIGH
+ // below, but the controller will take whichever is highest.
+ if (getRpcController() instanceof HBaseRpcController) {
+ HBaseRpcController original = (HBaseRpcController) getRpcController();
+ controller.setPriority(original.getPriority());
+ if (original.hasCallTimeout()) {
+ controller.setCallTimeout(original.getCallTimeout());
+ }
}
+ controller.setPriority(HConstants.HIGH_QOS);
+
try {
getStub().scan(controller, request);
} catch (Exception e) {
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/HBaseRpcControllerImpl.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/HBaseRpcControllerImpl.java
index 7dde67c..3058247 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/HBaseRpcControllerImpl.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/HBaseRpcControllerImpl.java
@@ -262,4 +262,11 @@ public class HBaseRpcControllerImpl implements HBaseRpcController {
action.run(false);
}
}
+
+ @Override public String toString() {
+ return "HBaseRpcControllerImpl{" + "callTimeout=" + callTimeout + ", done=" + done
+ + ", cancelled=" + cancelled + ", cancellationCbs=" + cancellationCbs + ", exception="
+ + exception + ", regionInfo=" + regionInfo + ", priority=" + priority + ", cellScanner="
+ + cellScanner + '}';
+ }
}
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
index 7aef284..882381b 100644
--- 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
@@ -32,14 +32,12 @@ 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;
@@ -74,6 +72,7 @@ import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos;
*/
@Category({ ClientTests.class, MediumTests.class })
public class TestTableRpcPriority {
+
@ClassRule
public static final HBaseClassTestRule CLASS_RULE =
HBaseClassTestRule.forClass(TestTableRpcPriority.class);
@@ -89,6 +88,7 @@ public class TestTableRpcPriority {
stub = mock(ClientProtos.ClientService.BlockingInterface.class);
Configuration conf = HBaseConfiguration.create();
+
ExecutorService executorService = Executors.newCachedThreadPool();
conn = new ConnectionImplementation(conf, executorService,
UserProvider.instantiate(conf).getCurrent(), new DoNothingConnectionRegistry(conf)) {
@@ -122,6 +122,16 @@ public class TestTableRpcPriority {
testForTable(TableName.valueOf(name.getMethodName()), Optional.of(19));
}
+ /**
+ * This test verifies that our closeScanner request honors the original
+ * priority of the scan if it's greater than our expected HIGH_QOS for close calls.
+ */
+ @Test
+ public void testScanSuperHighPriority() throws Exception {
+ mockScan(1000);
+ testForTable(TableName.valueOf(name.getMethodName()), Optional.of(1000));
+ }
+
@Test
public void testScanNormalTable() throws Exception {
mockScan(NORMAL_QOS);
@@ -153,11 +163,22 @@ public class TestTableRpcPriority {
// 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());
+ verify(stub, times(1)).scan(
+ assertControllerArgs(Math.max(priority.orElse(0), HIGH_QOS)), assertScannerCloseRequest());
}
private void mockScan(int scanPriority) throws ServiceException {
int scannerId = 1;
+
+ doAnswer(new Answer<ClientProtos.ScanResponse>() {
+ @Override public ClientProtos.ScanResponse answer(InvocationOnMock invocation)
+ throws Throwable {
+ throw new IllegalArgumentException(
+ "Call not covered by explicit mock for arguments controller="
+ + invocation.getArgument(0) + ", request=" + invocation.getArgument(1));
+ }
+ }).when(stub).scan(any(), any());
+
AtomicInteger scanNextCalled = new AtomicInteger(0);
doAnswer(new Answer<ClientProtos.ScanResponse>() {
@@ -182,7 +203,7 @@ public class TestTableRpcPriority {
return builder.setTtl(800).setMoreResultsInRegion(true).setMoreResults(true)
.addResults(ProtobufUtil.toResult(result)).build();
}
- }).when(stub).scan(assertPriority(scanPriority), any(ClientProtos.ScanRequest.class));
+ }).when(stub).scan(assertControllerArgs(scanPriority), any());
doAnswer(new Answer<ClientProtos.ScanResponse>() {
@@ -197,15 +218,19 @@ public class TestTableRpcPriority {
return ClientProtos.ScanResponse.getDefaultInstance();
}
- }).when(stub).scan(assertPriority(HIGH_QOS), assertScannerCloseRequest());
+ }).when(stub).scan(assertControllerArgs(Math.max(scanPriority, HIGH_QOS)),
+ assertScannerCloseRequest());
}
- private HBaseRpcController assertPriority(int priority) {
+ private HBaseRpcController assertControllerArgs(int priority) {
return argThat(new ArgumentMatcher<HBaseRpcController>() {
@Override
public boolean matches(HBaseRpcController controller) {
- return controller.getPriority() == priority;
+ // check specified priority, but also check that it has a timeout
+ // this ensures that our conversion from the base controller to the close-specific
+ // controller honored the original arguments.
+ return controller.getPriority() == priority && controller.hasCallTimeout();
}
});
}