/*
 * 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.ipc;

import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.timeout;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

import java.io.IOException;
import java.lang.reflect.Field;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Abortable;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.monitoring.MonitoredRPCHandlerImpl;
import org.apache.hadoop.hbase.regionserver.RSAnnotationReadingPriorityFunction;
import org.apache.hadoop.hbase.testclassification.MediumTests;
import org.apache.hadoop.hbase.testclassification.RPCTests;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.EnvironmentEdge;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
import org.apache.hadoop.hbase.util.Threads;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInfo;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.hbase.thirdparty.com.google.common.collect.ImmutableList;
import org.apache.hbase.thirdparty.com.google.common.collect.ImmutableMap;
import org.apache.hbase.thirdparty.com.google.common.collect.ImmutableSet;
import org.apache.hbase.thirdparty.com.google.common.collect.Maps;

import org.apache.hadoop.hbase.shaded.protobuf.RequestConverter;
import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.ScanRequest;
import org.apache.hadoop.hbase.shaded.protobuf.generated.RPCProtos;
import org.apache.hadoop.hbase.shaded.protobuf.generated.RPCProtos.RequestHeader;

@Tag(RPCTests.TAG)
@Tag(MediumTests.TAG)
public class TestSimpleRpcScheduler {

  private static final Logger LOG = LoggerFactory.getLogger(TestSimpleRpcScheduler.class);

  private final RpcScheduler.Context CONTEXT = new RpcScheduler.Context() {
    @Override
    public InetSocketAddress getListenerAddress() {
      return InetSocketAddress.createUnresolved("127.0.0.1", 1000);
    }
  };
  private Configuration conf;
  private String testMethodName;
  private final AtomicInteger requestProcessed = new AtomicInteger(0);

  @BeforeEach
  public void setUp(TestInfo testInfo) {
    testMethodName = testInfo.getTestMethod().get().getName();
    conf = HBaseConfiguration.create();
  }

  @Test
  public void testBasic() throws IOException, InterruptedException {

    PriorityFunction qosFunction = mock(PriorityFunction.class);
    RpcScheduler scheduler = new SimpleRpcScheduler(conf, 10, 0, 0, qosFunction, 0);
    scheduler.init(CONTEXT);
    scheduler.start();
    CallRunner task = createMockTask(HConstants.NORMAL_QOS);
    task.setStatus(new MonitoredRPCHandlerImpl("test"));
    scheduler.dispatch(task);
    verify(task, timeout(10000)).run();
    scheduler.stop();
  }

  private RpcScheduler disableHandlers(RpcScheduler scheduler) {
    try {
      Field ExecutorField = scheduler.getClass().getDeclaredField("callExecutor");
      ExecutorField.setAccessible(true);

      RpcExecutor rpcExecutor = (RpcExecutor) ExecutorField.get(scheduler);

      Field handlerCountField =
        rpcExecutor.getClass().getSuperclass().getSuperclass().getDeclaredField("handlerCount");

      handlerCountField.setAccessible(true);
      handlerCountField.set(rpcExecutor, 0);

      Field numCallQueuesField =
        rpcExecutor.getClass().getSuperclass().getSuperclass().getDeclaredField("numCallQueues");

      numCallQueuesField.setAccessible(true);
      numCallQueuesField.set(rpcExecutor, 1);

      Field currentQueueLimitField = rpcExecutor.getClass().getSuperclass().getSuperclass()
        .getDeclaredField("currentQueueLimit");

      currentQueueLimitField.setAccessible(true);
      currentQueueLimitField.set(rpcExecutor, 100);

    } catch (NoSuchFieldException e) {
      LOG.error("No such field exception" + e);
    } catch (IllegalAccessException e) {
      LOG.error("Illegal access exception" + e);
    }

    return scheduler;
  }

  @Test
  public void testCallQueueInfo() throws IOException, InterruptedException {

    PriorityFunction qosFunction = mock(PriorityFunction.class);
    RpcScheduler scheduler = new SimpleRpcScheduler(conf, 0, 0, 0, qosFunction, 0);

    scheduler.init(CONTEXT);

    // Set the handlers to zero. So that number of requests in call Queue can be tested
    scheduler = disableHandlers(scheduler);
    scheduler.start();

    int totalCallMethods = 10;
    for (int i = totalCallMethods; i > 0; i--) {
      CallRunner task = createMockTask(HConstants.NORMAL_QOS);
      task.setStatus(new MonitoredRPCHandlerImpl("test"));
      scheduler.dispatch(task);
    }

    CallQueueInfo callQueueInfo = scheduler.getCallQueueInfo();

    for (String callQueueName : callQueueInfo.getCallQueueNames()) {

      for (String calledMethod : callQueueInfo.getCalledMethodNames(callQueueName)) {
        assertEquals(totalCallMethods,
          callQueueInfo.getCallMethodCount(callQueueName, calledMethod));
      }

    }

    scheduler.stop();

  }

  @Test
  public void testHandlerIsolation() throws IOException, InterruptedException {
    CallRunner generalTask = createMockTask(HConstants.NORMAL_QOS);
    CallRunner priorityTask = createMockTask(HConstants.HIGH_QOS + 1);
    CallRunner replicationTask = createMockTask(HConstants.REPLICATION_QOS);
    List<CallRunner> tasks = ImmutableList.of(generalTask, priorityTask, replicationTask);
    Map<CallRunner, Integer> qos = ImmutableMap.of(generalTask, 0, priorityTask,
      HConstants.HIGH_QOS + 1, replicationTask, HConstants.REPLICATION_QOS);
    PriorityFunction qosFunction = mock(PriorityFunction.class);
    final Map<CallRunner, Thread> handlerThreads = Maps.newHashMap();
    final CountDownLatch countDownLatch = new CountDownLatch(tasks.size());
    Answer<Void> answerToRun = new Answer<Void>() {
      @Override
      public Void answer(InvocationOnMock invocationOnMock) throws Throwable {
        synchronized (handlerThreads) {
          handlerThreads.put((CallRunner) invocationOnMock.getMock(), Thread.currentThread());
        }
        countDownLatch.countDown();
        return null;
      }
    };
    for (CallRunner task : tasks) {
      task.setStatus(new MonitoredRPCHandlerImpl("test"));
      doAnswer(answerToRun).when(task).run();
    }

    RpcScheduler scheduler =
      new SimpleRpcScheduler(conf, 1, 1, 1, qosFunction, HConstants.HIGH_QOS);
    scheduler.init(CONTEXT);
    scheduler.start();
    for (CallRunner task : tasks) {
      when(qosFunction.getPriority(any(), any(), any())).thenReturn(qos.get(task));
      scheduler.dispatch(task);
    }
    for (CallRunner task : tasks) {
      verify(task, timeout(10000)).run();
    }
    scheduler.stop();

    // Tests that these requests are handled by three distinct threads.
    countDownLatch.await();
    assertEquals(3, ImmutableSet.copyOf(handlerThreads.values()).size());
  }

  private CallRunner createMockTask(int priority) {
    ServerCall call = mock(ServerCall.class);
    CallRunner task = mock(CallRunner.class);
    RequestHeader header = RequestHeader.newBuilder().setPriority(priority).build();
    when(task.getRpcCall()).thenReturn(call);
    when(call.getHeader()).thenReturn(header);
    when(call.getReceiveTime()).thenReturn(EnvironmentEdgeManager.currentTime());
    return task;
  }

  @Test
  public void testRpcScheduler() throws Exception {
    testRpcScheduler(RpcExecutor.CALL_QUEUE_TYPE_DEADLINE_CONF_VALUE);
    testRpcScheduler(RpcExecutor.CALL_QUEUE_TYPE_FIFO_CONF_VALUE);
  }

  @Test
  public void testPluggableRpcQueue() throws Exception {
    testRpcScheduler(RpcExecutor.CALL_QUEUE_TYPE_PLUGGABLE_CONF_VALUE,
      "org.apache.hadoop.hbase.ipc.TestPluggableQueueImpl");

    try {
      testRpcScheduler(RpcExecutor.CALL_QUEUE_TYPE_PLUGGABLE_CONF_VALUE, "MissingClass");
      fail("Expected a PluggableRpcQueueNotFound for unloaded class");
    } catch (PluggableRpcQueueNotFound e) {
      // expected
    } catch (Exception e) {
      fail("Expected a PluggableRpcQueueNotFound for unloaded class, but instead got " + e);
    }

    try {
      testRpcScheduler(RpcExecutor.CALL_QUEUE_TYPE_PLUGGABLE_CONF_VALUE,
        "org.apache.hadoop.hbase.ipc.SimpleRpcServer");
      fail("Expected a PluggableRpcQueueNotFound for incompatible class");
    } catch (PluggableRpcQueueNotFound e) {
      // expected
    } catch (Exception e) {
      fail("Expected a PluggableRpcQueueNotFound for incompatible class, but instead got " + e);
    }
  }

  @Test
  public void testPluggableRpcQueueWireUpWithFastPathExecutor() throws Exception {
    String queueType = RpcExecutor.CALL_QUEUE_TYPE_PLUGGABLE_CONF_VALUE;
    Configuration schedConf = HBaseConfiguration.create();
    schedConf.set(RpcExecutor.CALL_QUEUE_TYPE_CONF_KEY, queueType);
    schedConf.set(RpcExecutor.PLUGGABLE_CALL_QUEUE_CLASS_NAME,
      "org.apache.hadoop.hbase.ipc.TestPluggableQueueImpl");
    schedConf.setBoolean(RpcExecutor.PLUGGABLE_CALL_QUEUE_WITH_FAST_PATH_ENABLED, true);

    PriorityFunction priority = mock(PriorityFunction.class);
    when(priority.getPriority(any(), any(), any())).thenReturn(HConstants.NORMAL_QOS);
    SimpleRpcScheduler scheduler =
      new SimpleRpcScheduler(schedConf, 0, 0, 0, priority, HConstants.QOS_THRESHOLD);

    Field f = scheduler.getClass().getDeclaredField("callExecutor");
    f.setAccessible(true);
    assertTrue(f.get(scheduler) instanceof FastPathBalancedQueueRpcExecutor);
  }

  @Test
  public void testPluggableRpcQueueWireUpWithoutFastPathExecutor() throws Exception {
    String queueType = RpcExecutor.CALL_QUEUE_TYPE_PLUGGABLE_CONF_VALUE;
    Configuration schedConf = HBaseConfiguration.create();
    schedConf.set(RpcExecutor.CALL_QUEUE_TYPE_CONF_KEY, queueType);
    schedConf.set(RpcExecutor.PLUGGABLE_CALL_QUEUE_CLASS_NAME,
      "org.apache.hadoop.hbase.ipc.TestPluggableQueueImpl");

    PriorityFunction priority = mock(PriorityFunction.class);
    when(priority.getPriority(any(), any(), any())).thenReturn(HConstants.NORMAL_QOS);
    SimpleRpcScheduler scheduler =
      new SimpleRpcScheduler(schedConf, 0, 0, 0, priority, HConstants.QOS_THRESHOLD);

    Field f = scheduler.getClass().getDeclaredField("callExecutor");
    f.setAccessible(true);
    assertTrue(f.get(scheduler) instanceof BalancedQueueRpcExecutor);
  }

  @Test
  public void testPluggableRpcQueueCanListenToConfigurationChanges() throws Exception {

    Configuration schedConf = HBaseConfiguration.create();

    schedConf.setInt(HConstants.REGION_SERVER_HANDLER_COUNT, 2);
    schedConf.setInt("hbase.ipc.server.max.callqueue.length", 5);
    schedConf.set(RpcExecutor.CALL_QUEUE_TYPE_CONF_KEY,
      RpcExecutor.CALL_QUEUE_TYPE_PLUGGABLE_CONF_VALUE);
    schedConf.set(RpcExecutor.PLUGGABLE_CALL_QUEUE_CLASS_NAME,
      "org.apache.hadoop.hbase.ipc.TestPluggableQueueImpl");

    PriorityFunction priority = mock(PriorityFunction.class);
    when(priority.getPriority(any(), any(), any())).thenReturn(HConstants.NORMAL_QOS);
    SimpleRpcScheduler scheduler =
      new SimpleRpcScheduler(schedConf, 0, 0, 0, priority, HConstants.QOS_THRESHOLD);
    try {
      scheduler.start();

      CallRunner putCallTask = mock(CallRunner.class);
      ServerCall putCall = mock(ServerCall.class);
      putCall.param =
        RequestConverter.buildMutateRequest(Bytes.toBytes("abc"), new Put(Bytes.toBytes("row")));
      RequestHeader putHead = RequestHeader.newBuilder().setMethodName("mutate").build();
      when(putCallTask.getRpcCall()).thenReturn(putCall);
      when(putCall.getHeader()).thenReturn(putHead);

      assertTrue(scheduler.dispatch(putCallTask));

      schedConf.setInt("hbase.ipc.server.max.callqueue.length", 4);
      scheduler.onConfigurationChange(schedConf);
      assertTrue(TestPluggableQueueImpl.hasObservedARecentConfigurationChange());
      waitUntilQueueEmpty(scheduler);
    } finally {
      scheduler.stop();
    }
  }

  private void testRpcScheduler(final String queueType) throws Exception {
    testRpcScheduler(queueType, null);
  }

  private void testRpcScheduler(final String queueType, final String pluggableQueueClass)
    throws Exception {
    Configuration schedConf = HBaseConfiguration.create();
    schedConf.set(RpcExecutor.CALL_QUEUE_TYPE_CONF_KEY, queueType);

    if (RpcExecutor.CALL_QUEUE_TYPE_PLUGGABLE_CONF_VALUE.equals(queueType)) {
      schedConf.set(RpcExecutor.PLUGGABLE_CALL_QUEUE_CLASS_NAME, pluggableQueueClass);
    }

    PriorityFunction priority = mock(PriorityFunction.class);
    when(priority.getPriority(any(), any(), any())).thenReturn(HConstants.NORMAL_QOS);

    RpcScheduler scheduler =
      new SimpleRpcScheduler(schedConf, 1, 1, 1, priority, HConstants.QOS_THRESHOLD);
    try {
      scheduler.start();

      CallRunner smallCallTask = mock(CallRunner.class);
      ServerCall smallCall = mock(ServerCall.class);
      RequestHeader smallHead = RequestHeader.newBuilder().setCallId(1).build();
      when(smallCallTask.getRpcCall()).thenReturn(smallCall);
      when(smallCall.getHeader()).thenReturn(smallHead);

      CallRunner largeCallTask = mock(CallRunner.class);
      ServerCall largeCall = mock(ServerCall.class);
      RequestHeader largeHead = RequestHeader.newBuilder().setCallId(50).build();
      when(largeCallTask.getRpcCall()).thenReturn(largeCall);
      when(largeCall.getHeader()).thenReturn(largeHead);

      CallRunner hugeCallTask = mock(CallRunner.class);
      ServerCall hugeCall = mock(ServerCall.class);
      RequestHeader hugeHead = RequestHeader.newBuilder().setCallId(100).build();
      when(hugeCallTask.getRpcCall()).thenReturn(hugeCall);
      when(hugeCall.getHeader()).thenReturn(hugeHead);

      when(priority.getDeadline(eq(smallHead), any())).thenReturn(0L);
      when(priority.getDeadline(eq(largeHead), any())).thenReturn(50L);
      when(priority.getDeadline(eq(hugeHead), any())).thenReturn(100L);

      final ArrayList<Integer> work = new ArrayList<>();
      doAnswerTaskExecution(smallCallTask, work, 10, 250);
      doAnswerTaskExecution(largeCallTask, work, 50, 250);
      doAnswerTaskExecution(hugeCallTask, work, 100, 250);

      scheduler.dispatch(smallCallTask);
      scheduler.dispatch(smallCallTask);
      scheduler.dispatch(smallCallTask);
      scheduler.dispatch(hugeCallTask);
      scheduler.dispatch(smallCallTask);
      scheduler.dispatch(largeCallTask);
      scheduler.dispatch(smallCallTask);
      scheduler.dispatch(smallCallTask);

      while (work.size() < 8) {
        Thread.sleep(100);
      }

      int seqSum = 0;
      int totalTime = 0;
      for (int i = 0; i < work.size(); ++i) {
        LOG.debug("Request i=" + i + " value=" + work.get(i));
        seqSum += work.get(i);
        totalTime += seqSum;
      }
      LOG.debug("Total Time: " + totalTime);

      // -> [small small small huge small large small small]
      // -> NO REORDER [10 10 10 100 10 50 10 10] -> 930 (FIFO Queue)
      // -> WITH REORDER [10 10 10 10 10 10 50 100] -> 530 (Deadline Queue)
      if (queueType.equals(RpcExecutor.CALL_QUEUE_TYPE_DEADLINE_CONF_VALUE)) {
        assertEquals(530, totalTime);
      } else if (
        queueType.equals(RpcExecutor.CALL_QUEUE_TYPE_FIFO_CONF_VALUE)
          || queueType.equals(RpcExecutor.CALL_QUEUE_TYPE_PLUGGABLE_CONF_VALUE)
      ) {
        assertEquals(930, totalTime);
      }
    } finally {
      scheduler.stop();
    }
  }

  @Test
  public void testScanQueueWithZeroScanRatio() throws Exception {

    Configuration schedConf = HBaseConfiguration.create();
    schedConf.setFloat(RpcExecutor.CALL_QUEUE_HANDLER_FACTOR_CONF_KEY, 1.0f);
    schedConf.setFloat(RWQueueRpcExecutor.CALL_QUEUE_READ_SHARE_CONF_KEY, 0.5f);
    schedConf.setFloat(RWQueueRpcExecutor.CALL_QUEUE_SCAN_SHARE_CONF_KEY, 0f);

    PriorityFunction priority = mock(PriorityFunction.class);
    when(priority.getPriority(any(), any(), any())).thenReturn(HConstants.NORMAL_QOS);

    RpcScheduler scheduler =
      new SimpleRpcScheduler(schedConf, 2, 1, 1, priority, HConstants.QOS_THRESHOLD);
    assertNotEquals(null, scheduler);
  }

  @Test
  public void testScanQueues() throws Exception {
    Configuration schedConf = HBaseConfiguration.create();
    schedConf.setFloat(RpcExecutor.CALL_QUEUE_HANDLER_FACTOR_CONF_KEY, 1.0f);
    schedConf.setFloat(RWQueueRpcExecutor.CALL_QUEUE_READ_SHARE_CONF_KEY, 0.7f);
    schedConf.setFloat(RWQueueRpcExecutor.CALL_QUEUE_SCAN_SHARE_CONF_KEY, 0.5f);

    PriorityFunction priority = mock(PriorityFunction.class);
    when(priority.getPriority(any(), any(), any())).thenReturn(HConstants.NORMAL_QOS);

    RpcScheduler scheduler =
      new SimpleRpcScheduler(schedConf, 3, 1, 1, priority, HConstants.QOS_THRESHOLD);
    try {
      scheduler.start();

      CallRunner putCallTask = mock(CallRunner.class);
      ServerCall putCall = mock(ServerCall.class);
      putCall.param =
        RequestConverter.buildMutateRequest(Bytes.toBytes("abc"), new Put(Bytes.toBytes("row")));
      RequestHeader putHead = RequestHeader.newBuilder().setMethodName("mutate").build();
      when(putCallTask.getRpcCall()).thenReturn(putCall);
      when(putCall.getHeader()).thenReturn(putHead);
      when(putCall.getParam()).thenReturn(putCall.param);

      CallRunner getCallTask = mock(CallRunner.class);
      ServerCall getCall = mock(ServerCall.class);
      RequestHeader getHead = RequestHeader.newBuilder().setMethodName("get").build();
      when(getCallTask.getRpcCall()).thenReturn(getCall);
      when(getCall.getHeader()).thenReturn(getHead);

      CallRunner scanCallTask = mock(CallRunner.class);
      ServerCall scanCall = mock(ServerCall.class);
      scanCall.param = ScanRequest.newBuilder().build();
      RequestHeader scanHead = RequestHeader.newBuilder().setMethodName("scan").build();
      when(scanCallTask.getRpcCall()).thenReturn(scanCall);
      when(scanCall.getHeader()).thenReturn(scanHead);
      when(scanCall.getParam()).thenReturn(scanCall.param);

      CallRunner bulkLoadCallTask = mock(CallRunner.class);
      ServerCall bulkLoadCall = mock(ServerCall.class);
      bulkLoadCall.param = ScanRequest.newBuilder().build();
      RequestHeader bulkLadHead = RequestHeader.newBuilder().setMethodName("bulkload").build();
      when(bulkLoadCallTask.getRpcCall()).thenReturn(bulkLoadCall);
      when(bulkLoadCall.getHeader()).thenReturn(bulkLadHead);
      when(bulkLoadCall.getParam()).thenReturn(bulkLoadCall.param);

      ArrayList<Integer> work = new ArrayList<>();
      doAnswerTaskExecution(putCallTask, work, 1, 1000);
      doAnswerTaskExecution(getCallTask, work, 2, 1000);
      doAnswerTaskExecution(scanCallTask, work, 3, 1000);
      doAnswerTaskExecution(bulkLoadCallTask, work, 4, 1000);

      // There are 3 queues: [puts], [gets], [scans], [bulkload]
      // so the calls will be interleaved
      scheduler.dispatch(putCallTask);
      scheduler.dispatch(putCallTask);
      scheduler.dispatch(putCallTask);
      scheduler.dispatch(getCallTask);
      scheduler.dispatch(getCallTask);
      scheduler.dispatch(getCallTask);
      scheduler.dispatch(scanCallTask);
      scheduler.dispatch(scanCallTask);
      scheduler.dispatch(scanCallTask);
      scheduler.dispatch(bulkLoadCallTask);
      scheduler.dispatch(bulkLoadCallTask);
      scheduler.dispatch(bulkLoadCallTask);
      while (work.size() < 6) {
        Thread.sleep(100);
      }

      for (int i = 0; i < work.size() - 2; i += 3) {
        assertNotEquals(work.get(i + 0), work.get(i + 1));
        assertNotEquals(work.get(i + 0), work.get(i + 2));
        assertNotEquals(work.get(i + 1), work.get(i + 2));
      }
    } finally {
      scheduler.stop();
    }
  }

  private void doAnswerTaskExecution(final CallRunner callTask, final ArrayList<Integer> results,
    final int value, final int sleepInterval) {
    callTask.setStatus(new MonitoredRPCHandlerImpl("test"));
    doAnswer(new Answer<Object>() {
      @Override
      public Object answer(InvocationOnMock invocation) {
        synchronized (results) {
          results.add(value);
        }
        Threads.sleepWithoutInterrupt(sleepInterval);
        return null;
      }
    }).when(callTask).run();
  }

  private static void waitUntilQueueEmpty(SimpleRpcScheduler scheduler)
    throws InterruptedException {
    while (scheduler.getGeneralQueueLength() > 0) {
      Thread.sleep(100);
    }
  }

  @Test
  public void testSoftAndHardQueueLimits() throws Exception {

    Configuration schedConf = HBaseConfiguration.create();

    schedConf.setInt(HConstants.REGION_SERVER_HANDLER_COUNT, 0);
    schedConf.setInt("hbase.ipc.server.max.callqueue.length", 5);
    schedConf.set(RpcExecutor.CALL_QUEUE_TYPE_CONF_KEY,
      RpcExecutor.CALL_QUEUE_TYPE_DEADLINE_CONF_VALUE);

    PriorityFunction priority = mock(PriorityFunction.class);
    when(priority.getPriority(any(), any(), any())).thenReturn(HConstants.NORMAL_QOS);
    SimpleRpcScheduler scheduler =
      new SimpleRpcScheduler(schedConf, 0, 0, 0, priority, HConstants.QOS_THRESHOLD);
    try {
      scheduler.start();

      CallRunner putCallTask = mock(CallRunner.class);
      ServerCall putCall = mock(ServerCall.class);
      putCall.param =
        RequestConverter.buildMutateRequest(Bytes.toBytes("abc"), new Put(Bytes.toBytes("row")));
      RequestHeader putHead = RequestHeader.newBuilder().setMethodName("mutate").build();
      when(putCallTask.getRpcCall()).thenReturn(putCall);
      when(putCall.getHeader()).thenReturn(putHead);

      assertTrue(scheduler.dispatch(putCallTask));

      schedConf.setInt("hbase.ipc.server.max.callqueue.length", 0);
      scheduler.onConfigurationChange(schedConf);
      assertFalse(scheduler.dispatch(putCallTask));
      waitUntilQueueEmpty(scheduler);
      schedConf.setInt("hbase.ipc.server.max.callqueue.length", 1);
      scheduler.onConfigurationChange(schedConf);
      assertTrue(scheduler.dispatch(putCallTask));
    } finally {
      scheduler.stop();
    }
  }

  private final class CoDelEnvironmentEdge implements EnvironmentEdge {

    private long perRequestOffset;
    private long gapInRequests;
    private long testStartTime = 0;

    private int requestCount = 0;

    private final Set<String> threadNamePrefixes = new HashSet<>();
    private final String testThread = Thread.currentThread().getName();

    @Override
    public long currentTime() {
      String threadName = Thread.currentThread().getName();
      if (threadName.equals(testThread)) {
        // test thread will create a request and add that to scheduler
        // Replicating a constant rate of request arrival.
        long requestStartTime = testStartTime + requestCount * gapInRequests;
        requestCount++;
        return requestStartTime;
      }
      // handler thread will pick the request from queue, this time will depend on processing time
      // of handler
      for (String threadNamePrefix : threadNamePrefixes) {
        if (threadName.startsWith(threadNamePrefix)) {
          long requestPickTime;
          if (gapInRequests >= perRequestOffset) {
            // it means handler can complete the processing of last request and will be free when it
            // will come to serve next request. We don't need it in fast path but just in case for
            // other tests
            requestPickTime = testStartTime + requestProcessed.get() * gapInRequests;
          } else {
            // this means handler will still be busy processing the last requests when new request
            // will come in queue
            requestPickTime = testStartTime + requestProcessed.get() * perRequestOffset;
          }
          return requestPickTime;
        }
      }
      return System.currentTimeMillis();
    }

    public void startTestFor(int arrivalRatePerSecond, long requestProcessingTime) {
      this.perRequestOffset = requestProcessingTime;
      this.gapInRequests = 1000L / arrivalRatePerSecond;
      this.requestCount = 0;
      requestProcessed.set(0);
      this.testStartTime = System.currentTimeMillis();
    }
  }

  // This test is fixing the processing time and arrival rate of the request and checking if CoDel
  // can utilize the maximum capacity of system
  @Test
  public void testCoDelScheduling() throws Exception {
    CoDelEnvironmentEdge envEdge = new CoDelEnvironmentEdge();
    envEdge.threadNamePrefixes.add("RpcServer.default.FPBQ.Codel.handler");
    Configuration schedConf = HBaseConfiguration.create();
    schedConf.setInt(RpcScheduler.IPC_SERVER_MAX_CALLQUEUE_LENGTH, 250);
    schedConf.set(RpcExecutor.CALL_QUEUE_TYPE_CONF_KEY,
      RpcExecutor.CALL_QUEUE_TYPE_CODEL_CONF_VALUE);
    schedConf.setInt(RpcExecutor.CALL_QUEUE_CODEL_TARGET_DELAY, 5);
    PriorityFunction priority = mock(PriorityFunction.class);
    when(priority.getPriority(any(), any(), any())).thenReturn(HConstants.NORMAL_QOS);
    SimpleRpcScheduler scheduler =
      new SimpleRpcScheduler(schedConf, 1, 1, 1, priority, HConstants.QOS_THRESHOLD);

    try {
      scheduler.start();
      EnvironmentEdgeManager.injectEdge(envEdge);

      // incoming traffic < capacity
      envEdge.startTestFor(20, 40);
      for (int i = 0; i < 100; i++) {
        long time = EnvironmentEdgeManager.currentTime();
        CallRunner cr = getMockedCallRunner(time, 2);
        scheduler.dispatch(cr);
        Thread.sleep(50);
      }
      // LOG.info("Loop done");
      // make sure fast calls are handled
      waitUntilQueueEmpty(scheduler);
      Thread.sleep(100);
      assertEquals(0, scheduler.getNumGeneralCallsDropped(),
        "None of these calls should have been discarded");

      // incoming traffic = capacity
      envEdge.startTestFor(20, 50);
      for (int i = 0; i < 20; i++) {
        long time = EnvironmentEdgeManager.currentTime();
        CallRunner cr = getMockedCallRunner(time, 2);
        scheduler.dispatch(cr);
        // We have mocked the arrival time and the time request get picked from queue but we need to
        // also make sure that queue fill at the arrival rate
        Thread.sleep(50);
      }

      // make sure somewhat slow calls are handled
      waitUntilQueueEmpty(scheduler);
      Thread.sleep(100);
      assertEquals(0, scheduler.getNumGeneralCallsDropped(),
        "None of these calls should have been discarded");

      // incoming traffic > capacity
      envEdge.startTestFor(20, 60);
      // now slow calls and the ones to be dropped
      for (int i = 0; i < 60; i++) {
        long time = EnvironmentEdgeManager.currentTime();
        CallRunner cr = getMockedCallRunner(time, 100);
        scheduler.dispatch(cr);
        Thread.sleep(50);
      }

      // make sure somewhat slow calls are handled
      waitUntilQueueEmpty(scheduler);
      Thread.sleep(100);
      // 60 calls will take 3 second with 20 rps. And in 3 second with 60 processing time we can
      // only serve 50 request
      assertTrue(requestProcessed.get() >= 48,
        "Number of processed requests should be greater then 90% of capacity"
          + requestProcessed.get());
      assertTrue(scheduler.getNumGeneralCallsDropped() >= 9,
        "There should have been at least 9 calls dropped however there were "
          + scheduler.getNumGeneralCallsDropped());

    } finally {
      scheduler.stop();
    }
  }

  @Test
  public void testFastPathBalancedQueueRpcExecutorWithQueueLength0() throws Exception {
    String name = testMethodName;
    int handlerCount = 1;
    String callQueueType = RpcExecutor.CALL_QUEUE_TYPE_CODEL_CONF_VALUE;
    int maxQueueLength = 0;
    PriorityFunction priority = mock(PriorityFunction.class);
    Configuration conf = HBaseConfiguration.create();
    Abortable abortable = mock(Abortable.class);
    FastPathBalancedQueueRpcExecutor executor =
      Mockito.spy(new FastPathBalancedQueueRpcExecutor(name, handlerCount, callQueueType,
        maxQueueLength, priority, conf, abortable));
    CallRunner task = mock(CallRunner.class);
    assertFalse(executor.dispatch(task));
    // make sure we never internally get a handler, which would skip the queue validation
    Mockito.verify(executor, Mockito.never()).getHandler(Mockito.any(), Mockito.anyDouble(),
      Mockito.anyInt(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any());
  }

  @Test
  public void testMetaRWScanQueues() throws Exception {
    Configuration schedConf = HBaseConfiguration.create();
    schedConf.setFloat(MetaRWQueueRpcExecutor.META_CALL_QUEUE_HANDLER_FACTOR_CONF_KEY, 1.0f);
    schedConf.setFloat(MetaRWQueueRpcExecutor.META_CALL_QUEUE_READ_SHARE_CONF_KEY, 0.7f);
    schedConf.setFloat(MetaRWQueueRpcExecutor.META_CALL_QUEUE_SCAN_SHARE_CONF_KEY, 0.5f);

    PriorityFunction priority = mock(PriorityFunction.class);
    when(priority.getPriority(any(), any(), any())).thenReturn(HConstants.HIGH_QOS);

    RpcScheduler scheduler =
      new SimpleRpcScheduler(schedConf, 3, 3, 1, priority, HConstants.QOS_THRESHOLD);
    try {
      scheduler.start();

      CallRunner putCallTask = mock(CallRunner.class);
      ServerCall putCall = mock(ServerCall.class);
      putCall.param =
        RequestConverter.buildMutateRequest(Bytes.toBytes("abc"), new Put(Bytes.toBytes("row")));
      RequestHeader putHead = RequestHeader.newBuilder().setMethodName("mutate").build();
      when(putCallTask.getRpcCall()).thenReturn(putCall);
      when(putCall.getHeader()).thenReturn(putHead);
      when(putCall.getParam()).thenReturn(putCall.param);

      CallRunner clientReadCallTask = mock(CallRunner.class);
      ServerCall clientReadCall = mock(ServerCall.class);
      RequestHeader clientReadHead = RequestHeader.newBuilder().setMethodName("get").build();
      when(clientReadCallTask.getRpcCall()).thenReturn(clientReadCall);
      when(clientReadCall.getHeader()).thenReturn(clientReadHead);

      CallRunner internalReadCallTask = mock(CallRunner.class);
      ServerCall internalReadCall = mock(ServerCall.class);
      internalReadCall.param = ScanRequest.newBuilder().build();
      RequestHeader masterReadHead = RequestHeader.newBuilder().setMethodName("scan")
        .setPriority(RSAnnotationReadingPriorityFunction.INTERNAL_READ_QOS).build();
      when(internalReadCallTask.getRpcCall()).thenReturn(internalReadCall);
      when(internalReadCall.getHeader()).thenReturn(masterReadHead);
      when(internalReadCall.getParam()).thenReturn(internalReadCall.param);

      ArrayList<Integer> work = new ArrayList<>();
      doAnswerTaskExecution(putCallTask, work, 1, 1000);
      doAnswerTaskExecution(clientReadCallTask, work, 2, 1000);
      doAnswerTaskExecution(internalReadCallTask, work, 3, 1000);

      // There are 3 queues: [puts], [gets], [scans]
      // so the calls will be interleaved
      scheduler.dispatch(putCallTask);
      scheduler.dispatch(putCallTask);
      scheduler.dispatch(putCallTask);
      scheduler.dispatch(clientReadCallTask);
      scheduler.dispatch(clientReadCallTask);
      scheduler.dispatch(clientReadCallTask);
      scheduler.dispatch(internalReadCallTask);
      scheduler.dispatch(internalReadCallTask);
      scheduler.dispatch(internalReadCallTask);

      while (work.size() < 6) {
        Thread.sleep(100);
      }

      for (int i = 0; i < work.size() - 2; i += 3) {
        assertNotEquals(work.get(i + 0), work.get(i + 1));
        assertNotEquals(work.get(i + 0), work.get(i + 2));
        assertNotEquals(work.get(i + 1), work.get(i + 2));
      }
    } finally {
      scheduler.stop();
    }
  }

  // Get mocked call that has the CallRunner sleep for a while so that the fast
  // path isn't hit.
  private CallRunner getMockedCallRunner(long timestamp, final long sleepTime) throws IOException {
    ServerCall putCall = new ServerCall(1, null, null,
      RPCProtos.RequestHeader.newBuilder().setMethodName("mutate").build(),
      RequestConverter.buildMutateRequest(Bytes.toBytes("abc"), new Put(Bytes.toBytes("row"))),
      null, null, 9, null, timestamp, 0, null, null, null) {

      @Override
      public void sendResponseIfReady() throws IOException {
      }
    };

    return new CallRunner(null, putCall) {
      @Override
      public void run() {
        if (sleepTime <= 0) {
          return;
        }
        try {
          Thread.sleep(sleepTime);
          requestProcessed.incrementAndGet();
        } catch (InterruptedException e) {
        }
      }

      @Override
      public RpcCall getRpcCall() {
        return putCall;
      }

      @Override
      public void drop() {
      }
    };
  }

  /**
   * Test LIFO switching behavior through actual RPC calls. This test verifies that when the queue
   * fills beyond the LIFO threshold, newer calls are processed before older calls (LIFO mode).
   */
  @Test
  public void testCoDelLifoWithRpcCalls() throws Exception {
    Configuration testConf = HBaseConfiguration.create();
    testConf.set(RpcExecutor.CALL_QUEUE_TYPE_CONF_KEY,
      RpcExecutor.CALL_QUEUE_TYPE_CODEL_CONF_VALUE);
    int maxCallQueueLength = 50;
    double coDelLifoThreshold = 0.8;
    testConf.setInt(RpcScheduler.IPC_SERVER_MAX_CALLQUEUE_LENGTH, maxCallQueueLength);
    testConf.setDouble(RpcExecutor.CALL_QUEUE_CODEL_LIFO_THRESHOLD, coDelLifoThreshold);
    testConf.setInt(RpcExecutor.CALL_QUEUE_CODEL_TARGET_DELAY, 100);
    testConf.setInt(RpcExecutor.CALL_QUEUE_CODEL_INTERVAL, 100);
    testConf.setInt(HConstants.REGION_SERVER_HANDLER_COUNT, 1); // Single handler to control
                                                                // processing

    PriorityFunction priority = mock(PriorityFunction.class);
    when(priority.getPriority(any(), any(), any())).thenReturn(HConstants.NORMAL_QOS);
    SimpleRpcScheduler scheduler =
      new SimpleRpcScheduler(testConf, 1, 0, 0, priority, HConstants.QOS_THRESHOLD);

    try {
      scheduler.init(CONTEXT);
      scheduler.start();

      // Track completion order
      final List<Integer> completedCalls = Collections.synchronizedList(new ArrayList<>());

      // Dispatch many slow calls rapidly to fill the queue beyond 80% threshold
      // With queue limit of 50, we need > 40 calls to cross 80%
      int numCalls = 48;
      for (int i = 0; i < numCalls; i++) {
        final int callId = i;
        CallRunner call = createMockTask(HConstants.NORMAL_QOS);
        call.setStatus(new MonitoredRPCHandlerImpl("test"));
        doAnswer(invocation -> {
          completedCalls.add(callId);
          Thread.sleep(100); // Slow processing to allow queue to build up
          return null;
        }).when(call).run();
        scheduler.dispatch(call);
        // No delay between dispatches - rapidly fill the queue
      }

      // Wait for some calls to complete
      await().atMost(2, TimeUnit.SECONDS).until(() -> completedCalls.size() >= 3);

      // Check that we had LIFO switches
      long lifoSwitches = scheduler.getNumLifoModeSwitches();
      assertTrue(lifoSwitches > 0,
        "Should have switched to LIFO mode at least once, but got: " + lifoSwitches);

      // Verify LIFO behavior: Among first completed calls, we should see higher call IDs
      // (indicating later dispatched calls completed first)
      int maxCallIdCompleted = -1;
      for (int i = 0; i < completedCalls.size(); i++) {
        maxCallIdCompleted = Math.max(maxCallIdCompleted, completedCalls.get(i));
      }
      // At least one of the early completed calls should have a high ID (>20)
      // indicating LIFO processing
      assertTrue(maxCallIdCompleted > maxCallQueueLength * coDelLifoThreshold,
        "Expected LIFO behavior: early completed calls should include call arrived after threshold "
          + "maxCallIdCompleted: " + maxCallIdCompleted);

    } finally {
      scheduler.stop();
    }
  }

  /**
   * Test that CoDel queue returns to FIFO mode after draining below threshold.
   */
  @Test
  public void testCoDelQueueDrainAndFifoReturn() throws Exception {
    Configuration testConf = HBaseConfiguration.create();
    testConf.set(RpcExecutor.CALL_QUEUE_TYPE_CONF_KEY,
      RpcExecutor.CALL_QUEUE_TYPE_CODEL_CONF_VALUE);
    testConf.setLong(RpcExecutor.CALL_QUEUE_CODEL_TARGET_DELAY, 100);
    testConf.setInt(RpcScheduler.IPC_SERVER_MAX_CALLQUEUE_LENGTH, 50);
    testConf.setDouble(RpcExecutor.CALL_QUEUE_CODEL_LIFO_THRESHOLD, 0.8);
    testConf.setInt(HConstants.REGION_SERVER_HANDLER_COUNT, 2);

    PriorityFunction priority = mock(PriorityFunction.class);
    when(priority.getPriority(any(), any(), any())).thenReturn(HConstants.NORMAL_QOS);
    SimpleRpcScheduler scheduler =
      new SimpleRpcScheduler(testConf, 2, 0, 0, priority, HConstants.QOS_THRESHOLD);

    try {
      scheduler.init(CONTEXT);
      scheduler.start();

      final List<Integer> completedCalls = Collections.synchronizedList(new ArrayList<>());

      // Fill queue rapidly to trigger LIFO (>40 calls for 80% of 50)
      for (int i = 0; i < 48; i++) {
        final int callId = i;
        CallRunner call = createMockTask(HConstants.NORMAL_QOS);
        call.setStatus(new MonitoredRPCHandlerImpl("test"));
        doAnswer(invocation -> {
          completedCalls.add(callId);
          Thread.sleep(80);
          return null;
        }).when(call).run();
        scheduler.dispatch(call);
      }

      // Wait for calls to complete
      await().atMost(1, TimeUnit.SECONDS).until(() -> completedCalls.size() >= 3);
      assertTrue(scheduler.getNumLifoModeSwitches() > 0, "Should have entered LIFO mode");

      await().atMost(2, TimeUnit.SECONDS).until(
        () -> scheduler.getGeneralQueueLength() == 0 && scheduler.getActiveRpcHandlerCount() == 0);

      long pastNumLifoModeSwitches = scheduler.getNumLifoModeSwitches();
      // Send new calls - should process in FIFO order
      completedCalls.clear();
      for (int i = 100; i < 105; i++) {
        final int callId = i;
        CallRunner call = createMockTask(HConstants.NORMAL_QOS);
        call.setStatus(new MonitoredRPCHandlerImpl("test"));
        doAnswer(invocation -> {
          completedCalls.add(callId);
          Thread.sleep(50);
          return null;
        }).when(call).run();
        scheduler.dispatch(call);
      }

      // Wait for these calls to complete
      await().atMost(2, TimeUnit.SECONDS).until(() -> completedCalls.size() >= 2);

      long newLifoSwitch = scheduler.getNumLifoModeSwitches() - pastNumLifoModeSwitches;
      // Allow at most 1 violation due to concurrent execution by 2 handlers
      assertEquals(0, newLifoSwitch,
        "Queue should not switch to LIFO last 5 calls but number of LIFO switch are : "
          + newLifoSwitch);
    } finally {
      scheduler.stop();
    }
  }

}