hbase-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From mberto...@apache.org
Subject [6/8] hbase git commit: HBASE-13202 Procedure v2 - core framework
Date Tue, 24 Mar 2015 17:00:39 GMT
http://git-wip-us.apache.org/repos/asf/hbase/blob/e3db688a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/wal/WALProcedureStore.java
----------------------------------------------------------------------
diff --git a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/wal/WALProcedureStore.java b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/wal/WALProcedureStore.java
new file mode 100644
index 0000000..c2a5571
--- /dev/null
+++ b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/wal/WALProcedureStore.java
@@ -0,0 +1,683 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase.procedure2.store.wal;
+
+import java.io.IOException;
+import java.io.FileNotFoundException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+import java.util.Arrays;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.Set;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileAlreadyExistsException;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathFilter;
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.classification.InterfaceStability;
+import org.apache.hadoop.hbase.procedure2.Procedure;
+import org.apache.hadoop.hbase.procedure2.store.ProcedureStore;
+import org.apache.hadoop.hbase.procedure2.store.ProcedureStoreTracker;
+import org.apache.hadoop.hbase.procedure2.util.ByteSlot;
+import org.apache.hadoop.hbase.procedure2.util.StringUtils;
+import org.apache.hadoop.hbase.protobuf.generated.ProcedureProtos.ProcedureWALHeader;
+
+/**
+ * WAL implementation of the ProcedureStore.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public class WALProcedureStore implements ProcedureStore {
+  private static final Log LOG = LogFactory.getLog(WALProcedureStore.class);
+
+  public interface LeaseRecovery {
+    void recoverFileLease(FileSystem fs, Path path) throws IOException;
+  }
+
+  private static final int MAX_RETRIES_BEFORE_ABORT = 3;
+
+  private static final String SYNC_WAIT_MSEC_CONF_KEY = "hbase.procedure.store.wal.sync.wait.msec";
+  private static final int DEFAULT_SYNC_WAIT_MSEC = 50;
+
+  private final CopyOnWriteArrayList<ProcedureStoreListener> listeners =
+    new CopyOnWriteArrayList<ProcedureStoreListener>();
+
+  private final LinkedList<ProcedureWALFile> logs = new LinkedList<ProcedureWALFile>();
+  private final ProcedureStoreTracker storeTracker = new ProcedureStoreTracker();
+  private final AtomicBoolean running = new AtomicBoolean(false);
+  private final ReentrantLock lock = new ReentrantLock();
+  private final Condition waitCond = lock.newCondition();
+  private final Condition slotCond = lock.newCondition();
+  private final Condition syncCond = lock.newCondition();
+
+  private final LeaseRecovery leaseRecovery;
+  private final Configuration conf;
+  private final FileSystem fs;
+  private final Path logDir;
+
+  private AtomicBoolean inSync = new AtomicBoolean(false);
+  private ArrayBlockingQueue<ByteSlot> slotsCache = null;
+  private Set<ProcedureWALFile> corruptedLogs = null;
+  private FSDataOutputStream stream = null;
+  private long totalSynced = 0;
+  private long flushLogId = 0;
+  private int slotIndex = 0;
+  private Thread syncThread;
+  private ByteSlot[] slots;
+  private int syncWaitMsec;
+
+  public WALProcedureStore(final Configuration conf, final FileSystem fs, final Path logDir,
+      final LeaseRecovery leaseRecovery) {
+    this.fs = fs;
+    this.conf = conf;
+    this.logDir = logDir;
+    this.leaseRecovery = leaseRecovery;
+  }
+
+  @Override
+  public void start(int numSlots) throws IOException {
+    if (running.getAndSet(true)) {
+      return;
+    }
+
+    // Init buffer slots
+    slots = new ByteSlot[numSlots];
+    slotsCache = new ArrayBlockingQueue(numSlots, true);
+    while (slotsCache.remainingCapacity() > 0) {
+      slotsCache.offer(new ByteSlot());
+    }
+
+    // Tunings
+    syncWaitMsec = conf.getInt(SYNC_WAIT_MSEC_CONF_KEY, DEFAULT_SYNC_WAIT_MSEC);
+
+    // Init sync thread
+    syncThread = new Thread() {
+      @Override
+      public void run() {
+        while (running.get()) {
+          try {
+            syncLoop();
+          } catch (IOException e) {
+            LOG.error("got an exception from the sync-loop", e);
+            sendAbortProcessSignal();
+          }
+        }
+      }
+    };
+    syncThread.start();
+  }
+
+  @Override
+  public void stop() {
+    if (!running.getAndSet(false)) {
+      return;
+    }
+
+    LOG.info("Stopping the WAL Procedure Store");
+
+    lock.lock();
+    try {
+      waitCond.signalAll();
+    } finally {
+      lock.unlock();
+    }
+
+    try {
+      syncThread.join();
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+    }
+
+    // Close the writer
+    closeStream();
+
+    // Close the old logs
+    // they should be already closed, this is just in case the load fails
+    // and we call start() and then stop()
+    for (ProcedureWALFile log: logs) {
+      log.close();
+    }
+    logs.clear();
+  }
+
+  @Override
+  public void registerListener(ProcedureStoreListener listener) {
+    this.listeners.add(listener);
+  }
+
+  @Override
+  public boolean unregisterListener(ProcedureStoreListener listener) {
+    return this.listeners.remove(listener);
+  }
+
+  @Override
+  public void recoverLease() throws IOException {
+    LOG.info("Starting WAL Procedure Store lease recovery");
+    FileStatus[] oldLogs = getLogFiles();
+    while (running.get()) {
+      // Get Log-MaxID and recover lease on old logs
+      flushLogId = initOldLogs(oldLogs) + 1;
+
+      // Create new state-log
+      if (!rollWriter(flushLogId)) {
+        // someone else has already created this log
+        continue;
+      }
+
+      // We have the lease on the log
+      oldLogs = getLogFiles();
+      if (getMaxLogId(oldLogs) > flushLogId) {
+        // Someone else created new logs
+        logs.getLast().removeFile();
+        continue;
+      }
+
+      LOG.info("lease acquired flushLogId=" + flushLogId);
+      break;
+    }
+  }
+
+  @Override
+  public Iterator<Procedure> load() throws IOException {
+    if (logs.isEmpty()) {
+      throw new RuntimeException("recoverLease() must be called before loading data");
+    }
+
+    // Nothing to do, If we have only the current log.
+    if (logs.size() == 1) {
+      LOG.debug("No state logs to replay");
+      return null;
+    }
+
+    // Load the old logs
+    final ArrayList<ProcedureWALFile> toRemove = new ArrayList<ProcedureWALFile>();
+    Iterator<ProcedureWALFile> it = logs.descendingIterator();
+    it.next(); // Skip the current log
+    try {
+      return ProcedureWALFormat.load(it, storeTracker, new ProcedureWALFormat.Loader() {
+        @Override
+        public void removeLog(ProcedureWALFile log) {
+          toRemove.add(log);
+        }
+
+        @Override
+        public void markCorruptedWAL(ProcedureWALFile log, IOException e) {
+          if (corruptedLogs == null) {
+            corruptedLogs = new HashSet<ProcedureWALFile>();
+          }
+          corruptedLogs.add(log);
+          // TODO: sideline corrupted log
+        }
+      });
+    } finally {
+      if (!toRemove.isEmpty()) {
+        for (ProcedureWALFile log: toRemove) {
+          removeLogFile(log);
+        }
+      }
+    }
+  }
+
+  @Override
+  public void insert(final Procedure proc, final Procedure[] subprocs) {
+    if (LOG.isTraceEnabled()) {
+      LOG.trace("insert " + proc + " subproc=" + Arrays.toString(subprocs));
+    }
+
+    ByteSlot slot = acquireSlot();
+    long logId = -1;
+    try {
+      // Serialize the insert
+      if (subprocs != null) {
+        ProcedureWALFormat.writeInsert(slot, proc, subprocs);
+      } else {
+        assert !proc.hasParent();
+        ProcedureWALFormat.writeInsert(slot, proc);
+      }
+
+      // Push the transaction data and wait until it is persisted
+      logId = pushData(slot);
+    } catch (IOException e) {
+      // We are not able to serialize the procedure.
+      // this is a code error, and we are not able to go on.
+      LOG.fatal("Unable to serialize one of the procedure: proc=" + proc +
+                " subprocs=" + Arrays.toString(subprocs), e);
+      throw new RuntimeException(e);
+    } finally {
+      releaseSlot(slot);
+    }
+
+    // Update the store tracker
+    synchronized (storeTracker) {
+      if (logId == flushLogId) {
+        storeTracker.insert(proc, subprocs);
+      }
+    }
+  }
+
+  @Override
+  public void update(final Procedure proc) {
+    if (LOG.isTraceEnabled()) {
+      LOG.trace("update " + proc);
+    }
+
+    ByteSlot slot = acquireSlot();
+    long logId = -1;
+    try {
+      // Serialize the update
+      ProcedureWALFormat.writeUpdate(slot, proc);
+
+      // Push the transaction data and wait until it is persisted
+      logId = pushData(slot);
+    } catch (IOException e) {
+      // We are not able to serialize the procedure.
+      // this is a code error, and we are not able to go on.
+      LOG.fatal("Unable to serialize the procedure: " + proc, e);
+      throw new RuntimeException(e);
+    } finally {
+      releaseSlot(slot);
+    }
+
+    // Update the store tracker
+    boolean removeOldLogs = false;
+    synchronized (storeTracker) {
+      if (logId == flushLogId) {
+        storeTracker.update(proc);
+        removeOldLogs = storeTracker.isUpdated();
+      }
+    }
+
+    if (removeOldLogs) {
+      removeAllLogs(logId - 1);
+    }
+  }
+
+  @Override
+  public void delete(final long procId) {
+    if (LOG.isTraceEnabled()) {
+      LOG.trace("delete " + procId);
+    }
+
+    ByteSlot slot = acquireSlot();
+    long logId = -1;
+    try {
+      // Serialize the delete
+      ProcedureWALFormat.writeDelete(slot, procId);
+
+      // Push the transaction data and wait until it is persisted
+      logId = pushData(slot);
+    } catch (IOException e) {
+      // We are not able to serialize the procedure.
+      // this is a code error, and we are not able to go on.
+      LOG.fatal("Unable to serialize the procedure: " + procId, e);
+      throw new RuntimeException(e);
+    } finally {
+      releaseSlot(slot);
+    }
+
+    boolean removeOldLogs = false;
+    synchronized (storeTracker) {
+      if (logId == flushLogId) {
+        storeTracker.delete(procId);
+        if (storeTracker.isEmpty()) {
+          removeOldLogs = rollWriterOrDie(logId + 1);
+        }
+      }
+    }
+
+    if (removeOldLogs) {
+      removeAllLogs(logId);
+    }
+  }
+
+  private ByteSlot acquireSlot() {
+    ByteSlot slot = slotsCache.poll();
+    return slot != null ? slot : new ByteSlot();
+  }
+
+  private void releaseSlot(final ByteSlot slot) {
+    slot.reset();
+    slotsCache.offer(slot);
+  }
+
+  private long pushData(final ByteSlot slot) {
+    assert !logs.isEmpty() : "recoverLease() must be called before inserting data";
+    long logId = -1;
+
+    lock.lock();
+    try {
+      // Wait for the sync to be completed
+      while (true) {
+        if (inSync.get()) {
+          syncCond.await();
+        } else if (slotIndex == slots.length) {
+          slotCond.signal();
+          syncCond.await();
+        } else {
+          break;
+        }
+      }
+
+      slots[slotIndex++] = slot;
+      logId = flushLogId;
+
+      // Notify that there is new data
+      if (slotIndex == 1) {
+        waitCond.signal();
+      }
+
+      // Notify that the slots are full
+      if (slotIndex == slots.length) {
+        slotCond.signal();
+      }
+      syncCond.await();
+    } catch (InterruptedException e) {
+      sendAbortProcessSignal();
+    } finally {
+      lock.unlock();
+    }
+    return logId;
+  }
+
+  private void syncLoop() throws IOException {
+    inSync.set(false);
+    while (running.get()) {
+      lock.lock();
+      try {
+        // Wait until new data is available
+        if (slotIndex == 0) {
+          LOG.debug("Waiting for data. flushed=" + StringUtils.humanSize(totalSynced));
+          waitCond.await();
+          if (slotIndex == 0) {
+            // no data.. probably a stop()
+            continue;
+          }
+        }
+
+        // Wait SYNC_WAIT_MSEC or the signal of "slots full" before flushing
+        slotCond.await(syncWaitMsec, TimeUnit.MILLISECONDS);
+
+        inSync.set(true);
+        totalSynced += syncSlots();
+        slotIndex = 0;
+        inSync.set(false);
+        syncCond.signalAll();
+      } catch (InterruptedException e) {
+        sendAbortProcessSignal();
+      } finally {
+        lock.unlock();
+      }
+    }
+  }
+
+  private long syncSlots() {
+    int retry = 0;
+    long totalSynced = 0;
+    do {
+      try {
+        totalSynced = syncSlots(stream, slots, 0, slotIndex);
+        break;
+      } catch (IOException e) {
+        if (++retry == MAX_RETRIES_BEFORE_ABORT) {
+          LOG.error("sync slot failed, abort.", e);
+          sendAbortProcessSignal();
+        }
+      }
+    } while (running.get());
+    return totalSynced;
+  }
+
+  protected long syncSlots(FSDataOutputStream stream, ByteSlot[] slots, int offset, int count)
+      throws IOException {
+    long totalSynced = 0;
+    for (int i = 0; i < count; ++i) {
+      ByteSlot data = slots[offset + i];
+      data.writeTo(stream);
+      totalSynced += data.size();
+    }
+    stream.hsync();
+    return totalSynced;
+  }
+
+  private void sendAbortProcessSignal() {
+    if (!this.listeners.isEmpty()) {
+      for (ProcedureStoreListener listener : this.listeners) {
+        listener.abortProcess();
+      }
+    }
+  }
+
+  private boolean rollWriterOrDie(final long logId) {
+    try {
+      return rollWriter(logId);
+    } catch (IOException e) {
+      LOG.warn("Unable to roll the log", e);
+      sendAbortProcessSignal();
+      return false;
+    }
+  }
+
+  private boolean rollWriter(final long logId) throws IOException {
+    ProcedureWALHeader header = ProcedureWALHeader.newBuilder()
+      .setVersion(ProcedureWALFormat.HEADER_VERSION)
+      .setType(ProcedureWALFormat.LOG_TYPE_STREAM)
+      .setMinProcId(storeTracker.getMinProcId())
+      .setLogId(logId)
+      .build();
+
+    FSDataOutputStream newStream = null;
+    Path newLogFile = null;
+    long startPos = -1;
+    try {
+      newLogFile = getLogFilePath(logId);
+      newStream = fs.create(newLogFile, false);
+      ProcedureWALFormat.writeHeader(newStream, header);
+      startPos = newStream.getPos();
+    } catch (FileAlreadyExistsException e) {
+      LOG.error("Log file with id=" + logId + " already exists", e);
+      return false;
+    }
+    lock.lock();
+    try {
+      closeStream();
+      synchronized (storeTracker) {
+        storeTracker.resetUpdates();
+      }
+      stream = newStream;
+      flushLogId = logId;
+      totalSynced = 0;
+      logs.add(new ProcedureWALFile(fs, newLogFile, header, startPos));
+    } finally {
+      lock.unlock();
+    }
+    LOG.info("Roll new state log: " + logId);
+    return true;
+  }
+
+  private void closeStream() {
+    try {
+      if (stream != null) {
+        try {
+          ProcedureWALFormat.writeTrailer(stream, storeTracker);
+        } catch (IOException e) {
+          LOG.error("Unable to write the trailer", e);
+        }
+        stream.close();
+      }
+    } catch (IOException e) {
+      LOG.error("Unable to close the stream", e);
+    } finally {
+      stream = null;
+    }
+  }
+
+  private void removeAllLogs(long lastLogId) {
+    LOG.info("Remove all state logs with ID less then " + lastLogId);
+    while (!logs.isEmpty()) {
+      ProcedureWALFile log = logs.getFirst();
+      if (lastLogId < log.getLogId()) {
+        break;
+      }
+
+      removeLogFile(log);
+    }
+  }
+
+  private boolean removeLogFile(final ProcedureWALFile log) {
+    try {
+      LOG.debug("remove log: " + log);
+      log.removeFile();
+      logs.remove(log);
+    } catch (IOException e) {
+      LOG.error("unable to remove log " + log, e);
+      return false;
+    }
+    return true;
+  }
+
+  public Set<ProcedureWALFile> getCorruptedLogs() {
+    return corruptedLogs;
+  }
+
+  // ==========================================================================
+  //  FileSystem Log Files helpers
+  // ==========================================================================
+
+  protected Path getLogFilePath(final long logId) throws IOException {
+    return new Path(logDir, String.format("state-%020d.log", logId));
+  }
+
+  private static long getLogIdFromName(final String name) {
+    int end = name.lastIndexOf(".log");
+    int start = name.lastIndexOf('-') + 1;
+    while (start < end) {
+      if (name.charAt(start) != '0')
+        break;
+      start++;
+    }
+    return Long.parseLong(name.substring(start, end));
+  }
+
+  private FileStatus[] getLogFiles() throws IOException {
+    try {
+      return fs.listStatus(logDir, new PathFilter() {
+        @Override
+        public boolean accept(Path path) {
+          String name = path.getName();
+          return name.startsWith("state-") && name.endsWith(".log");
+        }
+      });
+    } catch (FileNotFoundException e) {
+      LOG.warn("log directory not found: " + e.getMessage());
+      return null;
+    }
+  }
+
+  private long getMaxLogId(final FileStatus[] logFiles) {
+    long maxLogId = 0;
+    if (logFiles != null && logFiles.length > 0) {
+      for (int i = 0; i < logFiles.length; ++i) {
+        maxLogId = Math.max(maxLogId, getLogIdFromName(logFiles[i].getPath().getName()));
+      }
+    }
+    return maxLogId;
+  }
+
+  /**
+   * @return Max-LogID of the specified log file set
+   */
+  private long initOldLogs(final FileStatus[] logFiles) throws IOException {
+    this.logs.clear();
+
+    long maxLogId = 0;
+    if (logFiles != null && logFiles.length > 0) {
+      for (int i = 0; i < logFiles.length; ++i) {
+        final Path logPath = logFiles[i].getPath();
+        leaseRecovery.recoverFileLease(fs, logPath);
+        maxLogId = Math.max(maxLogId, getLogIdFromName(logPath.getName()));
+
+        ProcedureWALFile log = initOldLog(logFiles[i]);
+        if (log != null) {
+          this.logs.add(log);
+        }
+      }
+      Collections.sort(this.logs);
+      initTrackerFromOldLogs();
+    }
+    return maxLogId;
+  }
+
+  private void initTrackerFromOldLogs() {
+    // TODO: Load the most recent tracker available
+    if (!logs.isEmpty()) {
+      ProcedureWALFile log = logs.getLast();
+      try {
+        log.readTracker(storeTracker);
+      } catch (IOException e) {
+        LOG.error("Unable to read tracker for " + log, e);
+        // try the next one...
+        storeTracker.clear();
+        storeTracker.setPartialFlag(true);
+      }
+    }
+  }
+
+  private ProcedureWALFile initOldLog(final FileStatus logFile) throws IOException {
+    ProcedureWALFile log = new ProcedureWALFile(fs, logFile);
+    try {
+      log.open();
+      LOG.debug("opening state-log: " + log);
+      if (log.isCompacted()) {
+        try {
+          log.readTrailer();
+        } catch (IOException e) {
+          // unfinished compacted log throw it away
+          LOG.warn("Unfinished compacted log " + logFile, e);
+          log.removeFile();
+          return null;
+        }
+      }
+    } catch (IOException e) {
+      if (logFile.getLen() == 0) {
+        LOG.warn("Remove uninitialized log " + logFile, e);
+        log.removeFile();
+        return null;
+      }
+
+      String msg = "Unable to read state log " + logFile;
+      LOG.error(msg, e);
+      throw new IOException(msg, e);
+    }
+    return log;
+  }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/hbase/blob/e3db688a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/util/ByteSlot.java
----------------------------------------------------------------------
diff --git a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/util/ByteSlot.java b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/util/ByteSlot.java
new file mode 100644
index 0000000..8904116
--- /dev/null
+++ b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/util/ByteSlot.java
@@ -0,0 +1,111 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase.procedure2.util;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.Arrays;
+
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.classification.InterfaceStability;
+
+/**
+ * Similar to the ByteArrayOutputStream, with the exception that we can prepend an header.
+ * e.g. you write some data and you want to prepend an header that contains the data len or cksum.
+ * <code>
+ * ByteSlot slot = new ByteSlot();
+ * // write data
+ * slot.write(...);
+ * slot.write(...);
+ * // write header with the size of the written data
+ * slot.markHead();
+ * slot.write(Bytes.toBytes(slot.size()));
+ * // flush to stream as [header, data]
+ * slot.writeTo(stream);
+ * </code>
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public class ByteSlot extends OutputStream {
+  private static final int DOUBLE_GROW_LIMIT = 1 << 20;
+  private static final int GROW_ALIGN = 128;
+
+  private byte[] buf;
+  private int head;
+  private int size;
+
+  public void reset() {
+    head = 0;
+    size = 0;
+  }
+
+  public void markHead() {
+    head = size;
+  }
+
+  public int getHead() {
+    return head;
+  }
+
+  public int size() {
+    return size;
+  }
+
+  public byte[] getBuffer() {
+    return buf;
+  }
+
+  public void writeAt(int offset, int b) {
+    head = Math.min(head, offset);
+    buf[offset] = (byte)b;
+  }
+
+  public void write(int b) {
+    ensureCapacity(size + 1);
+    buf[size++] = (byte)b;
+  }
+
+  public void write(byte[] b, int off, int len) {
+    ensureCapacity(size + len);
+    System.arraycopy(b, off, buf, size, len);
+    size += len;
+  }
+
+  public void writeTo(final OutputStream stream) throws IOException {
+    if (head != 0) {
+      stream.write(buf, head, size - head);
+      stream.write(buf, 0, head);
+    } else {
+      stream.write(buf, 0, size);
+    }
+  }
+
+  private void ensureCapacity(int minCapacity) {
+    minCapacity = (minCapacity + (GROW_ALIGN - 1)) & -GROW_ALIGN;
+    if (buf == null) {
+      buf = new byte[minCapacity];
+    } else if (minCapacity > buf.length) {
+      int newCapacity = buf.length << 1;
+      if (minCapacity > newCapacity || newCapacity > DOUBLE_GROW_LIMIT) {
+        newCapacity = minCapacity;
+      }
+      buf = Arrays.copyOf(buf, newCapacity);
+    }
+  }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/hbase/blob/e3db688a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/util/StringUtils.java
----------------------------------------------------------------------
diff --git a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/util/StringUtils.java b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/util/StringUtils.java
new file mode 100644
index 0000000..97134c2
--- /dev/null
+++ b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/util/StringUtils.java
@@ -0,0 +1,80 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase.procedure2.util;
+
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.classification.InterfaceStability;
+
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public final class StringUtils {
+  private StringUtils() {}
+
+  public static String humanTimeDiff(long timeDiff) {
+    StringBuilder buf = new StringBuilder();
+    long hours = timeDiff / (60*60*1000);
+    long rem = (timeDiff % (60*60*1000));
+    long minutes =  rem / (60*1000);
+    rem = rem % (60*1000);
+    float seconds = rem / 1000.0f;
+
+    if (hours != 0){
+      buf.append(hours);
+      buf.append("hrs, ");
+    }
+    if (minutes != 0){
+      buf.append(minutes);
+      buf.append("mins, ");
+    }
+    if (hours > 0 || minutes > 0) {
+      buf.append(seconds);
+      buf.append("sec");
+    } else {
+      buf.append(String.format("%.4fsec", seconds));
+    }
+    return buf.toString();
+  }
+
+  public static String humanSize(double size) {
+    if (size >= (1L << 40)) return String.format("%.1fT", size / (1L << 40));
+    if (size >= (1L << 30)) return String.format("%.1fG", size / (1L << 30));
+    if (size >= (1L << 20)) return String.format("%.1fM", size / (1L << 20));
+    if (size >= (1L << 10)) return String.format("%.1fK", size / (1L << 10));
+    return String.format("%.0f", size);
+  }
+
+  public static boolean isEmpty(final String input) {
+    return input == null || input.length() == 0;
+  }
+
+  public static String buildString(final String... parts) {
+    StringBuilder sb = new StringBuilder();
+    for (int i = 0; i < parts.length; ++i) {
+      sb.append(parts[i]);
+    }
+    return sb.toString();
+  }
+
+  public static StringBuilder appendStrings(final StringBuilder sb, final String... parts) {
+    for (int i = 0; i < parts.length; ++i) {
+      sb.append(parts[i]);
+    }
+    return sb;
+  }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/hbase/blob/e3db688a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/util/TimeoutBlockingQueue.java
----------------------------------------------------------------------
diff --git a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/util/TimeoutBlockingQueue.java b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/util/TimeoutBlockingQueue.java
new file mode 100644
index 0000000..368d26f
--- /dev/null
+++ b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/util/TimeoutBlockingQueue.java
@@ -0,0 +1,216 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase.procedure2.util;
+
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.classification.InterfaceStability;
+
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public class TimeoutBlockingQueue<E> {
+  public static interface TimeoutRetriever<T> {
+    long getTimeout(T object);
+    TimeUnit getTimeUnit(T object);
+  }
+
+  private final ReentrantLock lock = new ReentrantLock();
+  private final Condition waitCond = lock.newCondition();
+  private final TimeoutRetriever<? super E> timeoutRetriever;
+
+  private E[] objects;
+  private int head = 0;
+  private int tail = 0;
+
+  public TimeoutBlockingQueue(TimeoutRetriever<? super E> timeoutRetriever) {
+    this(32, timeoutRetriever);
+  }
+
+  @SuppressWarnings("unchecked")
+  public TimeoutBlockingQueue(int capacity, TimeoutRetriever<? super E> timeoutRetriever) {
+    this.objects = (E[])new Object[capacity];
+    this.timeoutRetriever = timeoutRetriever;
+  }
+
+  public void dump() {
+    for (int i = 0; i < objects.length; ++i) {
+      if (i == head) {
+        System.out.print("[" + objects[i] + "] ");
+      } else if (i == tail) {
+        System.out.print("]" + objects[i] + "[ ");
+      } else {
+        System.out.print(objects[i] + " ");
+      }
+    }
+    System.out.println();
+  }
+
+  public void clear() {
+    lock.lock();
+    try {
+      if (head != tail) {
+        for (int i = head; i < tail; ++i) {
+          objects[i] = null;
+        }
+        head = 0;
+        tail = 0;
+        waitCond.signal();
+      }
+    } finally {
+      lock.unlock();
+    }
+  }
+
+  public void add(E e) {
+    if (e == null) throw new NullPointerException();
+
+    lock.lock();
+    try {
+      addElement(e);
+      waitCond.signal();
+    } finally {
+      lock.unlock();
+    }
+  }
+
+  @edu.umd.cs.findbugs.annotations.SuppressWarnings("WA_AWAIT_NOT_IN_LOOP")
+  public E poll() {
+    lock.lock();
+    try {
+      if (isEmpty()) {
+        waitCond.await();
+        return null;
+      }
+
+      E elem = objects[head];
+      long nanos = getNanosTimeout(elem);
+      nanos = waitCond.awaitNanos(nanos);
+      return nanos > 0 ? null : removeFirst();
+    } catch (InterruptedException e) {
+      return null;
+    } finally {
+      lock.unlock();
+    }
+  }
+
+  public int size() {
+    return tail - head;
+  }
+
+  public boolean isEmpty() {
+    return (tail - head) == 0;
+  }
+
+  public void signalAll() {
+    lock.lock();
+    try {
+      waitCond.signalAll();
+    } finally {
+      lock.unlock();
+    }
+  }
+
+  private void addElement(E elem) {
+    int size = (tail - head);
+    if ((objects.length - size) == 0) {
+      int capacity = size + ((size < 64) ? (size + 2) : (size >> 1));
+      E[] newObjects = (E[])new Object[capacity];
+
+      if (compareTimeouts(objects[tail - 1], elem) <= 0) {
+        // Append
+        System.arraycopy(objects, head, newObjects, 0, tail);
+        tail -= head;
+        newObjects[tail++] = elem;
+      } else if (compareTimeouts(objects[head], elem) > 0) {
+        // Prepend
+        System.arraycopy(objects, head, newObjects, 1, tail);
+        newObjects[0] = elem;
+        tail -= (head - 1);
+      } else {
+        // Insert in the middle
+        int index = upperBound(head, tail - 1, elem);
+        int newIndex = (index - head);
+        System.arraycopy(objects, head, newObjects, 0, newIndex);
+        newObjects[newIndex] = elem;
+        System.arraycopy(objects, index, newObjects, newIndex + 1, tail - index);
+        tail -= (head - 1);
+      }
+      head = 0;
+      objects = newObjects;
+    } else {
+      if (tail == objects.length) {
+        // shift down |-----AAAAAAA|
+        tail -= head;
+        System.arraycopy(objects, head, objects, 0, tail);
+        head = 0;
+      }
+
+      if (tail == head || compareTimeouts(objects[tail - 1], elem) <= 0) {
+        // Append
+        objects[tail++] = elem;
+      } else if (head > 0 && compareTimeouts(objects[head], elem) > 0) {
+        // Prepend
+        objects[--head] = elem;
+      } else {
+        // Insert in the middle
+        int index = upperBound(head, tail - 1, elem);
+        System.arraycopy(objects, index, objects, index + 1, tail - index);
+        objects[index] = elem;
+        tail++;
+      }
+    }
+  }
+
+  private E removeFirst() {
+    E elem = objects[head];
+    objects[head] = null;
+    head = (head + 1) % objects.length;
+    if (head == 0) tail = 0;
+    return elem;
+  }
+
+  private int upperBound(int start, int end, E key) {
+    while (start < end) {
+      int mid = (start + end) >>> 1;
+      E mitem = objects[mid];
+      int cmp = compareTimeouts(mitem, key);
+      if (cmp > 0) {
+        end = mid;
+      } else {
+        start = mid + 1;
+      }
+    }
+    return start;
+  }
+
+  private int compareTimeouts(final E a, final E b) {
+    long t1 = getNanosTimeout(a);
+    long t2 = getNanosTimeout(b);
+    return (t1 < t2) ? -1 : (t1 > t2) ? 1 : 0;
+  }
+
+  private long getNanosTimeout(final E obj) {
+    TimeUnit unit = timeoutRetriever.getTimeUnit(obj);
+    long timeout = timeoutRetriever.getTimeout(obj);
+    return unit.toNanos(timeout);
+  }
+}

http://git-wip-us.apache.org/repos/asf/hbase/blob/e3db688a/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/ProcedureTestingUtility.java
----------------------------------------------------------------------
diff --git a/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/ProcedureTestingUtility.java b/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/ProcedureTestingUtility.java
new file mode 100644
index 0000000..c73300d
--- /dev/null
+++ b/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/ProcedureTestingUtility.java
@@ -0,0 +1,130 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase.procedure2;
+
+import java.io.IOException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.util.Threads;
+import org.apache.hadoop.hbase.procedure2.store.ProcedureStore;
+import org.apache.hadoop.hbase.procedure2.store.wal.WALProcedureStore;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class ProcedureTestingUtility {
+  private static final Log LOG = LogFactory.getLog(ProcedureTestingUtility.class);
+
+  private ProcedureTestingUtility() {
+  }
+
+  public static ProcedureStore createStore(final Configuration conf, final FileSystem fs,
+      final Path baseDir) throws IOException {
+    return createWalStore(conf, fs, baseDir);
+  }
+
+  public static WALProcedureStore createWalStore(final Configuration conf, final FileSystem fs,
+      final Path logDir) throws IOException {
+    return new WALProcedureStore(conf, fs, logDir, new WALProcedureStore.LeaseRecovery() {
+      @Override
+      public void recoverFileLease(FileSystem fs, Path path) throws IOException {
+        // no-op
+      }
+    });
+  }
+
+  public static <TEnv> void restart(ProcedureExecutor<TEnv> procExecutor)
+      throws Exception {
+    ProcedureStore procStore = procExecutor.getStore();
+    int nthreads = procExecutor.getNumThreads();
+    procExecutor.stop();
+    procStore.stop();
+    procExecutor.join();
+    procStore.start(nthreads);
+    procExecutor.start(nthreads);
+  }
+
+  public static <TEnv> void setKillBeforeStoreUpdate(ProcedureExecutor<TEnv> procExecutor,
+      boolean value) {
+    if (procExecutor.testing == null) {
+      procExecutor.testing = new ProcedureExecutor.Testing();
+    }
+    procExecutor.testing.killBeforeStoreUpdate = value;
+    LOG.warn("Set Kill before store update to: " + procExecutor.testing.killBeforeStoreUpdate);
+  }
+
+  public static <TEnv> void setToggleKillBeforeStoreUpdate(ProcedureExecutor<TEnv> procExecutor,
+      boolean value) {
+    if (procExecutor.testing == null) {
+      procExecutor.testing = new ProcedureExecutor.Testing();
+    }
+    procExecutor.testing.toggleKillBeforeStoreUpdate = value;
+  }
+
+  public static <TEnv> void toggleKillBeforeStoreUpdate(ProcedureExecutor<TEnv> procExecutor) {
+    if (procExecutor.testing == null) {
+      procExecutor.testing = new ProcedureExecutor.Testing();
+    }
+    procExecutor.testing.killBeforeStoreUpdate = !procExecutor.testing.killBeforeStoreUpdate;
+    LOG.warn("Set Kill before store update to: " + procExecutor.testing.killBeforeStoreUpdate);
+  }
+
+  public static <TEnv> long submitAndWait(ProcedureExecutor<TEnv> procExecutor, Procedure proc) {
+    long procId = procExecutor.submitProcedure(proc);
+    waitProcedure(procExecutor, procId);
+    return procId;
+  }
+
+  public static <TEnv> void waitProcedure(ProcedureExecutor<TEnv> procExecutor, long procId) {
+    while (!procExecutor.isFinished(procId) && procExecutor.isRunning()) {
+      Threads.sleepWithoutInterrupt(250);
+    }
+  }
+
+  public static <TEnv> void assertProcNotYetCompleted(ProcedureExecutor<TEnv> procExecutor,
+      long procId) {
+    assertFalse(procExecutor.isFinished(procId));
+    assertEquals(null, procExecutor.getResult(procId));
+  }
+
+  public static <TEnv> void assertProcNotFailed(ProcedureExecutor<TEnv> procExecutor,
+      long procId) {
+    ProcedureResult result = procExecutor.getResult(procId);
+    assertTrue(result != null);
+    assertProcNotFailed(result);
+  }
+
+  public static void assertProcNotFailed(final ProcedureResult result) {
+    Exception exception = result.getException();
+    String msg = exception != null ? exception.toString() : "no exception found";
+    assertFalse(msg, result.isFailed());
+  }
+
+  public static void assertIsAbortException(final ProcedureResult result) {
+    LOG.info(result.getException());
+    assertTrue(result.isFailed());
+    assertTrue(result.getException().getCause() instanceof ProcedureAbortedException);
+  }
+}

http://git-wip-us.apache.org/repos/asf/hbase/blob/e3db688a/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/TestProcedureExecution.java
----------------------------------------------------------------------
diff --git a/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/TestProcedureExecution.java b/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/TestProcedureExecution.java
new file mode 100644
index 0000000..43c5362
--- /dev/null
+++ b/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/TestProcedureExecution.java
@@ -0,0 +1,300 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase.procedure2;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HBaseCommonTestingUtility;
+import org.apache.hadoop.hbase.procedure2.store.ProcedureStore;
+import org.apache.hadoop.hbase.protobuf.generated.ProcedureProtos.ProcedureState;
+import org.apache.hadoop.hbase.testclassification.SmallTests;
+import org.apache.hadoop.hbase.testclassification.MasterTests;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+@Category({MasterTests.class, SmallTests.class})
+public class TestProcedureExecution {
+  private static final Log LOG = LogFactory.getLog(TestProcedureExecution.class);
+
+  private static final int PROCEDURE_EXECUTOR_SLOTS = 1;
+  private static final Procedure NULL_PROC = null;
+
+  private ProcedureExecutor<Void> procExecutor;
+  private ProcedureStore procStore;
+
+  private HBaseCommonTestingUtility htu;
+  private FileSystem fs;
+  private Path testDir;
+  private Path logDir;
+
+  @Before
+  public void setUp() throws IOException {
+    htu = new HBaseCommonTestingUtility();
+    testDir = htu.getDataTestDir();
+    fs = testDir.getFileSystem(htu.getConfiguration());
+    assertTrue(testDir.depth() > 1);
+
+    logDir = new Path(testDir, "proc-logs");
+    procStore = ProcedureTestingUtility.createWalStore(htu.getConfiguration(), fs, logDir);
+    procExecutor = new ProcedureExecutor(htu.getConfiguration(), null, procStore);
+    procStore.start(PROCEDURE_EXECUTOR_SLOTS);
+    procExecutor.start(PROCEDURE_EXECUTOR_SLOTS);
+  }
+
+  @After
+  public void tearDown() throws IOException {
+    procExecutor.stop();
+    procStore.stop();
+    fs.delete(logDir, true);
+  }
+
+  private static class TestProcedureException extends Exception {
+    public TestProcedureException(String msg) { super(msg); }
+  }
+
+  public static class TestSequentialProcedure extends SequentialProcedure<Void> {
+    private final Procedure[] subProcs;
+    private final List<String> state;
+    private final Exception failure;
+    private final String name;
+
+    public TestSequentialProcedure() {
+      throw new UnsupportedOperationException("recovery should not be triggered here");
+    }
+
+    public TestSequentialProcedure(String name, List<String> state, Procedure... subProcs) {
+      this.state = state;
+      this.subProcs = subProcs;
+      this.name = name;
+      this.failure = null;
+    }
+
+    public TestSequentialProcedure(String name, List<String> state, Exception failure) {
+      this.state = state;
+      this.subProcs = null;
+      this.name = name;
+      this.failure = failure;
+    }
+
+    @Override
+    protected Procedure[] execute(Void env) {
+      state.add(name + "-execute");
+      if (failure != null) {
+        setFailure(new RemoteProcedureException(name + "-failure", failure));
+        return null;
+      }
+      return subProcs;
+    }
+
+    @Override
+    protected void rollback(Void env) {
+      state.add(name + "-rollback");
+    }
+
+    @Override
+    protected void abort(Void env) {
+      state.add(name + "-abort");
+    }
+  }
+
+  @Test(timeout=30000)
+  public void testBadSubprocList() {
+    List<String> state = new ArrayList<String>();
+    Procedure subProc2 = new TestSequentialProcedure("subProc2", state);
+    Procedure subProc1 = new TestSequentialProcedure("subProc1", state, subProc2, NULL_PROC);
+    Procedure rootProc = new TestSequentialProcedure("rootProc", state, subProc1);
+    long rootId = ProcedureTestingUtility.submitAndWait(procExecutor, rootProc);
+
+    // subProc1 has a "null" subprocedure which is catched as InvalidArgument
+    // failed state with 2 execute and 2 rollback
+    LOG.info(state);
+    ProcedureResult result = procExecutor.getResult(rootId);
+    LOG.info(result.getException());
+    assertTrue(state.toString(), result.isFailed());
+    assertTrue(result.getException().toString(),
+      result.getException().getCause() instanceof IllegalArgumentException);
+
+    assertEquals(state.toString(), 4, state.size());
+    assertEquals("rootProc-execute", state.get(0));
+    assertEquals("subProc1-execute", state.get(1));
+    assertEquals("subProc1-rollback", state.get(2));
+    assertEquals("rootProc-rollback", state.get(3));
+  }
+
+  @Test(timeout=30000)
+  public void testSingleSequentialProc() {
+    List<String> state = new ArrayList<String>();
+    Procedure subProc2 = new TestSequentialProcedure("subProc2", state);
+    Procedure subProc1 = new TestSequentialProcedure("subProc1", state, subProc2);
+    Procedure rootProc = new TestSequentialProcedure("rootProc", state, subProc1);
+    long rootId = ProcedureTestingUtility.submitAndWait(procExecutor, rootProc);
+
+    // successful state, with 3 execute
+    LOG.info(state);
+    ProcedureResult result = procExecutor.getResult(rootId);
+    ProcedureTestingUtility.assertProcNotFailed(result);
+    assertEquals(state.toString(), 3, state.size());
+  }
+
+  @Test(timeout=30000)
+  public void testSingleSequentialProcRollback() {
+    List<String> state = new ArrayList<String>();
+    Procedure subProc2 = new TestSequentialProcedure("subProc2", state,
+                                                     new TestProcedureException("fail test"));
+    Procedure subProc1 = new TestSequentialProcedure("subProc1", state, subProc2);
+    Procedure rootProc = new TestSequentialProcedure("rootProc", state, subProc1);
+    long rootId = ProcedureTestingUtility.submitAndWait(procExecutor, rootProc);
+
+    // the 3rd proc fail, rollback after 2 successful execution
+    LOG.info(state);
+    ProcedureResult result = procExecutor.getResult(rootId);
+    LOG.info(result.getException());
+    assertTrue(state.toString(), result.isFailed());
+    assertTrue(result.getException().toString(),
+      result.getException().getCause() instanceof TestProcedureException);
+
+    assertEquals(state.toString(), 6, state.size());
+    assertEquals("rootProc-execute", state.get(0));
+    assertEquals("subProc1-execute", state.get(1));
+    assertEquals("subProc2-execute", state.get(2));
+    assertEquals("subProc2-rollback", state.get(3));
+    assertEquals("subProc1-rollback", state.get(4));
+    assertEquals("rootProc-rollback", state.get(5));
+  }
+
+  public static class TestWaitingProcedure extends SequentialProcedure<Void> {
+    private final List<String> state;
+    private final boolean hasChild;
+    private final String name;
+
+    public TestWaitingProcedure() {
+      throw new UnsupportedOperationException("recovery should not be triggered here");
+    }
+
+    public TestWaitingProcedure(String name, List<String> state, boolean hasChild) {
+      this.hasChild = hasChild;
+      this.state = state;
+      this.name = name;
+    }
+
+    @Override
+    protected Procedure[] execute(Void env) {
+      state.add(name + "-execute");
+      setState(ProcedureState.WAITING_TIMEOUT);
+      return hasChild ? new Procedure[] { new TestWaitChild(name, state) } : null;
+    }
+
+    @Override
+    protected void rollback(Void env) {
+      state.add(name + "-rollback");
+    }
+
+    @Override
+    protected void abort(Void env) {
+      state.add(name + "-abort");
+    }
+
+    public static class TestWaitChild extends SequentialProcedure<Void> {
+      private final List<String> state;
+      private final String name;
+
+      public TestWaitChild() {
+        throw new UnsupportedOperationException("recovery should not be triggered here");
+      }
+
+      public TestWaitChild(String name, List<String> state) {
+        this.name = name;
+        this.state = state;
+      }
+
+      @Override
+      protected Procedure[] execute(Void env) {
+        state.add(name + "-child-execute");
+        return null;
+      }
+
+      @Override
+      protected void rollback(Void env) {
+        state.add(name + "-child-rollback");
+      }
+
+      @Override
+      protected void abort(Void env) {
+        state.add(name + "-child-abort");
+      }
+    }
+  }
+
+  @Test(timeout=30000)
+  public void testAbortTimeout() {
+    final int PROC_TIMEOUT_MSEC = 2500;
+    List<String> state = new ArrayList<String>();
+    Procedure proc = new TestWaitingProcedure("wproc", state, false);
+    proc.setTimeout(PROC_TIMEOUT_MSEC);
+    long startTime = EnvironmentEdgeManager.currentTime();
+    long rootId = ProcedureTestingUtility.submitAndWait(procExecutor, proc);
+    long execTime = EnvironmentEdgeManager.currentTime() - startTime;
+    LOG.info(state);
+    assertTrue("we didn't wait enough execTime=" + execTime, execTime >= PROC_TIMEOUT_MSEC);
+    ProcedureResult result = procExecutor.getResult(rootId);
+    LOG.info(result.getException());
+    assertTrue(state.toString(), result.isFailed());
+    assertTrue(result.getException().toString(),
+               result.getException().getCause() instanceof TimeoutException);
+    assertEquals(state.toString(), 2, state.size());
+    assertEquals("wproc-execute", state.get(0));
+    assertEquals("wproc-rollback", state.get(1));
+  }
+
+  @Test(timeout=30000)
+  public void testAbortTimeoutWithChildren() {
+    List<String> state = new ArrayList<String>();
+    Procedure proc = new TestWaitingProcedure("wproc", state, true);
+    proc.setTimeout(2500);
+    long rootId = ProcedureTestingUtility.submitAndWait(procExecutor, proc);
+    LOG.info(state);
+    ProcedureResult result = procExecutor.getResult(rootId);
+    LOG.info(result.getException());
+    assertTrue(state.toString(), result.isFailed());
+    assertTrue(result.getException().toString(),
+               result.getException().getCause() instanceof TimeoutException);
+    assertEquals(state.toString(), 4, state.size());
+    assertEquals("wproc-execute", state.get(0));
+    assertEquals("wproc-child-execute", state.get(1));
+    assertEquals("wproc-child-rollback", state.get(2));
+    assertEquals("wproc-rollback", state.get(3));
+  }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/hbase/blob/e3db688a/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/TestProcedureRecovery.java
----------------------------------------------------------------------
diff --git a/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/TestProcedureRecovery.java b/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/TestProcedureRecovery.java
new file mode 100644
index 0000000..ffc401c
--- /dev/null
+++ b/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/TestProcedureRecovery.java
@@ -0,0 +1,486 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase.procedure2;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HBaseCommonTestingUtility;
+import org.apache.hadoop.hbase.procedure2.store.ProcedureStore;
+import org.apache.hadoop.hbase.testclassification.SmallTests;
+import org.apache.hadoop.hbase.testclassification.MasterTests;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+import org.apache.hadoop.hbase.util.Threads;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+@Category({MasterTests.class, SmallTests.class})
+public class TestProcedureRecovery {
+  private static final Log LOG = LogFactory.getLog(TestProcedureRecovery.class);
+
+  private static final int PROCEDURE_EXECUTOR_SLOTS = 1;
+  private static final Procedure NULL_PROC = null;
+
+  private static ProcedureExecutor<Void> procExecutor;
+  private static ProcedureStore procStore;
+  private static int procSleepInterval;
+
+  private HBaseCommonTestingUtility htu;
+  private FileSystem fs;
+  private Path testDir;
+  private Path logDir;
+
+  @Before
+  public void setUp() throws IOException {
+    htu = new HBaseCommonTestingUtility();
+    testDir = htu.getDataTestDir();
+    fs = testDir.getFileSystem(htu.getConfiguration());
+    assertTrue(testDir.depth() > 1);
+
+    logDir = new Path(testDir, "proc-logs");
+    procStore = ProcedureTestingUtility.createStore(htu.getConfiguration(), fs, logDir);
+    procExecutor = new ProcedureExecutor(htu.getConfiguration(), null, procStore);
+    procExecutor.testing = new ProcedureExecutor.Testing();
+    procStore.start(PROCEDURE_EXECUTOR_SLOTS);
+    procExecutor.start(PROCEDURE_EXECUTOR_SLOTS);
+    procSleepInterval = 0;
+  }
+
+  @After
+  public void tearDown() throws IOException {
+    procExecutor.stop();
+    procStore.stop();
+    fs.delete(logDir, true);
+  }
+
+  private void restart() throws Exception {
+    dumpLogDirState();
+    ProcedureTestingUtility.restart(procExecutor);
+    dumpLogDirState();
+  }
+
+  public static class TestSingleStepProcedure extends SequentialProcedure<Void> {
+    private int step = 0;
+
+    public TestSingleStepProcedure() { }
+
+    @Override
+    protected Procedure[] execute(Void env) {
+      LOG.debug("execute procedure " + this + " step=" + step);
+      step++;
+      setResult(Bytes.toBytes(step));
+      return null;
+    }
+
+    @Override
+    protected void rollback(Void env) { }
+
+    @Override
+    protected void abort(Void env) { }
+  }
+
+  public static class BaseTestStepProcedure extends SequentialProcedure<Void> {
+    private AtomicBoolean abort = new AtomicBoolean(false);
+    private int step = 0;
+
+    @Override
+    protected Procedure[] execute(Void env) {
+      LOG.debug("execute procedure " + this + " step=" + step);
+      ProcedureTestingUtility.toggleKillBeforeStoreUpdate(procExecutor);
+      step++;
+      Threads.sleepWithoutInterrupt(procSleepInterval);
+      if (isAborted()) {
+        setFailure(new RemoteProcedureException(getClass().getName(),
+          new ProcedureAbortedException(
+            "got an abort at " + getClass().getName() + " step=" + step)));
+        return null;
+      }
+      return null;
+    }
+
+    @Override
+    protected void rollback(Void env) {
+      LOG.debug("rollback procedure " + this + " step=" + step);
+      ProcedureTestingUtility.toggleKillBeforeStoreUpdate(procExecutor);
+      step++;
+    }
+
+    @Override
+    protected void abort(Void env) {
+      abort.set(true);
+    }
+
+    private boolean isAborted() {
+      boolean aborted = abort.get();
+      BaseTestStepProcedure proc = this;
+      while (proc.hasParent() && !aborted) {
+        proc = (BaseTestStepProcedure)procExecutor.getProcedure(proc.getParentProcId());
+        aborted = proc.isAborted();
+      }
+      return aborted;
+    }
+  }
+
+  public static class TestMultiStepProcedure extends BaseTestStepProcedure {
+    public TestMultiStepProcedure() { }
+
+    @Override
+    public Procedure[] execute(Void env) {
+      super.execute(env);
+      return isFailed() ? null : new Procedure[] { new Step1Procedure() };
+    }
+
+    public static class Step1Procedure extends BaseTestStepProcedure {
+      public Step1Procedure() { }
+
+      @Override
+      protected Procedure[] execute(Void env) {
+        super.execute(env);
+        return isFailed() ? null : new Procedure[] { new Step2Procedure() };
+      }
+    }
+
+    public static class Step2Procedure extends BaseTestStepProcedure {
+      public Step2Procedure() { }
+    }
+  }
+
+  @Test
+  public void testNoopLoad() throws Exception {
+    restart();
+  }
+
+  @Test(timeout=30000)
+  public void testSingleStepProcRecovery() throws Exception {
+    Procedure proc = new TestSingleStepProcedure();
+    procExecutor.testing.killBeforeStoreUpdate = true;
+    long procId = ProcedureTestingUtility.submitAndWait(procExecutor, proc);
+    assertFalse(procExecutor.isRunning());
+    procExecutor.testing.killBeforeStoreUpdate = false;
+
+    // Restart and verify that the procedures restart
+    long restartTs = EnvironmentEdgeManager.currentTime();
+    restart();
+    waitProcedure(procId);
+    ProcedureResult result = procExecutor.getResult(procId);
+    assertTrue(result.getLastUpdate() > restartTs);
+    ProcedureTestingUtility.assertProcNotFailed(result);
+    assertEquals(1, Bytes.toInt(result.getResult()));
+    long resultTs = result.getLastUpdate();
+
+    // Verify that after another restart the result is still there
+    restart();
+    result = procExecutor.getResult(procId);
+    ProcedureTestingUtility.assertProcNotFailed(result);
+    assertEquals(resultTs, result.getLastUpdate());
+    assertEquals(1, Bytes.toInt(result.getResult()));
+  }
+
+  @Test(timeout=30000)
+  public void testMultiStepProcRecovery() throws Exception {
+    // Step 0 - kill
+    Procedure proc = new TestMultiStepProcedure();
+    long procId = ProcedureTestingUtility.submitAndWait(procExecutor, proc);
+    assertFalse(procExecutor.isRunning());
+
+    // Step 0 exec && Step 1 - kill
+    restart();
+    waitProcedure(procId);
+    ProcedureTestingUtility.assertProcNotYetCompleted(procExecutor, procId);
+    assertFalse(procExecutor.isRunning());
+
+    // Step 1 exec && step 2 - kill
+    restart();
+    waitProcedure(procId);
+    ProcedureTestingUtility.assertProcNotYetCompleted(procExecutor, procId);
+    assertFalse(procExecutor.isRunning());
+
+    // Step 2 exec
+    restart();
+    waitProcedure(procId);
+    assertTrue(procExecutor.isRunning());
+
+    // The procedure is completed
+    ProcedureResult result = procExecutor.getResult(procId);
+    ProcedureTestingUtility.assertProcNotFailed(result);
+  }
+
+  @Test(timeout=30000)
+  public void testMultiStepRollbackRecovery() throws Exception {
+    // Step 0 - kill
+    Procedure proc = new TestMultiStepProcedure();
+    long procId = ProcedureTestingUtility.submitAndWait(procExecutor, proc);
+    assertFalse(procExecutor.isRunning());
+
+    // Step 0 exec && Step 1 - kill
+    restart();
+    waitProcedure(procId);
+    ProcedureTestingUtility.assertProcNotYetCompleted(procExecutor, procId);
+    assertFalse(procExecutor.isRunning());
+
+    // Step 1 exec && step 2 - kill
+    restart();
+    waitProcedure(procId);
+    ProcedureTestingUtility.assertProcNotYetCompleted(procExecutor, procId);
+    assertFalse(procExecutor.isRunning());
+
+    // Step 2 exec - rollback - kill
+    procSleepInterval = 2500;
+    restart();
+    assertTrue(procExecutor.abort(procId));
+    waitProcedure(procId);
+    assertFalse(procExecutor.isRunning());
+
+    // rollback - kill
+    restart();
+    waitProcedure(procId);
+    ProcedureTestingUtility.assertProcNotYetCompleted(procExecutor, procId);
+    assertFalse(procExecutor.isRunning());
+
+    // rollback - complete
+    restart();
+    waitProcedure(procId);
+    ProcedureTestingUtility.assertProcNotYetCompleted(procExecutor, procId);
+    assertFalse(procExecutor.isRunning());
+
+    // Restart the executor and get the result
+    restart();
+    waitProcedure(procId);
+
+    // The procedure is completed
+    ProcedureResult result = procExecutor.getResult(procId);
+    ProcedureTestingUtility.assertIsAbortException(result);
+  }
+
+  public static class TestStateMachineProcedure
+      extends StateMachineProcedure<Void, TestStateMachineProcedure.State> {
+    enum State { STATE_1, STATE_2, STATE_3, DONE }
+
+    public TestStateMachineProcedure() {}
+
+    private AtomicBoolean aborted = new AtomicBoolean(false);
+    private int iResult = 0;
+
+    @Override
+    protected StateMachineProcedure.Flow executeFromState(Void env, State state) {
+      if (state == null) {
+        LOG.info("Initializing " + this);
+        state = State.STATE_1;
+        setNextState(state);
+      }
+
+      switch (state) {
+        case STATE_1:
+          LOG.info("execute step 1 " + this);
+          setNextState(State.STATE_2);
+          iResult += 3;
+          break;
+        case STATE_2:
+          LOG.info("execute step 2 " + this);
+          setNextState(State.STATE_3);
+          iResult += 5;
+          break;
+        case STATE_3:
+          LOG.info("execute step 3 " + this);
+          Threads.sleepWithoutInterrupt(procSleepInterval);
+          if (aborted.get()) {
+            LOG.info("aborted step 3 " + this);
+            setAbortFailure("test", "aborted");
+            break;
+          }
+          setNextState(State.DONE);
+          iResult += 7;
+          setResult(Bytes.toBytes(iResult));
+          return Flow.NO_MORE_STATE;
+        default:
+          throw new UnsupportedOperationException();
+      }
+      return Flow.HAS_MORE_STATE;
+    }
+
+    @Override
+    protected void rollbackState(Void env, final State state) {
+      switch (state) {
+        case STATE_1:
+          LOG.info("rollback step 1 " + this);
+          break;
+        case STATE_2:
+          LOG.info("rollback step 2 " + this);
+          break;
+        case STATE_3:
+          LOG.info("rollback step 3 " + this);
+          break;
+        default:
+          throw new UnsupportedOperationException();
+      }
+    }
+
+    @Override
+    protected State getState(final int stateId) {
+      return State.values()[stateId];
+    }
+
+    private void setNextState(final State state) {
+      setNextState(state.ordinal());
+    }
+
+    @Override
+    protected void abort(Void env) {
+      aborted.set(true);
+    }
+
+    @Override
+    protected void serializeStateData(final OutputStream stream) throws IOException {
+      super.serializeStateData(stream);
+      stream.write(Bytes.toBytes(iResult));
+    }
+
+    @Override
+    protected void deserializeStateData(final InputStream stream) throws IOException {
+      super.deserializeStateData(stream);
+      byte[] data = new byte[4];
+      stream.read(data);
+      iResult = Bytes.toInt(data);
+    }
+  }
+
+  @Test(timeout=30000)
+  public void testStateMachineRecovery() throws Exception {
+    ProcedureTestingUtility.setToggleKillBeforeStoreUpdate(procExecutor, true);
+    ProcedureTestingUtility.setKillBeforeStoreUpdate(procExecutor, true);
+
+    // Step 1 - kill
+    Procedure proc = new TestStateMachineProcedure();
+    long procId = ProcedureTestingUtility.submitAndWait(procExecutor, proc);
+    assertFalse(procExecutor.isRunning());
+
+    // Step 1 exec && Step 2 - kill
+    restart();
+    waitProcedure(procId);
+    ProcedureTestingUtility.assertProcNotYetCompleted(procExecutor, procId);
+    assertFalse(procExecutor.isRunning());
+
+    // Step 2 exec && step 3 - kill
+    restart();
+    waitProcedure(procId);
+    ProcedureTestingUtility.assertProcNotYetCompleted(procExecutor, procId);
+    assertFalse(procExecutor.isRunning());
+
+    // Step 3 exec
+    restart();
+    waitProcedure(procId);
+    assertTrue(procExecutor.isRunning());
+
+    // The procedure is completed
+    ProcedureResult result = procExecutor.getResult(procId);
+    ProcedureTestingUtility.assertProcNotFailed(result);
+    assertEquals(15, Bytes.toInt(result.getResult()));
+  }
+
+  @Test(timeout=30000)
+  public void testStateMachineRollbackRecovery() throws Exception {
+    ProcedureTestingUtility.setToggleKillBeforeStoreUpdate(procExecutor, true);
+    ProcedureTestingUtility.setKillBeforeStoreUpdate(procExecutor, true);
+
+    // Step 1 - kill
+    Procedure proc = new TestStateMachineProcedure();
+    long procId = ProcedureTestingUtility.submitAndWait(procExecutor, proc);
+    ProcedureTestingUtility.assertProcNotYetCompleted(procExecutor, procId);
+    assertFalse(procExecutor.isRunning());
+
+    // Step 1 exec && Step 2 - kill
+    restart();
+    waitProcedure(procId);
+    ProcedureTestingUtility.assertProcNotYetCompleted(procExecutor, procId);
+    assertFalse(procExecutor.isRunning());
+
+    // Step 2 exec && step 3 - kill
+    restart();
+    waitProcedure(procId);
+    ProcedureTestingUtility.assertProcNotYetCompleted(procExecutor, procId);
+    assertFalse(procExecutor.isRunning());
+
+    // Step 3 exec - rollback step 3 - kill
+    procSleepInterval = 2500;
+    restart();
+    assertTrue(procExecutor.abort(procId));
+    waitProcedure(procId);
+    ProcedureTestingUtility.assertProcNotYetCompleted(procExecutor, procId);
+    assertFalse(procExecutor.isRunning());
+
+    // Rollback step 3 - rollback step 2 - kill
+    restart();
+    waitProcedure(procId);
+    assertFalse(procExecutor.isRunning());
+    ProcedureTestingUtility.assertProcNotYetCompleted(procExecutor, procId);
+
+    // Rollback step 2 - step 1 - kill
+    restart();
+    waitProcedure(procId);
+    assertFalse(procExecutor.isRunning());
+    ProcedureTestingUtility.assertProcNotYetCompleted(procExecutor, procId);
+
+    // Rollback step 1 - complete
+    restart();
+    waitProcedure(procId);
+    assertTrue(procExecutor.isRunning());
+
+    // The procedure is completed
+    ProcedureResult result = procExecutor.getResult(procId);
+    ProcedureTestingUtility.assertIsAbortException(result);
+  }
+
+  private void waitProcedure(final long procId) {
+    ProcedureTestingUtility.waitProcedure(procExecutor, procId);
+    dumpLogDirState();
+  }
+
+  private void dumpLogDirState() {
+    try {
+      FileStatus[] files = fs.listStatus(logDir);
+      if (files != null && files.length > 0) {
+        for (FileStatus file: files) {
+          assertTrue(file.toString(), file.isFile());
+          LOG.debug("log file " + file.getPath() + " size=" + file.getLen());
+        }
+      } else {
+        LOG.debug("no files under: " + logDir);
+      }
+    } catch (IOException e) {
+      LOG.warn("Unable to dump " + logDir, e);
+    }
+  }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/hbase/blob/e3db688a/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/TestProcedureRunQueues.java
----------------------------------------------------------------------
diff --git a/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/TestProcedureRunQueues.java b/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/TestProcedureRunQueues.java
new file mode 100644
index 0000000..c040be6
--- /dev/null
+++ b/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/TestProcedureRunQueues.java
@@ -0,0 +1,72 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase.procedure2;
+
+import org.apache.hadoop.hbase.testclassification.SmallTests;
+import org.apache.hadoop.hbase.testclassification.MasterTests;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import static org.junit.Assert.assertEquals;
+
+@Category({MasterTests.class, SmallTests.class})
+public class TestProcedureRunQueues {
+  class TestRunQueue implements ProcedureFairRunQueues.FairObject {
+    private final String name;
+    private final int priority;
+
+    public TestRunQueue(String name, int priority) {
+      this.name = name;
+      this.priority = priority;
+    }
+
+    @Override
+    public String toString() {
+      return name;
+    }
+
+    @Override
+    public boolean isAvailable() {
+      return true;
+    }
+
+    @Override
+    public int getPriority() {
+      return priority;
+    }
+  }
+
+  @Test
+  public void testFairQueues() throws Exception {
+    ProcedureFairRunQueues<String, TestRunQueue> fairq
+      = new ProcedureFairRunQueues<String, TestRunQueue>(1);
+    TestRunQueue a = fairq.add("A", new TestRunQueue("A", 1));
+    TestRunQueue b = fairq.add("B", new TestRunQueue("B", 1));
+    TestRunQueue m = fairq.add("M", new TestRunQueue("M", 2));
+
+    for (int i = 0; i < 3; ++i) {
+      assertEquals(a, fairq.poll());
+      assertEquals(b, fairq.poll());
+      assertEquals(m, fairq.poll());
+      assertEquals(m, fairq.poll());
+    }
+  }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/hbase/blob/e3db688a/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/store/TestProcedureStoreTracker.java
----------------------------------------------------------------------
diff --git a/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/store/TestProcedureStoreTracker.java b/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/store/TestProcedureStoreTracker.java
new file mode 100644
index 0000000..2c51782
--- /dev/null
+++ b/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/store/TestProcedureStoreTracker.java
@@ -0,0 +1,168 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase.procedure2.store;
+
+import java.io.InputStream;
+import java.io.OutputStream;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hbase.procedure2.Procedure;
+import org.apache.hadoop.hbase.testclassification.SmallTests;
+import org.apache.hadoop.hbase.testclassification.MasterTests;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+@Category({MasterTests.class, SmallTests.class})
+public class TestProcedureStoreTracker {
+  private static final Log LOG = LogFactory.getLog(TestProcedureStoreTracker.class);
+
+  static class TestProcedure extends Procedure<Void> {
+    public TestProcedure(long procId) {
+      setProcId(procId);
+    }
+
+    @Override
+    protected Procedure[] execute(Void env) { return null; }
+
+    @Override
+    protected void rollback(Void env) { /* no-op */ }
+
+    @Override
+    protected void abort(Void env) { /* no-op */ }
+
+    @Override
+    protected void serializeStateData(final OutputStream stream) { /* no-op */ }
+
+    @Override
+    protected void deserializeStateData(final InputStream stream) { /* no-op */ }
+  }
+
+  @Test
+  public void testSeqInsertAndDelete() {
+    ProcedureStoreTracker tracker = new ProcedureStoreTracker();
+    assertTrue(tracker.isEmpty());
+
+    final int MIN_PROC = 1;
+    final int MAX_PROC = 1 << 10;
+
+    // sequential insert
+    for (int i = MIN_PROC; i < MAX_PROC; ++i) {
+      tracker.insert(i);
+
+      // All the proc that we inserted should not be deleted
+      for (int j = MIN_PROC; j <= i; ++j) {
+        assertEquals(ProcedureStoreTracker.DeleteState.NO, tracker.isDeleted(j));
+      }
+      // All the proc that are not yet inserted should be result as deleted
+      for (int j = i + 1; j < MAX_PROC; ++j) {
+        assertTrue(tracker.isDeleted(j) != ProcedureStoreTracker.DeleteState.NO);
+      }
+    }
+
+    // sequential delete
+    for (int i = MIN_PROC; i < MAX_PROC; ++i) {
+      tracker.delete(i);
+
+      // All the proc that we deleted should be deleted
+      for (int j = MIN_PROC; j <= i; ++j) {
+        assertEquals(ProcedureStoreTracker.DeleteState.YES, tracker.isDeleted(j));
+      }
+      // All the proc that are not yet deleted should be result as not deleted
+      for (int j = i + 1; j < MAX_PROC; ++j) {
+        assertEquals(ProcedureStoreTracker.DeleteState.NO, tracker.isDeleted(j));
+      }
+    }
+    assertTrue(tracker.isEmpty());
+  }
+
+  @Test
+  public void testPartialTracker() {
+    ProcedureStoreTracker tracker = new ProcedureStoreTracker();
+    tracker.setPartialFlag(true);
+
+    // nothing in the tracker, the state is unknown
+    assertTrue(tracker.isEmpty());
+    assertEquals(ProcedureStoreTracker.DeleteState.MAYBE, tracker.isDeleted(1));
+    assertEquals(ProcedureStoreTracker.DeleteState.MAYBE, tracker.isDeleted(579));
+
+    // Mark 1 as deleted, now that is a known state
+    tracker.setDeleted(1, true);
+    tracker.dump();
+    assertEquals(ProcedureStoreTracker.DeleteState.YES, tracker.isDeleted(1));
+    assertEquals(ProcedureStoreTracker.DeleteState.MAYBE, tracker.isDeleted(2));
+    assertEquals(ProcedureStoreTracker.DeleteState.MAYBE, tracker.isDeleted(579));
+
+    // Mark 579 as non-deleted, now that is a known state
+    tracker.setDeleted(579, false);
+    assertEquals(ProcedureStoreTracker.DeleteState.YES, tracker.isDeleted(1));
+    assertEquals(ProcedureStoreTracker.DeleteState.MAYBE, tracker.isDeleted(2));
+    assertEquals(ProcedureStoreTracker.DeleteState.NO, tracker.isDeleted(579));
+    assertEquals(ProcedureStoreTracker.DeleteState.MAYBE, tracker.isDeleted(577));
+    assertEquals(ProcedureStoreTracker.DeleteState.MAYBE, tracker.isDeleted(580));
+  }
+
+  @Test
+  public void testBasicCRUD() {
+    ProcedureStoreTracker tracker = new ProcedureStoreTracker();
+    assertTrue(tracker.isEmpty());
+
+    Procedure[] procs = new TestProcedure[] {
+      new TestProcedure(1), new TestProcedure(2), new TestProcedure(3),
+      new TestProcedure(4), new TestProcedure(5), new TestProcedure(6),
+    };
+
+    tracker.insert(procs[0], null);
+    tracker.insert(procs[1], new Procedure[] { procs[2], procs[3], procs[4] });
+    assertFalse(tracker.isEmpty());
+    assertTrue(tracker.isUpdated());
+
+    tracker.resetUpdates();
+    assertFalse(tracker.isUpdated());
+
+    for (int i = 0; i < 4; ++i) {
+      tracker.update(procs[i]);
+      assertFalse(tracker.isEmpty());
+      assertFalse(tracker.isUpdated());
+    }
+
+    tracker.update(procs[4]);
+    assertFalse(tracker.isEmpty());
+    assertTrue(tracker.isUpdated());
+
+    tracker.update(procs[5]);
+    assertFalse(tracker.isEmpty());
+    assertTrue(tracker.isUpdated());
+
+    for (int i = 0; i < 5; ++i) {
+      tracker.delete(procs[i].getProcId());
+      assertFalse(tracker.isEmpty());
+      assertTrue(tracker.isUpdated());
+    }
+    tracker.delete(procs[5].getProcId());
+    assertTrue(tracker.isEmpty());
+  }
+}

