HDDS-1201. Reporting Corruptions in Containers to SCM (#912)

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

@@ -302,7 +302,7 @@ public class HddsDispatcher implements ContainerDispatcher, Auditor {
             containerState == State.OPEN || containerState == State.CLOSING);
         // mark and persist the container state to be unhealthy
         try {
-          handler.markContainerUhealthy(container);
+          handler.markContainerUnhealthy(container);
         } catch (IOException ioe) {
           // just log the error here in case marking the container fails,
           // Return the actual failure response to the client

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/Handler.java

@@ -135,7 +135,7 @@ public abstract class Handler {
    * @param container container to update
    * @throws IOException in case of exception
    */
-  public abstract void markContainerUhealthy(Container container)
+  public abstract void markContainerUnhealthy(Container container)
       throws IOException;
 
   /**

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java

@@ -884,20 +884,20 @@ public class KeyValueHandler extends Handler {
   @Override
   public void markContainerForClose(Container container)
       throws IOException {
-    State currentState = container.getContainerState();
     // Move the container to CLOSING state only if it's OPEN
-    if (currentState == State.OPEN) {
+    if (container.getContainerState() == State.OPEN) {
       container.markContainerForClose();
       sendICR(container);
     }
   }
 
   @Override
-  public void markContainerUhealthy(Container container)
+  public void markContainerUnhealthy(Container container)
       throws IOException {
-    // this will mark the container unhealthy and a close container action will
+    if (container.getContainerState() != State.UNHEALTHY) {
-    // be sent from the dispatcher ton SCM to close down this container.
       container.markContainerUnhealthy();
+      sendICR(container);
+    }
   }
 
   @Override

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/ContainerController.java

@@ -133,11 +133,11 @@ public class ContainerController {
    * @param container Container
    * @return handler of the container
    */
-  Handler getHandler(final Container container) {
+  private Handler getHandler(final Container container) {
     return handlers.get(container.getContainerType());
   }
 
-  Iterator<Container> getContainerSetIterator() {
+  public Iterator<Container> getContainers() {
     return containerSet.getContainerIterator();
   }
 }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/ContainerScrubber.java

@@ -22,7 +22,6 @@ import org.apache.commons.net.ntp.TimeStamp;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
 import org.apache.hadoop.ozone.container.common.interfaces.Container;
-import org.apache.hadoop.ozone.container.common.interfaces.Handler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -57,7 +56,11 @@ public class ContainerScrubber implements Runnable {
     LOG.info("Background ContainerScrubber starting up");
     while (true) {
 
+      try {
         scrub();
+      } catch (StorageContainerException e) {
+        LOG.error("Scrubber encountered StorageContainerException.");
+      }
 
       if (this.halt) {
         break; // stop and exit if requested
@@ -126,25 +129,20 @@ public class ContainerScrubber implements Runnable {
     }
   }
 
-  private void scrub() {
+  private void scrub() throws StorageContainerException {
-
+    Iterator<Container> containerIt = controller.getContainers();
-    Iterator<Container> containerIt = controller.getContainerSetIterator();
     long count = 0;
 
-    while (containerIt.hasNext()) {
+    while (containerIt.hasNext() && !halt) {
       TimeStamp startTime = new TimeStamp(System.currentTimeMillis());
       Container container = containerIt.next();
-      Handler containerHandler = controller.getHandler(container);
-
-      if (this.halt) {
-        break; // stop if requested
-      }
 
       try {
         container.check();
       } catch (StorageContainerException e) {
         LOG.error("Error unexpected exception {} for Container {}", e,
             container.getContainerData().getContainerID());
+        container.markContainerUnhealthy();
         // XXX Action required here
       }
       count++;