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/impl/HddsDispatcher.java

@@ -17,6 +17,7 @@
 
 package org.apache.hadoop.ozone.container.common.impl;
 
+import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.DISK_OUT_OF_SPACE;
 import static org.apache.hadoop.hdds.scm.protocolPB.ContainerCommandResponseBuilders.malformedRequest;
 import static org.apache.hadoop.hdds.scm.protocolPB.ContainerCommandResponseBuilders.unsupportedRequest;
 import static org.apache.hadoop.ozone.audit.AuditLogger.PerformanceStringBuilder;
@@ -336,7 +337,11 @@ && getMissingContainerSet().contains(containerID)) {
     // Small performance optimization. We check if the operation is of type
     // write before trying to send CloseContainerAction.
     if (!HddsUtils.isReadOnly(msg)) {
-      sendCloseContainerActionIfNeeded(container);
+      try {
+        sendCloseContainerActionIfNeeded(container, cmdType);
+      } catch (StorageContainerException e) {
+        return ContainerUtils.logAndReturnError(LOG, e, msg);
+      }
     }
     Handler handler = getHandler(containerType);
     if (handler == null) {
@@ -404,7 +409,11 @@ && 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);
+        try {
+          sendCloseContainerActionIfNeeded(container, cmdType);
+        } catch (StorageContainerException e) {
+          LOG.error("Exception while handling disk / space full check for containerId " + containerID, e);
+        }
       }
       if (cmdType == Type.CreateContainer
           && result == Result.SUCCESS && dispatcherContext != null) {
@@ -577,9 +586,11 @@ 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 cmdType request command type
    */
-  private void sendCloseContainerActionIfNeeded(Container container) {
+  private void sendCloseContainerActionIfNeeded(Container container, Type cmdType) throws StorageContainerException {
     // 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);
@@ -594,11 +605,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)) {
@@ -609,6 +615,17 @@ private void sendCloseContainerActionIfNeeded(Container container) {
           LOG.warn("Triggered immediate heartbeat because of full volume.");
         }
       }
+      if (isVolumeFull) {
+        HddsVolume volume = containerData.getVolume();
+        LOG.warn("Volume [{}] is full. containerID: {}. Volume usage: [{}].", volume, containerData.getContainerID(),
+            volume.getCurrentUsage());
+        if (cmdType == Type.WriteChunk || cmdType == Type.PutBlock || cmdType == Type.PutSmallFile) {
+          // If the container is full, we should not allow any more writes.
+          // This will prevent the client from writing to a full container.
+          throw new StorageContainerException("Container creation failed, due to volume out of space",
+              DISK_OUT_OF_SPACE);
+        }
+      }
     }
   }
 

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

@@ -339,8 +339,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