@Test
  public void ensureCapacityFailureTest() throws Exception {
    int numSegments = (int) (LOG_CAPACITY / SEGMENT_CAPACITY);
    LogSegment segment = getLogSegment(LogSegmentName.fromPositionAndGeneration(0, 0), SEGMENT_CAPACITY, true);
    LogSegmentName lastName = LogSegmentName.fromPositionAndGeneration(0, 0);
    List<Pair<LogSegmentName, String>> segmentNameAndFileNamesDesired = new ArrayList<>();
    for (int i = 0; i < numSegments; i++) {
      lastName = lastName.getNextPositionName();
      String fileName = lastName.toFilename();
      segmentNameAndFileNamesDesired.add(new Pair<>(lastName, fileName));
    }

    File file = new File("1_0_log");
    File mockFile = Mockito.spy(file);
    when(mockFile.exists()).thenReturn(false);
    DiskSpaceAllocator mockDiskAllocator = Mockito.spy(StoreTestUtils.DEFAULT_DISK_SPACE_ALLOCATOR);
    Log log = new Log(tempDir.getAbsolutePath(), LOG_CAPACITY, mockDiskAllocator,
        createStoreConfig(SEGMENT_CAPACITY, setFilePermissionEnabled), metrics, true,
        Collections.singletonList(segment), segmentNameAndFileNamesDesired.iterator(), null);
    Log mockLog = Mockito.spy(log);
    when(mockLog.allocate(anyString(), anyLong(), anyBoolean())).thenReturn(mockFile);
    long initialUnallocatedSegments = mockLog.getRemainingUnallocatedSegments();

    // write enough so that all segments are allocated
    ByteBuffer buffer = ByteBuffer.allocate((int) (segment.getCapacityInBytes() - segment.getStartOffset()));
    // write first segment
    buffer.rewind();
    CHANNEL_APPENDER.append(mockLog, buffer);

    // Test 1: try to write second segment triggering roll over and calls ensureCapacity()
    try {
      buffer.rewind();
      CHANNEL_APPENDER.append(mockLog, buffer);
      fail("Should fail because log segment instantiation encounters FileNotFound exception");
    } catch (StoreException e) {
      assertEquals("Mismatch in store error code", StoreErrorCodes.UnknownError, e.getErrorCode());
      assertEquals(
          "The number of unallocated segments should decrease because exception occurred when freeing the segment",
          initialUnallocatedSegments - 1, mockLog.getRemainingUnallocatedSegments());
    }

    // Test 2: the segment with exception is freed successfully and remainingUnallocatedSegments counter is restored.
    doAnswer((Answer) invocation -> null).when(mockDiskAllocator)
        .free(any(File.class), any(Long.class), anyString(), anyBoolean());
    try {
      buffer.rewind();
      CHANNEL_APPENDER.append(mockLog, buffer);
      fail("Should fail because log segment instantiation encounters FileNotFound exception");
    } catch (StoreException e) {
      assertEquals("Mismatch in store error code", StoreErrorCodes.FileNotFound, e.getErrorCode());
      // Note that the number should be initialUnallocatedSegments - 1 because in previous test the segment with exception
      // didn't get freed due to exception in diskSpaceAllocator.free().
      assertEquals(
          "The number of unallocated segments should stay unchanged because exception segment is successfully freed",
          initialUnallocatedSegments - 1, mockLog.getRemainingUnallocatedSegments());
    }
  }
Test several failure scenarios where exception occurs when instantiating new log segment in ensureCapacity() method.
The method should catch the exception, free log segment and restore the counter.
@throws Exception
ambry-store/src/test/java/com/github/ambry/store/LogTest.java