http://git-wip-us.apache.org/repos/asf/hbase/blob/e3db688a/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/store/wal/TestWALProcedureStore.java
----------------------------------------------------------------------
diff --git a/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/store/wal/TestWALProcedureStore.java b/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/store/wal/TestWALProcedureStore.java
new file mode 100644
index 0000000..71899ad
--- /dev/null
+++ b/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/store/wal/TestWALProcedureStore.java
@@ -0,0 +1,267 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase.procedure2.store.wal;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Iterator;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HBaseCommonTestingUtility;
+import org.apache.hadoop.hbase.procedure2.Procedure;
+import org.apache.hadoop.hbase.procedure2.ProcedureTestingUtility;
+import org.apache.hadoop.hbase.procedure2.SequentialProcedure;
+import org.apache.hadoop.hbase.testclassification.SmallTests;
+import org.apache.hadoop.hbase.testclassification.MasterTests;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.io.IOUtils;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+@Category({MasterTests.class, SmallTests.class})
+public class TestWALProcedureStore {
+  private static final Log LOG = LogFactory.getLog(TestWALProcedureStore.class);
+
+  private static final int PROCEDURE_STORE_SLOTS = 1;
+  private static final Procedure NULL_PROC = null;
+
+  private WALProcedureStore procStore;
+
+  private HBaseCommonTestingUtility htu;
+  private FileSystem fs;
+  private Path testDir;
+  private Path logDir;
+
+  @Before
+  public void setUp() throws IOException {
+    htu = new HBaseCommonTestingUtility();
+    testDir = htu.getDataTestDir();
+    fs = testDir.getFileSystem(htu.getConfiguration());
+    assertTrue(testDir.depth() > 1);
+
+    logDir = new Path(testDir, "proc-logs");
+    procStore = ProcedureTestingUtility.createWalStore(htu.getConfiguration(), fs, logDir);
+    procStore.start(PROCEDURE_STORE_SLOTS);
+    procStore.recoverLease();
+  }
+
+  @After
+  public void tearDown() throws IOException {
+    procStore.stop();
+    fs.delete(logDir, true);
+  }
+
+  private Iterator<Procedure> storeRestart() throws Exception {
+    procStore.stop();
+    procStore.start(PROCEDURE_STORE_SLOTS);
+    procStore.recoverLease();
+    return procStore.load();
+  }
+
+  @Test
+  public void testEmptyLogLoad() throws Exception {
+    Iterator<Procedure> loader = storeRestart();
+    assertEquals(0, countProcedures(loader));
+  }
+
+  @Test
+  public void testLoad() throws Exception {
+    Set<Long> procIds = new HashSet<>();
+
+    // Insert something in the log
+    Procedure proc1 = new TestSequentialProcedure();
+    procIds.add(proc1.getProcId());
+    procStore.insert(proc1, null);
+
+    Procedure proc2 = new TestSequentialProcedure();
+    Procedure[] child2 = new Procedure[2];
+    child2[0] = new TestSequentialProcedure();
+    child2[1] = new TestSequentialProcedure();
+
+    procIds.add(proc2.getProcId());
+    procIds.add(child2[0].getProcId());
+    procIds.add(child2[1].getProcId());
+    procStore.insert(proc2, child2);
+
+    // Verify that everything is there
+    verifyProcIdsOnRestart(procIds);
+
+    // Update and delete something
+    procStore.update(proc1);
+    procStore.update(child2[1]);
+    procStore.delete(child2[1].getProcId());
+    procIds.remove(child2[1].getProcId());
+
+    // Verify that everything is there
+    verifyProcIdsOnRestart(procIds);
+
+    // Remove 4 byte from the trailers
+    procStore.stop();
+    FileStatus[] logs = fs.listStatus(logDir);
+    assertEquals(3, logs.length);
+    for (int i = 0; i < logs.length; ++i) {
+      corruptLog(logs[i], 4);
+    }
+    verifyProcIdsOnRestart(procIds);
+  }
+
+  @Test
+  public void testCorruptedTrailer() throws Exception {
+    // Insert something
+    for (int i = 0; i < 100; ++i) {
+      procStore.insert(new TestSequentialProcedure(), null);
+    }
+
+    // Stop the store
+    procStore.stop();
+
+    // Remove 4 byte from the trailer
+    FileStatus[] logs = fs.listStatus(logDir);
+    assertEquals(1, logs.length);
+    corruptLog(logs[0], 4);
+
+    int count = countProcedures(storeRestart());
+    assertEquals(100, count);
+  }
+
+  @Test
+  public void testCorruptedEntries() throws Exception {
+    // Insert something
+    for (int i = 0; i < 100; ++i) {
+      procStore.insert(new TestSequentialProcedure(), null);
+    }
+
+    // Stop the store
+    procStore.stop();
+
+    // Remove some byte from the log
+    // (enough to cut the trailer and corrupt some entries)
+    FileStatus[] logs = fs.listStatus(logDir);
+    assertEquals(1, logs.length);
+    corruptLog(logs[0], 1823);
+
+    int count = countProcedures(storeRestart());
+    assertTrue(procStore.getCorruptedLogs() != null);
+    assertEquals(1, procStore.getCorruptedLogs().size());
+    assertEquals(85, count);
+  }
+
+  private void corruptLog(final FileStatus logFile, final long dropBytes)
+      throws IOException {
+    assertTrue(logFile.getLen() > dropBytes);
+    LOG.debug("corrupt log " + logFile.getPath() +
+              " size=" + logFile.getLen() + " drop=" + dropBytes);
+    Path tmpPath = new Path(testDir, "corrupted.log");
+    InputStream in = fs.open(logFile.getPath());
+    OutputStream out =  fs.create(tmpPath);
+    IOUtils.copyBytes(in, out, logFile.getLen() - dropBytes, true);
+    fs.rename(tmpPath, logFile.getPath());
+  }
+
+  private void verifyProcIdsOnRestart(final Set<Long> procIds) throws Exception {
+    int count = 0;
+    Iterator<Procedure> loader = storeRestart();
+    while (loader.hasNext()) {
+      Procedure proc = loader.next();
+      LOG.debug("loading procId=" + proc.getProcId());
+      assertTrue("procId=" + proc.getProcId() + " unexpected", procIds.contains(proc.getProcId()));
+      count++;
+    }
+    assertEquals(procIds.size(), count);
+  }
+
+  private void assertIsEmpty(Iterator<Procedure> iterator) {
+    assertEquals(0, countProcedures(iterator));
+  }
+
+  private int countProcedures(Iterator<Procedure> iterator) {
+    int count = 0;
+    while (iterator.hasNext()) {
+      Procedure proc = iterator.next();
+      LOG.trace("loading procId=" + proc.getProcId());
+      count++;
+    }
+    return count;
+  }
+
+  private void assertEmptyLogDir() {
+    try {
+      FileStatus[] status = fs.listStatus(logDir);
+      assertTrue("expected empty state-log dir", status == null || status.length == 0);
+    } catch (FileNotFoundException e) {
+      fail("expected the state-log dir to be present: " + logDir);
+    } catch (IOException e) {
+      fail("got en exception on state-log dir list: " + e.getMessage());
+    }
+  }
+
+  public static class TestSequentialProcedure extends SequentialProcedure<Void> {
+    private static long seqid = 0;
+
+    public TestSequentialProcedure() {
+      setProcId(++seqid);
+    }
+
+    @Override
+    protected Procedure[] execute(Void env) { return null; }
+
+    @Override
+    protected void rollback(Void env) { }
+
+    @Override
+    protected void abort(Void env) { }
+
+    @Override
+    protected void serializeStateData(final OutputStream stream) throws IOException {
+      long procId = getProcId();
+      if (procId % 2 == 0) {
+        stream.write(Bytes.toBytes(procId));
+      }
+    }
+
+    @Override
+    protected void deserializeStateData(InputStream stream) throws IOException {
+      long procId = getProcId();
+      if (procId % 2 == 0) {
+        byte[] bProcId = new byte[8];
+        assertEquals(8, stream.read(bProcId));
+        assertEquals(procId, Bytes.toLong(bProcId));
+      } else {
+        assertEquals(0, stream.available());
+      }
+    }
+  }
+}
\ No newline at end of file


Mime
View raw message