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.jupiter.api.Assertions.assertEquals;
021import static org.junit.jupiter.api.Assertions.assertTrue;
022import static org.mockito.ArgumentMatchers.any;
023import static org.mockito.ArgumentMatchers.argThat;
024import static org.mockito.Mockito.doAnswer;
025import static org.mockito.Mockito.mock;
026import static org.mockito.Mockito.times;
027import static org.mockito.Mockito.verify;
028
029import java.io.IOException;
030import java.util.concurrent.TimeUnit;
031import java.util.stream.Stream;
032import org.apache.hadoop.conf.Configuration;
033import org.apache.hadoop.hbase.HBaseConfiguration;
034import org.apache.hadoop.hbase.HBaseParameterizedTestTemplate;
035import org.apache.hadoop.hbase.HRegionLocation;
036import org.apache.hadoop.hbase.ServerName;
037import org.apache.hadoop.hbase.TableName;
038import org.apache.hadoop.hbase.client.metrics.ScanMetrics;
039import org.apache.hadoop.hbase.exceptions.OutOfOrderScannerNextException;
040import org.apache.hadoop.hbase.exceptions.ScannerResetException;
041import org.apache.hadoop.hbase.ipc.HBaseRpcController;
042import org.apache.hadoop.hbase.ipc.HBaseRpcControllerImpl;
043import org.apache.hadoop.hbase.security.User;
044import org.apache.hadoop.hbase.testclassification.ClientTests;
045import org.apache.hadoop.hbase.testclassification.MediumTests;
046import org.junit.jupiter.api.Tag;
047import org.junit.jupiter.api.TestTemplate;
048import org.junit.jupiter.params.provider.Arguments;
049
050import org.apache.hbase.thirdparty.com.google.protobuf.RpcCallback;
051
052import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.ClientService;
053import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.ScanRequest;
054import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.ScanResponse;
055
056@Tag(ClientTests.TAG)
057@Tag(MediumTests.TAG)
058@HBaseParameterizedTestTemplate(name = "[{index}] remote={0}")
059public class TestAsyncScanSingleRegionRpcRetryingCaller {
060
061  private final boolean remote;
062
063  public TestAsyncScanSingleRegionRpcRetryingCaller(boolean remote) {
064    this.remote = remote;
065  }
066
067  public static Stream<Arguments> parameters() {
068    return Stream.of(Arguments.of(false), Arguments.of(true));
069  }
070
071  private void assertRetryMetrics(IOException error, boolean closeScanner) throws Exception {
072    Configuration conf = HBaseConfiguration.create();
073    User user = User.getCurrent();
074    try (AsyncConnectionImpl conn = new AsyncConnectionImpl(conf,
075      new DoNothingConnectionRegistry(conf, user), "test", null, user)) {
076      ClientService.Interface stub = mock(ClientService.Interface.class);
077      doAnswer(invocation -> {
078        HBaseRpcController controller = invocation.getArgument(0);
079        ScanRequest request = invocation.getArgument(1);
080        RpcCallback<ScanResponse> done = invocation.getArgument(2);
081        if (request.getCloseScanner()) {
082          done.run(ScanResponse.getDefaultInstance());
083        } else {
084          controller.setFailed(error);
085          done.run(null);
086        }
087        return null;
088      }).when(stub).scan(any(), any(), any());
089
090      HRegionLocation location =
091        new HRegionLocation(RegionInfoBuilder.newBuilder(TableName.valueOf("test")).build(),
092          ServerName.valueOf("rs", 16020, 1));
093      ScanMetrics metrics = new ScanMetrics();
094      AsyncScanSingleRegionRpcRetryingCaller caller = conn.callerFactory.scanSingleRegion().id(1)
095        .setScan(new Scan().setScanMetricsEnabled(true)).metrics(metrics).remote(remote)
096        .resultCache(new CompleteScanResultCache()).consumer(mock(AdvancedScanResultConsumer.class))
097        .stub(stub).location(location).maxAttempts(3).build();
098
099      // A successful open is followed by a scan-next RPC that fails and requires reopening.
100      assertTrue(
101        caller
102          .start(new HBaseRpcControllerImpl(),
103            ScanResponse.newBuilder().setMoreResultsInRegion(true).build())
104          .get(5, TimeUnit.SECONDS));
105      verify(stub).scan(any(), argThat(request -> !request.getCloseScanner()), any());
106      verify(stub, times(closeScanner ? 1 : 0)).scan(any(), argThat(ScanRequest::getCloseScanner),
107        any());
108      assertEquals(1, metrics.countOfRPCRetries.get());
109      assertEquals(remote ? 1 : 0, metrics.countOfRemoteRPCRetries.get());
110    }
111  }
112
113  @TestTemplate
114  public void testRetryMetricsAfterScannerReset() throws Exception {
115    assertRetryMetrics(new ScannerResetException("Injected scanner reset"), false);
116  }
117
118  @TestTemplate
119  public void testRetryMetricsAfterOutOfOrderScannerNext() throws Exception {
120    assertRetryMetrics(new OutOfOrderScannerNextException("Injected out-of-order scan"), true);
121  }
122}