/*
 * 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.client.ConnectionConfiguration.HBASE_CLIENT_META_READ_RPC_TIMEOUT_KEY;
import static org.apache.hadoop.hbase.client.ConnectionConfiguration.HBASE_CLIENT_META_SCANNER_TIMEOUT;
import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;

import java.io.IOException;
import java.time.Duration;
import java.util.function.Supplier;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseTestingUtil;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.SingleProcessHBaseCluster.MiniHBaseClusterRegionServer;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.exceptions.OutOfOrderScannerNextException;
import org.apache.hadoop.hbase.ipc.CallTimeoutException;
import org.apache.hadoop.hbase.regionserver.HRegionServer;
import org.apache.hadoop.hbase.regionserver.RSRpcServices;
import org.apache.hadoop.hbase.testclassification.ClientTests;
import org.apache.hadoop.hbase.testclassification.LargeTests;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.hbase.thirdparty.com.google.protobuf.RpcController;
import org.apache.hbase.thirdparty.com.google.protobuf.ServiceException;

import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.ScanRequest;
import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.ScanResponse;

@Tag(ClientTests.TAG)
@Tag(LargeTests.TAG)
public class TestClientScannerTimeouts {

  private static final Logger LOG = LoggerFactory.getLogger(TestClientScannerTimeouts.class);
  private final static HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();

  private static AsyncConnection ASYNC_CONN;
  private static Connection CONN;
  private static final byte[] FAMILY = Bytes.toBytes("testFamily");
  private static final byte[] QUALIFIER = Bytes.toBytes("testQualifier");
  private static final byte[] VALUE = Bytes.toBytes("testValue");

  private static final byte[] ROW0 = Bytes.toBytes("row-0");
  private static final byte[] ROW1 = Bytes.toBytes("row-1");
  private static final byte[] ROW2 = Bytes.toBytes("row-2");
  private static final byte[] ROW3 = Bytes.toBytes("row-3");
  private static final int rpcTimeout = 1000;
  private static final int scanTimeout = 3 * rpcTimeout;
  private static final int metaScanTimeout = 6 * rpcTimeout;
  private static final int CLIENT_RETRIES_NUMBER = 3;

  private static Table table;
  private static AsyncTable<AdvancedScanResultConsumer> asyncTable;

  @BeforeAll
  public static void setUpBeforeClass() throws Exception {
    Configuration conf = TEST_UTIL.getConfiguration();
    conf.setInt(HConstants.HBASE_RPC_TIMEOUT_KEY, rpcTimeout);
    conf.setStrings(HConstants.REGION_SERVER_IMPL, RegionServerWithScanTimeout.class.getName());
    conf.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, CLIENT_RETRIES_NUMBER);
    conf.setInt(HConstants.HBASE_CLIENT_PAUSE, 1000);
    TEST_UTIL.startMiniCluster(1);

    conf.setInt(HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD, scanTimeout);
    conf.setInt(HBASE_CLIENT_META_READ_RPC_TIMEOUT_KEY, metaScanTimeout);
    conf.setInt(HBASE_CLIENT_META_SCANNER_TIMEOUT, metaScanTimeout);
    ASYNC_CONN = ConnectionFactory.createAsyncConnection(conf).get();
    CONN = ASYNC_CONN.toConnection();
  }

  @AfterAll
  public static void tearDownAfterClass() throws Exception {
    CONN.close();
    ASYNC_CONN.close();
    TEST_UTIL.shutdownMiniCluster();
  }

  public void setup(boolean isSystemTable, TestInfo testInfo) throws IOException {
    RSRpcServicesWithScanTimeout.reset();

    String nameAsString = testInfo.getTestMethod().get().getName();
    if (isSystemTable) {
      nameAsString = NamespaceDescriptor.SYSTEM_NAMESPACE_NAME_STR + ":" + nameAsString;
    }
    final TableName tableName = TableName.valueOf(nameAsString);

    TEST_UTIL.createTable(tableName, FAMILY);
    table = CONN.getTable(tableName);
    asyncTable = ASYNC_CONN.getTable(tableName);
    putToTable(table, ROW0);
    putToTable(table, ROW1);
    putToTable(table, ROW2);
    putToTable(table, ROW3);
    LOG.info("Wrote our four values");

    table.getRegionLocator().getAllRegionLocations();

    // reset again incase the creation/population caused anything to trigger
    RSRpcServicesWithScanTimeout.reset();
  }

  private void expectRow(byte[] expected, Result result) {
    assertTrue(Bytes.equals(expected, result.getRow()),
      "Expected row: " + Bytes.toString(expected));
  }

  private void expectNumTries(int expected) {
    assertEquals(expected, RSRpcServicesWithScanTimeout.tryNumber,
      "Expected tryNumber=" + expected + ", actual=" + RSRpcServicesWithScanTimeout.tryNumber);
    // reset for next
    RSRpcServicesWithScanTimeout.tryNumber = 0;
  }

  /**
   * verify that we don't miss any data when encountering an OutOfOrderScannerNextException.
   * Typically, the only way to naturally trigger this is if a client-side timeout causes an
   * erroneous next() call. This is relatively hard to do these days because the server attempts to
   * always return before the timeout. In this test we force the server to throw this exception, so
   * that we can test the retry logic appropriately.
   */
  @Test
  public void testRetryOutOfOrderScannerNextException(TestInfo testInfo) throws IOException {
    expectRetryOutOfOrderScannerNext(this::getScanner, testInfo);
  }

  /**
   * AsyncTable version of above
   */
  @Test
  public void testRetryOutOfOrderScannerNextExceptionAsync(TestInfo testInfo) throws IOException {
    expectRetryOutOfOrderScannerNext(this::getAsyncScanner, testInfo);
  }

  /**
   * verify that we honor the {@link HConstants#HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD} for normal
   * scans.
   */
  @Test
  public void testNormalScanTimeoutOnNext(TestInfo testInfo) throws IOException {
    setup(false, testInfo);
    expectTimeoutOnNext(scanTimeout, this::getScanner);
  }

  /**
   * AsyncTable version of above
   */
  @Test
  public void testNormalScanTimeoutOnNextAsync(TestInfo testInfo) throws IOException {
    setup(false, testInfo);
    expectTimeoutOnNext(scanTimeout, this::getAsyncScanner);
  }

  /**
   * verify that we honor {@link HConstants#HBASE_RPC_READ_TIMEOUT_KEY} for openScanner() calls for
   * meta scans
   */
  @Test
  public void testNormalScanTimeoutOnOpenScanner(TestInfo testInfo) throws IOException {
    setup(false, testInfo);
    expectTimeoutOnOpenScanner(rpcTimeout, this::getScanner);
  }

  /**
   * AsyncTable version of above
   */
  @Test
  public void testNormalScanTimeoutOnOpenScannerAsync(TestInfo testInfo) throws IOException {
    setup(false, testInfo);
    expectTimeoutOnOpenScanner(rpcTimeout, this::getAsyncScanner);
  }

  /**
   * verify that we honor {@link ConnectionConfiguration#HBASE_CLIENT_META_SCANNER_TIMEOUT} for
   * next() calls in meta scans
   */
  @Test
  public void testMetaScanTimeoutOnNext(TestInfo testInfo) throws IOException {
    setup(true, testInfo);
    expectTimeoutOnNext(metaScanTimeout, this::getScanner);
  }

  /**
   * AsyncTable version of above
   */
  @Test
  public void testMetaScanTimeoutOnNextAsync(TestInfo testInfo) throws IOException {
    setup(true, testInfo);
    expectTimeoutOnNext(metaScanTimeout, this::getAsyncScanner);
  }

  /**
   * verify that we honor {@link ConnectionConfiguration#HBASE_CLIENT_META_READ_RPC_TIMEOUT_KEY} for
   * openScanner() calls for meta scans
   */
  @Test
  public void testMetaScanTimeoutOnOpenScanner(TestInfo testInfo) throws IOException {
    setup(true, testInfo);
    expectTimeoutOnOpenScanner(metaScanTimeout, this::getScanner);
  }

  /**
   * AsyncTable version of above
   */
  @Test
  public void testMetaScanTimeoutOnOpenScannerAsync(TestInfo testInfo) throws IOException {
    setup(true, testInfo);
    expectTimeoutOnOpenScanner(metaScanTimeout, this::getAsyncScanner);
  }

  private void expectRetryOutOfOrderScannerNext(Supplier<ResultScanner> scannerSupplier,
    TestInfo testInfo) throws IOException {
    setup(false, testInfo);
    RSRpcServicesWithScanTimeout.seqNoToThrowOn = 1;

    LOG.info(
      "Opening scanner, expecting no errors from first next() call from openScanner response");
    try (ResultScanner scanner = scannerSupplier.get()) {
      Result result = scanner.next();
      expectRow(ROW0, result);
      expectNumTries(0);

      LOG.info("Making first next() RPC, expecting no errors for seqNo 0");
      result = scanner.next();
      expectRow(ROW1, result);
      expectNumTries(0);

      LOG.info(
        "Making second next() RPC, expecting OutOfOrderScannerNextException and appropriate retry");
      result = scanner.next();
      expectRow(ROW2, result);
      expectNumTries(1);

      // reset so no errors. since last call restarted the scan and following
      // call would otherwise fail
      RSRpcServicesWithScanTimeout.seqNoToThrowOn = -1;

      LOG.info("Finishing scan, expecting no errors");
      result = scanner.next();
      expectRow(ROW3, result);
    }

    // the close operation maybe asynchronous, wait until closed to avoid messing up later
    // assertions.
    await().atMost(Duration.ofSeconds(5)).until(() -> RSRpcServicesWithScanTimeout.closed);

    LOG.info("Testing always throw exception");
    byte[][] expectedResults = new byte[][] { ROW0, ROW1, ROW2, ROW3 };

    RSRpcServicesWithScanTimeout.reset();
    RSRpcServicesWithScanTimeout.throwAlways = true;

    // test the case that RPC always throws
    try (ResultScanner scanner = scannerSupplier.get()) {
      for (int i = 0; i < expectedResults.length; i++) {
        LOG.info("Calling scanner.next()");
        Result result = scanner.next();
        assertNotNull(result,
          "missing row index=" + i + ", row=" + Bytes.toStringBinary(expectedResults[i]));
        expectRow(expectedResults[i], result);
      }
    }

    // expect all but the first row (which came from initial openScanner) to have thrown an error
    expectNumTries(expectedResults.length - 1);

  }

  private void expectTimeoutOnNext(int timeout, Supplier<ResultScanner> scannerSupplier)
    throws IOException {
    RSRpcServicesWithScanTimeout.seqNoToSleepOn = 1;
    RSRpcServicesWithScanTimeout.setSleepForTimeout(timeout);

    LOG.info(
      "Opening scanner, expecting no timeouts from first next() call from openScanner response");
    ResultScanner scanner = scannerSupplier.get();
    Result result = scanner.next();
    expectRow(ROW0, result);

    LOG.info("Making first next() RPC, expecting no timeout for seqNo 0");
    result = scanner.next();
    expectRow(ROW1, result);

    LOG.info("Making second next() RPC, expecting timeout");
    long start = System.nanoTime();
    try {
      scanner.next();
      fail("Expected CallTimeoutException");
    } catch (RetriesExhaustedException e) {
      assertTrue(e.getCause() instanceof CallTimeoutException, "Expected CallTimeoutException");
    }
    expectTimeout(start, timeout);
  }

  private void expectTimeoutOnOpenScanner(int timeout, Supplier<ResultScanner> scannerSupplier)
    throws IOException {
    RSRpcServicesWithScanTimeout.setSleepForTimeout(timeout);
    RSRpcServicesWithScanTimeout.sleepOnOpen = true;
    LOG.info("Opening scanner, expecting timeout from first next() call from openScanner response");
    long start = System.nanoTime();
    try {
      scannerSupplier.get().next();
      fail("Expected CallTimeoutException");
    } catch (RetriesExhaustedException e) {
      assertTrue(e.getCause() instanceof CallTimeoutException,
        "Expected CallTimeoutException, but was " + e.getCause());
    }
    expectTimeout(start, timeout);
  }

  private void expectTimeout(long start, int timeout) {
    long duration = System.nanoTime() - start;
    assertTrue(duration >= timeout, "Expected duration >= " + timeout + ", but was " + duration);
  }

  private ResultScanner getScanner() {
    Scan scan = new Scan();
    scan.setCaching(1);
    // make sure we will not fetch rows in background
    scan.setMaxResultSize(1);
    try {
      return table.getScanner(scan);
    } catch (IOException e) {
      throw new RuntimeException(e);
    }
  }

  private ResultScanner getAsyncScanner() {
    Scan scan = new Scan();
    scan.setCaching(1);
    // make sure we will not fetch rows in background
    scan.setMaxResultSize(1);
    return asyncTable.getScanner(scan);
  }

  private void putToTable(Table ht, byte[] rowkey) throws IOException {
    Put put = new Put(rowkey);
    put.addColumn(FAMILY, QUALIFIER, VALUE);
    ht.put(put);
  }

  private static class RegionServerWithScanTimeout extends MiniHBaseClusterRegionServer {
    public RegionServerWithScanTimeout(Configuration conf)
      throws IOException, InterruptedException {
      super(conf);
    }

    @Override
    protected RSRpcServices createRpcServices() throws IOException {
      return new RSRpcServicesWithScanTimeout(this);
    }
  }

  private static class RSRpcServicesWithScanTimeout extends RSRpcServices {
    private long tableScannerId;

    private static volatile long seqNoToThrowOn = -1;
    private static volatile boolean throwAlways = false;
    private static volatile boolean threw;

    private static volatile long seqNoToSleepOn = -1;
    private static volatile boolean sleepOnOpen = false;
    private static volatile boolean slept;
    private static volatile int tryNumber = 0;
    private static volatile boolean closed = false;

    private static volatile int sleepTime = rpcTimeout + 500;

    public static void setSleepForTimeout(int timeout) {
      sleepTime = timeout + 500;
    }

    public static void reset() {
      setSleepForTimeout(scanTimeout);

      seqNoToSleepOn = -1;
      seqNoToThrowOn = -1;
      throwAlways = false;
      threw = false;
      sleepOnOpen = false;
      slept = false;
      tryNumber = 0;
      closed = false;
    }

    public RSRpcServicesWithScanTimeout(HRegionServer rs) throws IOException {
      super(rs);
    }

    @Override
    public ScanResponse scan(final RpcController controller, final ScanRequest request)
      throws ServiceException {
      if (request.hasScannerId()) {
        ScanResponse scanResponse = super.scan(controller, request);
        if (tableScannerId != request.getScannerId()) {
          return scanResponse;
        }
        if (request.getCloseScanner()) {
          closed = true;
          return scanResponse;
        }

        if (
          throwAlways
            || (!threw && request.hasNextCallSeq() && seqNoToThrowOn == request.getNextCallSeq())
        ) {
          threw = true;
          tryNumber++;
          LOG.info("THROWING exception, tryNumber={}, tableScannerId={}", tryNumber,
            tableScannerId);
          throw new ServiceException(new OutOfOrderScannerNextException());
        }

        if (!slept && request.hasNextCallSeq() && seqNoToSleepOn == request.getNextCallSeq()) {
          try {
            LOG.info("SLEEPING " + sleepTime);
            Thread.sleep(sleepTime);
          } catch (InterruptedException e) {
          }
          slept = true;
          tryNumber++;
        }
        return scanResponse;
      } else {
        ScanResponse scanRes = super.scan(controller, request);
        String regionName = Bytes.toString(request.getRegion().getValue().toByteArray());
        if (!regionName.contains(TableName.META_TABLE_NAME.getNameAsString())) {
          tableScannerId = scanRes.getScannerId();
          if (sleepOnOpen) {
            try {
              LOG.info("openScanner SLEEPING " + sleepTime);
              Thread.sleep(sleepTime);
            } catch (InterruptedException e) {
            }
          }
        }
        return scanRes;
      }
    }
  }
}