@Test
  public void testConnectionReplenishment() {
    AtomicInteger nextCorrelationId = new AtomicInteger(1);
    Function<Integer, List<RequestInfo>> requestGen = numRequests -> IntStream.range(0, numRequests)
        .mapToObj(
            i -> new RequestInfo(sslHost, sslPort, new MockSend(nextCorrelationId.getAndIncrement()), replicaOnSslNode,
                null))
        .collect(Collectors.toList());
    // 1 host x 1 port x 3 connections x 100%
    int warmUpPercentage = 100;
    AtomicInteger expectedConnectCalls = new AtomicInteger(warmUpPercentage * 3 / 100);
    Runnable checkConnectCalls = () -> Assert.assertEquals(expectedConnectCalls.get(), selector.connectCallCount());

    networkClient.warmUpConnections(Collections.singletonList(replicaOnSslNode.getDataNodeId()), warmUpPercentage,
        TIME_FOR_WARM_UP_MS, new ArrayList<>());
    checkConnectCalls.run();

    selector.setState(MockSelectorState.Good);
    // 1. this sendAndPoll() should use one of the pre-warmed connections
    List<ResponseInfo> responseInfoList =
        networkClient.sendAndPoll(requestGen.apply(3), Collections.emptySet(), POLL_TIMEOUT_MS);
    checkConnectCalls.run();
    Assert.assertEquals(3, responseInfoList.size());
    responseInfoList.forEach(ResponseInfo::release);

    // 2. this sendAndPoll() should disconnect two of the pre-warmed connections
    selector.setState(MockSelectorState.DisconnectOnSend);
    responseInfoList = networkClient.sendAndPoll(requestGen.apply(2), Collections.emptySet(), POLL_TIMEOUT_MS);
    checkConnectCalls.run();
    Assert.assertEquals(2, responseInfoList.size());
    responseInfoList.forEach(ResponseInfo::release);

    // 3. the two connections lost in the previous sendAndPoll should not be replenished yet since a second has not yet
    // passed since startup
    selector.setState(MockSelectorState.Good);
    responseInfoList = networkClient.sendAndPoll(requestGen.apply(1), Collections.emptySet(), POLL_TIMEOUT_MS);
    checkConnectCalls.run();
    Assert.assertEquals(1, responseInfoList.size());
    responseInfoList.forEach(ResponseInfo::release);

    // 4. one of the connection lost in sendAndPoll 3 should be replenished
    time.setCurrentMilliseconds(time.milliseconds() + Time.MsPerSec);
    selector.setState(MockSelectorState.Good);
    responseInfoList = networkClient.sendAndPoll(requestGen.apply(0), Collections.emptySet(), POLL_TIMEOUT_MS);
    expectedConnectCalls.addAndGet(1);
    checkConnectCalls.run();
    Assert.assertEquals(0, responseInfoList.size());

    // 5. no connections replenished this time since only half a second passed.
    time.setCurrentMilliseconds(time.milliseconds() + 500);
    selector.setState(MockSelectorState.Good);
    responseInfoList = networkClient.sendAndPoll(requestGen.apply(0), Collections.emptySet(), POLL_TIMEOUT_MS);
    checkConnectCalls.run();
    Assert.assertEquals(0, responseInfoList.size());

    // 6. the second connection lost in sendAndPoll 3 should be replenished
    time.setCurrentMilliseconds(time.milliseconds() + 500);
    selector.setState(MockSelectorState.Good);
    responseInfoList = networkClient.sendAndPoll(requestGen.apply(0), Collections.emptySet(), POLL_TIMEOUT_MS);
    expectedConnectCalls.addAndGet(1);
    checkConnectCalls.run();
    Assert.assertEquals(0, responseInfoList.size());

    // 7. this call should use the existing connections in the pool
    selector.setState(MockSelectorState.Good);
    responseInfoList = networkClient.sendAndPoll(requestGen.apply(3), Collections.emptySet(), POLL_TIMEOUT_MS);
    checkConnectCalls.run();
    Assert.assertEquals(3, responseInfoList.size());
    responseInfoList.forEach(ResponseInfo::release);
  }
Test that connections get replenished in {@link SocketNetworkClient#sendAndPoll(List, Set, int)} to maintain the minimum
number of active connections.
ambry-network/src/test/java/com/github/ambry/network/SocketNetworkClientTest.java