HDDS-12151. Fail write when volume is full considering min free space

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ContainerUtils.java

@@ -352,7 +352,7 @@ public static void assertSpaceAvailability(long containerId, HddsVolume volume,
 
     if (currentUsage.getAvailable() - spared < sizeRequested) {
       throw new StorageContainerException("Failed to write " + sizeRequested + " bytes to container "
-          + containerId + " due to volume " + volume.getStorageID() + " out of space "
+          + containerId + " due to volume " + volume + " out of space "
           + currentUsage + ", minimum free space spared="  + spared, DISK_OUT_OF_SPACE);
     }
   }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/HddsDispatcher.java

@@ -69,7 +69,6 @@
 import org.apache.hadoop.ozone.container.common.interfaces.Handler;
 import org.apache.hadoop.ozone.container.common.statemachine.StateContext;
 import org.apache.hadoop.ozone.container.common.transport.server.ratis.DispatcherContext;
-import org.apache.hadoop.ozone.container.common.volume.HddsVolume;
 import org.apache.hadoop.ozone.container.common.volume.VolumeSet;
 import org.apache.hadoop.ozone.container.ozoneimpl.OnDemandContainerDataScanner;
 import org.apache.hadoop.util.Time;
@@ -335,8 +334,22 @@ && getMissingContainerSet().contains(containerID)) {
     }
     // Small performance optimization. We check if the operation is of type
     // write before trying to send CloseContainerAction.
+    boolean isVolumeFullForWrite = false;
     if (!HddsUtils.isReadOnly(msg)) {
-      sendCloseContainerActionIfNeeded(container);
+      try {
+        if (container != null && container.getContainerState() == State.OPEN) {
+          ContainerUtils.assertSpaceAvailability(containerID, container.getContainerData().getVolume(), 0);
+        }
+      } catch (StorageContainerException e) {
+        LOG.warn(e.getMessage());
+        isVolumeFullForWrite = true;
+        if (cmdType == Type.WriteChunk || cmdType == Type.PutBlock || cmdType == Type.PutSmallFile) {
+          audit(action, eventType, msg, dispatcherContext, AuditEventStatus.FAILURE, e);
+          return ContainerUtils.logAndReturnError(LOG, e, msg);
+        }
+      } finally {
+        sendCloseContainerActionIfNeeded(container, isVolumeFullForWrite);
+      }
     }
     Handler handler = getHandler(containerType);
     if (handler == null) {
@@ -404,7 +417,7 @@ && getMissingContainerSet().contains(containerID)) {
         // in any case, the in memory state of the container should be unhealthy
         Preconditions.checkArgument(
             container.getContainerData().getState() == State.UNHEALTHY);
-        sendCloseContainerActionIfNeeded(container);
+        sendCloseContainerActionIfNeeded(container, isVolumeFullForWrite);
       }
       if (cmdType == Type.CreateContainer
           && result == Result.SUCCESS && dispatcherContext != null) {
@@ -577,12 +590,12 @@ public void validateContainerCommand(
   /**
    * If the container usage reaches the close threshold or the container is
    * marked unhealthy we send Close ContainerAction to SCM.
-   * @param container current state of container
+   *
+   * @param container    current state of container
+   * @param isVolumeFull volume full flag for open containers
    */
-  private void sendCloseContainerActionIfNeeded(Container container) {
+  private void sendCloseContainerActionIfNeeded(Container container, boolean isVolumeFull) {
     // We have to find a more efficient way to close a container.
-    boolean isOpen = container != null && container.getContainerState() == State.OPEN;
-    boolean isVolumeFull = isOpen && isVolumeFullExcludingCommittedSpace(container);
     boolean isSpaceFull = isVolumeFull || isContainerFull(container);
     boolean shouldClose = isSpaceFull || isContainerUnhealthy(container);
     if (shouldClose) {
@@ -594,11 +607,6 @@ private void sendCloseContainerActionIfNeeded(Container container) {
           .setContainerID(containerData.getContainerID())
           .setAction(ContainerAction.Action.CLOSE).setReason(reason).build();
       context.addContainerActionIfAbsent(action);
-      if (isVolumeFull) {
-        HddsVolume volume = containerData.getVolume();
-        LOG.warn("Volume [{}] is full. containerID: {}. Volume usage: [{}].", volume, containerData.getContainerID(),
-            volume.getCurrentUsage());
-      }
       AtomicBoolean immediateCloseActionSent = containerData.getImmediateCloseActionSent();
       // if an immediate heartbeat has not been triggered already, trigger it now
       if (immediateCloseActionSent.compareAndSet(false, true)) {
@@ -626,10 +634,6 @@ private boolean isContainerFull(Container container) {
     }
   }
 
-  private boolean isVolumeFullExcludingCommittedSpace(Container container) {
-    return container.getContainerData().getVolume().isVolumeFull();
-  }
-
   private boolean isContainerUnhealthy(Container container) {
     return Optional.ofNullable(container).map(
             cont -> (cont.getContainerState() ==

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/volume/HddsVolume.java

@@ -198,12 +198,6 @@ public VolumeInfoMetrics getVolumeInfoStats() {
     return volumeInfoMetrics;
   }
 
-  public boolean isVolumeFull() {
-    SpaceUsageSource currentUsage = getCurrentUsage();
-    // if the volume is failed, this method will implicitly return true because available space will be 0
-    return currentUsage.getAvailable() - getFreeSpaceToSpare(currentUsage.getCapacity()) <= 0;
-  }
-
   @Override
   protected StorageLocationReport.Builder reportBuilder() {
     StorageLocationReport.Builder builder = super.reportBuilder();

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestHddsDispatcher.java

@@ -342,8 +342,7 @@ public void testContainerCloseActionWhenVolumeFull(
       usedSpace.addAndGet(60);
       ContainerCommandResponseProto response = hddsDispatcher
           .dispatch(getWriteChunkRequest(dd.getUuidString(), 1L, 1L), null);
-      assertEquals(ContainerProtos.Result.SUCCESS,
-          response.getResult());
+      assertEquals(ContainerProtos.Result.DISK_OUT_OF_SPACE, response.getResult());
       verify(context, times(1))
           .addContainerActionIfAbsent(any(ContainerAction.class));
       // verify that immediate heartbeat is triggered

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/impl/TestKeyValueStreamDataChannel.java

@@ -165,7 +165,6 @@ public void testVolumeFullCase() throws Exception {
     tempFile.deleteOnExit();
     HddsVolume mockVolume = mock(HddsVolume.class);
     when(mockVolume.getStorageID()).thenReturn("storageId");
-    when(mockVolume.isVolumeFull()).thenReturn(true);
     when(mockVolume.getCurrentUsage()).thenReturn(new SpaceUsageSource.Fixed(100L, 0L, 100L));
     ContainerData mockContainerData = mock(ContainerData.class);
     when(mockContainerData.getContainerID()).thenReturn(123L);