void handleTimeout(RaftClientRequest request, PendingWatch pending) {
  if (removeExisting(pending)) {
    pending.getFuture().completeExceptionally(
        new NotReplicatedException(request.getCallId(), replication, pending.getIndex()));
    LOG.debug("{}: timeout {}, {}", name, pending, request);
  }
}
 

static void assertNotReplicatedException(long logIndex, ReplicationLevel replication, Throwable t) {
  Assert.assertSame(NotReplicatedException.class, t.getClass());
  final NotReplicatedException nre = (NotReplicatedException) t;
  Assert.assertNotNull(nre);
  Assert.assertEquals(logIndex, nre.getLogIndex());
  Assert.assertEquals(replication, nre.getRequiredReplication());
}
 

void handleTimeout(RaftClientRequest request, PendingWatch pending) {
  if (removeExisting(pending)) {
    pending.getFuture().completeExceptionally(
        new NotReplicatedException(request.getCallId(), replication, pending.getIndex()));
    LOG.debug("{}: timeout {}, {}", name, pending, request);
  }
}
 

static void assertNotReplicatedException(long logIndex, ReplicationLevel replication, RaftClientReply reply) {
  Assert.assertFalse(reply.isSuccess());
  final NotReplicatedException nre = reply.getNotReplicatedException();
  Assert.assertNotNull(nre);
  Assert.assertEquals(logIndex, nre.getLogIndex());
  Assert.assertEquals(replication, nre.getRequiredReplication());
}
 

@Test
public void testReleaseBuffersOnException() throws Exception {
  int capacity = 2;
  BufferPool bufferPool = new BufferPool(chunkSize, capacity);
  XceiverClientManager clientManager = new XceiverClientManager(conf);
  ContainerWithPipeline container = storageContainerLocationClient
      .allocateContainer(HddsProtos.ReplicationType.RATIS,
          HddsProtos.ReplicationFactor.THREE, OzoneConsts.OZONE);
  Pipeline pipeline = container.getPipeline();
  long containerId = container.getContainerInfo().getContainerID();
  XceiverClientSpi xceiverClient = clientManager.acquireClient(pipeline);
  Assert.assertEquals(1, xceiverClient.getRefcount());
  Assert.assertTrue(xceiverClient instanceof XceiverClientRatis);
  XceiverClientRatis ratisClient = (XceiverClientRatis) xceiverClient;
  CommitWatcher watcher = new CommitWatcher(bufferPool, ratisClient);
  BlockID blockID = ContainerTestHelper.getTestBlockID(containerId);
  List<XceiverClientReply> replies = new ArrayList<>();
  long length = 0;
  List<CompletableFuture<ContainerProtos.ContainerCommandResponseProto>>
      futures = new ArrayList<>();
  for (int i = 0; i < capacity; i++) {
    ContainerProtos.ContainerCommandRequestProto writeChunkRequest =
        ContainerTestHelper
            .getWriteChunkRequest(pipeline, blockID, chunkSize, null);
    // add the data to the buffer pool
    final ChunkBuffer byteBuffer = bufferPool.allocateBufferIfNeeded(0);
    byteBuffer.put(writeChunkRequest.getWriteChunk().getData());
    ratisClient.sendCommandAsync(writeChunkRequest);
    ContainerProtos.ContainerCommandRequestProto putBlockRequest =
        ContainerTestHelper
            .getPutBlockRequest(pipeline, writeChunkRequest.getWriteChunk());
    XceiverClientReply reply = ratisClient.sendCommandAsync(putBlockRequest);
    final List<ChunkBuffer> bufferList = singletonList(byteBuffer);
    length += byteBuffer.position();
    CompletableFuture<ContainerProtos.ContainerCommandResponseProto> future =
        reply.getResponse().thenApply(v -> {
          watcher.updateCommitInfoMap(reply.getLogIndex(), bufferList);
          return v;
        });
    futures.add(future);
    watcher.getFutureMap().put(length, future);
    replies.add(reply);
  }

  Assert.assertTrue(replies.size() == 2);
  // wait on the 1st putBlock to complete
  CompletableFuture<ContainerProtos.ContainerCommandResponseProto> future1 =
      futures.get(0);
  CompletableFuture<ContainerProtos.ContainerCommandResponseProto> future2 =
      futures.get(1);
  future1.get();
  Assert.assertNotNull(watcher.getFutureMap().get(new Long(chunkSize)));
  Assert.assertTrue(
      watcher.getFutureMap().get(new Long(chunkSize)).equals(future1));
  // wait on 2nd putBlock to complete
  future2.get();
  Assert.assertNotNull(watcher.getFutureMap().get(new Long(2 * chunkSize)));
  Assert.assertTrue(
      watcher.getFutureMap().get(new Long(2 * chunkSize)).equals(future2));
  Assert.assertTrue(watcher.getCommitIndex2flushedDataMap().size() == 2);
  watcher.watchOnFirstIndex();
  Assert.assertFalse(watcher.getCommitIndex2flushedDataMap()
      .containsKey(replies.get(0).getLogIndex()));
  Assert.assertFalse(watcher.getFutureMap().containsKey(chunkSize));
  Assert.assertTrue(watcher.getTotalAckDataLength() >= chunkSize);
  cluster.shutdownHddsDatanode(pipeline.getNodes().get(0));
  cluster.shutdownHddsDatanode(pipeline.getNodes().get(1));
  try {
    // just watch for a higher index so as to ensure, it does an actual
    // call to Ratis. Otherwise, it may just return in case the commitInfoMap
    // is updated to the latest index in putBlock response.
    watcher.watchForCommit(replies.get(1).getLogIndex() + 100);
    Assert.fail("Expected exception not thrown");
  } catch(IOException ioe) {
    // with retry count set to lower limit and a lower watch request
    // timeout, watch request will eventually
    // fail with RaftRetryFailure exception from ratis client or the client
    // can itself get AlreadyClosedException from the Ratis Server
    Throwable t = HddsClientUtils.checkForException(ioe);
    Assert.assertTrue(t instanceof RaftRetryFailureException ||
            t instanceof AlreadyClosedException ||
            t instanceof NotReplicatedException);
  }
  if (ratisClient.getReplicatedMinCommitIndex() < replies.get(1)
      .getLogIndex()) {
    Assert.assertTrue(watcher.getTotalAckDataLength() == chunkSize);
    Assert.assertTrue(watcher.getCommitIndex2flushedDataMap().size() == 1);
    Assert.assertTrue(watcher.getFutureMap().size() == 1);
  } else {
    Assert.assertTrue(watcher.getTotalAckDataLength() == 2 * chunkSize);
    Assert.assertTrue(watcher.getFutureMap().isEmpty());
    Assert.assertTrue(watcher.getCommitIndex2flushedDataMap().isEmpty());
  }
}