Revert "HBASE-16092 Procedure v2 - complete child procedure support"

This reverts commit 96c4054.
apache · Jun 24, 2016 · 1d06850 · 1d06850
1 parent e8599a2
commit 1d06850
Show file tree

Hide file tree

Showing 15 changed files with 78 additions and 678 deletions.
diff --git a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/Procedure.java b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/Procedure.java
@@ -589,11 +589,6 @@ protected synchronized boolean childrenCountDown() {
     return --childrenLatch == 0;
   }
 
-  @InterfaceAudience.Private
-  protected synchronized boolean hasChildren() {
-    return childrenLatch > 0;
-  }
-
   /**
    * Called by the RootProcedureState on procedure execution.
    * Each procedure store its stack-index positions.
@@ -611,7 +606,7 @@ protected synchronized void addStackIndex(final int index) {
 
   @InterfaceAudience.Private
   protected synchronized boolean removeStackIndex() {
-    if (stackIndexes != null && stackIndexes.length > 1) {
+    if (stackIndexes.length > 1) {
       stackIndexes = Arrays.copyOf(stackIndexes, stackIndexes.length - 1);
       return false;
     } else {

diff --git a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/ProcedureExecutor.java b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/ProcedureExecutor.java
@@ -18,18 +18,16 @@
 
 package org.apache.hadoop.hbase.procedure2;
 
-import com.google.common.base.Preconditions;
-
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
-import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.HashSet;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
@@ -57,6 +55,8 @@
 import org.apache.hadoop.hbase.util.NonceKey;
 import org.apache.hadoop.hbase.util.Pair;
 
+import com.google.common.base.Preconditions;
+
 /**
  * Thread Pool that executes the submitted procedures.
  * The executor has a ProcedureStore associated.
@@ -314,7 +314,7 @@ public void handleCorrupted(ProcedureIterator procIter) throws IOException {
           corruptedCount++;
         }
         if (abortOnCorruption && corruptedCount > 0) {
-          throw new IOException("found " + corruptedCount + " corrupted procedure(s) on replay");
+          throw new IOException("found " + corruptedCount + " procedures on replay");
         }
       }
     });
@@ -388,10 +388,10 @@ private void loadProcedures(final ProcedureIterator procIter,
         continue;
       }
 
-      if (proc.hasParent()) {
+      if (proc.hasParent() && !proc.isFinished()) {
         Procedure parent = procedures.get(proc.getParentProcId());
         // corrupted procedures are handled later at step 3
-        if (parent != null && !proc.isFinished()) {
+        if (parent != null) {
           parent.incChildrenLatch();
         }
       }
@@ -403,11 +403,6 @@ private void loadProcedures(final ProcedureIterator procIter,
         case RUNNABLE:
           runnableList.add(proc);
           break;
-        case WAITING:
-          if (!proc.hasChildren()) {
-            runnableList.add(proc);
-          }
-          break;
         case WAITING_TIMEOUT:
           if (waitingSet == null) {
             waitingSet = new HashSet<Procedure>();
@@ -418,8 +413,8 @@ private void loadProcedures(final ProcedureIterator procIter,
           if (proc.hasException()) {
             // add the proc to the runnables to perform the rollback
             runnables.addBack(proc);
+            break;
           }
-          break;
         case ROLLEDBACK:
         case INITIALIZING:
           String msg = "Unexpected " + proc.getState() + " state for " + proc;
@@ -438,7 +433,7 @@ private void loadProcedures(final ProcedureIterator procIter,
       RootProcedureState procStack = entry.getValue();
       if (procStack.isValid()) continue;
 
-      for (Procedure proc: procStack.getSubproceduresStack()) {
+      for (Procedure proc: procStack.getSubprocedures()) {
         LOG.error("corrupted procedure: " + proc);
         procedures.remove(proc.getProcId());
         runnableList.remove(proc);
@@ -945,7 +940,7 @@ private boolean executeRollback(final long rootProcId, final RootProcedureState
       store.update(rootProc);
     }
 
-    List<Procedure> subprocStack = procStack.getSubproceduresStack();
+    List<Procedure> subprocStack = procStack.getSubprocedures();
     assert subprocStack != null : "Called rollback with no steps executed rootProc=" + rootProc;
 
     int stackTail = subprocStack.size();
@@ -1027,12 +1022,7 @@ private boolean executeRollback(final Procedure proc) {
         store.delete(proc.getProcId());
         procedures.remove(proc.getProcId());
       } else {
-        final long[] childProcIds = rollbackStack.get(proc.getProcId()).getSubprocedureIds();
+        store.update(proc);
-        if (childProcIds != null) {
-          store.delete(proc, childProcIds);
-        } else {
-          store.update(proc);
-        }
       }
     } else {
       store.update(proc);
@@ -1112,7 +1102,6 @@ private void execProcedure(final RootProcedureState procStack, final Procedure p
               assert subproc.getState() == ProcedureState.INITIALIZING : subproc;
               subproc.setParentProcId(procedure.getProcId());
               subproc.setProcId(nextProcId());
-              procStack.addSubProcedure(subproc);
             }
 
             if (!procedure.isFailed()) {
@@ -1149,7 +1138,17 @@ private void execProcedure(final RootProcedureState procStack, final Procedure p
       }
 
       // Commit the transaction
-      updateStoreOnExec(procStack, procedure, subprocs);
+      if (subprocs != null && !procedure.isFailed()) {
+        if (LOG.isTraceEnabled()) {
+          LOG.trace("Store add " + procedure + " children " + Arrays.toString(subprocs));
+        }
+        store.insert(procedure, subprocs);
+      } else {
+        if (LOG.isTraceEnabled()) {
+          LOG.trace("Store update " + procedure);
+        }
+        store.update(procedure);
+      }
 
       // if the store is not running we are aborting
       if (!store.isRunning()) {
@@ -1199,34 +1198,6 @@ private void execProcedure(final RootProcedureState procStack, final Procedure p
     }
   }
 
-  private void updateStoreOnExec(final RootProcedureState procStack,
-      final Procedure procedure, final Procedure[] subprocs) {
-    if (subprocs != null && !procedure.isFailed()) {
-      if (LOG.isTraceEnabled()) {
-        LOG.trace("Store add " + procedure + " children " + Arrays.toString(subprocs));
-      }
-      store.insert(procedure, subprocs);
-    } else {
-      if (LOG.isTraceEnabled()) {
-        LOG.trace("Store update " + procedure);
-      }
-      if (procedure.isFinished() && !procedure.hasParent()) {
-        // remove child procedures
-        final long[] childProcIds = procStack.getSubprocedureIds();
-        if (childProcIds != null) {
-          store.delete(procedure, childProcIds);
-          for (int i = 0; i < childProcIds.length; ++i) {
-            procedures.remove(childProcIds[i]);
-          }
-        } else {
-          store.update(procedure);
-        }
-      } else {
-        store.update(procedure);
-      }
-    }
-  }
-
   private void handleInterruptedException(final Procedure proc, final InterruptedException e) {
     if (LOG.isTraceEnabled()) {
       LOG.trace("got an interrupt during " + proc + ". suspend and retry it later.", e);

diff --git a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/RootProcedureState.java b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/RootProcedureState.java
@@ -19,9 +19,7 @@
 package org.apache.hadoop.hbase.procedure2;
 
 import java.util.ArrayList;
-import java.util.HashSet;
 import java.util.List;
-import java.util.Set;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -51,8 +49,7 @@ private enum State {
     ROLLINGBACK,     // The Procedure failed and the execution was rolledback
   }
 
-  private Set<Procedure> subprocs = null;
+  private ArrayList<Procedure> subprocedures = null;
-  private ArrayList<Procedure> subprocStack = null;
   private State state = State.RUNNING;
   private int running = 0;
 
@@ -90,23 +87,13 @@ protected synchronized void unsetRollback() {
     state = State.FAILED;
   }
 
-  protected synchronized long[] getSubprocedureIds() {
+  protected synchronized List<Procedure> getSubprocedures() {
-    if (subprocs == null) return null;
+    return subprocedures;
-    int index = 0;
-    final long[] subIds = new long[subprocs.size()];
-    for (Procedure proc: subprocs) {
-      subIds[index++] = proc.getProcId();
-    }
-    return subIds;
-  }
-
-  protected synchronized List<Procedure> getSubproceduresStack() {
-    return subprocStack;
   }
 
   protected synchronized RemoteProcedureException getException() {
-    if (subprocStack != null) {
+    if (subprocedures != null) {
-      for (Procedure proc: subprocStack) {
+      for (Procedure proc: subprocedures) {
         if (proc.hasException()) {
           return proc.getException();
         }
@@ -146,19 +133,11 @@ protected synchronized void addRollbackStep(final Procedure proc) {
     if (proc.isFailed()) {
       state = State.FAILED;
     }
-    if (subprocStack == null) {
+    if (subprocedures == null) {
-      subprocStack = new ArrayList<Procedure>();
+      subprocedures = new ArrayList<Procedure>();
-    }
-    proc.addStackIndex(subprocStack.size());
-    subprocStack.add(proc);
-  }
-
-  protected synchronized void addSubProcedure(final Procedure proc) {
-    if (!proc.hasParent()) return;
-    if (subprocs == null) {
-      subprocs = new HashSet<Procedure>();
     }
-    subprocs.add(proc);
+    proc.addStackIndex(subprocedures.size());
+    subprocedures.add(proc);
   }
 
   /**
@@ -169,19 +148,18 @@ protected synchronized void addSubProcedure(final Procedure proc) {
    * on load we recreate the full stack by aggregating each procedure stack-positions.
    */
   protected synchronized void loadStack(final Procedure proc) {
-    addSubProcedure(proc);
     int[] stackIndexes = proc.getStackIndexes();
     if (stackIndexes != null) {
-      if (subprocStack == null) {
+      if (subprocedures == null) {
-        subprocStack = new ArrayList<Procedure>();
+        subprocedures = new ArrayList<Procedure>();
       }
-      int diff = (1 + stackIndexes[stackIndexes.length - 1]) - subprocStack.size();
+      int diff = (1 + stackIndexes[stackIndexes.length - 1]) - subprocedures.size();
       if (diff > 0) {
-        subprocStack.ensureCapacity(1 + stackIndexes[stackIndexes.length - 1]);
+        subprocedures.ensureCapacity(1 + stackIndexes[stackIndexes.length - 1]);
-        while (diff-- > 0) subprocStack.add(null);
+        while (diff-- > 0) subprocedures.add(null);
       }
       for (int i = 0; i < stackIndexes.length; ++i) {
-        subprocStack.set(stackIndexes[i], proc);
+        subprocedures.set(stackIndexes[i], proc);
       }
     }
     if (proc.getState() == ProcedureState.ROLLEDBACK) {
@@ -195,8 +173,8 @@ protected synchronized void loadStack(final Procedure proc) {
    * Called on store load by the ProcedureExecutor to validate the procedure stack.
    */
   protected synchronized boolean isValid() {
-    if (subprocStack != null) {
+    if (subprocedures != null) {
-      for (Procedure proc: subprocStack) {
+      for (Procedure proc: subprocedures) {
         if (proc == null) {
           return false;
         }

diff --git a/...-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/NoopProcedureStore.java b/...-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/NoopProcedureStore.java
@@ -70,9 +70,4 @@ public void update(Procedure proc) {
   public void delete(long procId) {
     // no-op
   }
-
-  @Override
-  public void delete(Procedure proc, long[] subprocs) {
-    // no-op
-  }
 }
diff --git a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/ProcedureStore.java b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/ProcedureStore.java
@@ -20,9 +20,9 @@
 
 import java.io.IOException;
 
-import org.apache.hadoop.hbase.ProcedureInfo;
 import org.apache.hadoop.hbase.classification.InterfaceAudience;
 import org.apache.hadoop.hbase.classification.InterfaceStability;
+import org.apache.hadoop.hbase.ProcedureInfo;
 import org.apache.hadoop.hbase.procedure2.Procedure;
 
 /**
@@ -188,12 +188,4 @@ public interface ProcedureLoader {
    * @param procId the ID of the procedure to remove.
    */
   void delete(long procId);
-
-  /**
-   * The parent procedure completed.
-   * Update the state and mark all the child deleted.
-   * @param parentProc the parent procedure to serialize and write to the store.
-   * @param subProcIds the IDs of the sub-procedure to remove.
-   */
-  void delete(Procedure parentProc, long[] subProcIds);
 }
diff --git a/...ocedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/ProcedureStoreTracker.java b/...ocedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/ProcedureStoreTracker.java
@@ -21,7 +21,6 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
-import java.util.Arrays;
 import java.util.Iterator;
 import java.util.Map;
 import java.util.TreeMap;
@@ -395,14 +394,6 @@ public void delete(long procId) {
     trackProcIds(procId);
   }
 
-  public void delete(long[] procIds) {
-    // TODO: optimize
-    Arrays.sort(procIds);
-    for (int i = 0; i < procIds.length; ++i) {
-      delete(procIds[i]);
-    }
-  }
-
   private void trackProcIds(long procId) {
     minUpdatedProcId = Math.min(minUpdatedProcId, procId);
     maxUpdatedProcId = Math.max(maxUpdatedProcId, procId);

diff --git a/...cedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/wal/ProcedureWALFormat.java b/...cedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/wal/ProcedureWALFormat.java
@@ -18,8 +18,6 @@
 
 package org.apache.hadoop.hbase.procedure2.store.wal;
 
-import com.google.protobuf.InvalidProtocolBufferException;
-
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
@@ -38,6 +36,8 @@
 import org.apache.hadoop.hbase.protobuf.generated.ProcedureProtos.ProcedureWALHeader;
 import org.apache.hadoop.hbase.protobuf.generated.ProcedureProtos.ProcedureWALTrailer;
 
+import com.google.protobuf.InvalidProtocolBufferException;
+
 /**
  * Helper class that contains the WAL serialization utils.
  */
@@ -231,18 +231,4 @@ public static void writeDelete(ByteSlot slot, long procId)
     builder.setProcId(procId);
     builder.build().writeDelimitedTo(slot);
   }
-
-  public static void writeDelete(ByteSlot slot, Procedure proc, long[] subprocs)
-      throws IOException {
-    ProcedureWALEntry.Builder builder = ProcedureWALEntry.newBuilder();
-    builder.setType(ProcedureWALEntry.Type.PROCEDURE_WAL_DELETE);
-    builder.setProcId(proc.getProcId());
-    if (subprocs != null) {
-      builder.addProcedure(Procedure.convert(proc));
-      for (int i = 0; i < subprocs.length; ++i) {
-        builder.addChildId(subprocs[i]);
-      }
-    }
-    builder.build().writeDelimitedTo(slot);
-  }
 }