001/**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.procedure2.store.wal;
019
020import java.io.FileNotFoundException;
021import java.io.IOException;
022import java.util.ArrayList;
023import java.util.Arrays;
024import java.util.Comparator;
025import java.util.HashSet;
026import java.util.Iterator;
027import java.util.LinkedList;
028import java.util.Set;
029import java.util.concurrent.LinkedTransferQueue;
030import java.util.concurrent.TimeUnit;
031import java.util.concurrent.atomic.AtomicBoolean;
032import java.util.concurrent.atomic.AtomicLong;
033import java.util.concurrent.atomic.AtomicReference;
034import java.util.concurrent.locks.Condition;
035import java.util.concurrent.locks.ReentrantLock;
036import org.apache.hadoop.conf.Configuration;
037import org.apache.hadoop.fs.FSDataOutputStream;
038import org.apache.hadoop.fs.FSError;
039import org.apache.hadoop.fs.FileAlreadyExistsException;
040import org.apache.hadoop.fs.FileStatus;
041import org.apache.hadoop.fs.FileSystem;
042import org.apache.hadoop.fs.Path;
043import org.apache.hadoop.fs.PathFilter;
044import org.apache.hadoop.fs.StreamCapabilities;
045import org.apache.hadoop.hbase.HBaseConfiguration;
046import org.apache.hadoop.hbase.HConstants;
047import org.apache.hadoop.hbase.log.HBaseMarkers;
048import org.apache.hadoop.hbase.procedure2.Procedure;
049import org.apache.hadoop.hbase.procedure2.ProcedureExecutor;
050import org.apache.hadoop.hbase.procedure2.store.LeaseRecovery;
051import org.apache.hadoop.hbase.procedure2.store.ProcedureStore;
052import org.apache.hadoop.hbase.procedure2.store.ProcedureStoreBase;
053import org.apache.hadoop.hbase.procedure2.util.ByteSlot;
054import org.apache.hadoop.hbase.procedure2.util.StringUtils;
055import org.apache.hadoop.hbase.util.CommonFSUtils;
056import org.apache.hadoop.hbase.util.Threads;
057import org.apache.hadoop.ipc.RemoteException;
058import org.apache.yetus.audience.InterfaceAudience;
059import org.slf4j.Logger;
060import org.slf4j.LoggerFactory;
061
062import org.apache.hbase.thirdparty.org.apache.commons.collections4.queue.CircularFifoQueue;
063
064import org.apache.hadoop.hbase.shaded.protobuf.generated.ProcedureProtos.ProcedureWALHeader;
065
066/**
067 * WAL implementation of the ProcedureStore.
068 * <p/>
069 * When starting, the upper layer will first call {@link #start(int)}, then {@link #recoverLease()},
070 * then {@link #load(ProcedureLoader)}.
071 * <p/>
072 * In {@link #recoverLease()}, we will get the lease by closing all the existing wal files(by
073 * calling recoverFileLease), and creating a new wal writer. And we will also get the list of all
074 * the old wal files.
075 * <p/>
076 * FIXME: notice that the current recover lease implementation is problematic, it can not deal with
077 * the races if there are two master both wants to acquire the lease...
078 * <p/>
079 * In {@link #load(ProcedureLoader)} method, we will load all the active procedures. See the
080 * comments of this method for more details.
081 * <p/>
082 * The actual logging way is a bit like our FileSystem based WAL implementation as RS side. There is
083 * a {@link #slots}, which is more like the ring buffer, and in the insert, update and delete
084 * methods we will put thing into the {@link #slots} and wait. And there is a background sync
085 * thread(see the {@link #syncLoop()} method) which get data from the {@link #slots} and write them
086 * to the FileSystem, and notify the caller that we have finished.
087 * <p/>
088 * TODO: try using disruptor to increase performance and simplify the logic?
089 * <p/>
090 * The {@link #storeTracker} keeps track of the modified procedures in the newest wal file, which is
091 * also the one being written currently. And the deleted bits in it are for all the procedures, not
092 * only the ones in the newest wal file. And when rolling a log, we will first store it in the
093 * trailer of the current wal file, and then reset its modified bits, so that it can start to track
094 * the modified procedures for the new wal file.
095 * <p/>
096 * The {@link #holdingCleanupTracker} is used to test whether we are safe to delete the oldest wal
097 * file. When there are log rolling and there are more than 1 wal files, we will make use of it. It
098 * will first be initialized to the oldest file's tracker(which is stored in the trailer), using the
099 * method {@link ProcedureStoreTracker#resetTo(ProcedureStoreTracker, boolean)}, and then merge it
100 * with the tracker of every newer wal files, using the
101 * {@link ProcedureStoreTracker#setDeletedIfModifiedInBoth(ProcedureStoreTracker)}.
102 * If we find out
103 * that all the modified procedures for the oldest wal file are modified or deleted in newer wal
104 * files, then we can delete it. This is because that, every time we call
105 * {@link ProcedureStore#insert(Procedure[])} or {@link ProcedureStore#update(Procedure)}, we will
106 * persist the full state of a Procedure, so the earlier wal records for this procedure can all be
107 * deleted.
108 * @see ProcedureWALPrettyPrinter for printing content of a single WAL.
109 * @see #main(String[]) to parse a directory of MasterWALProcs.
110 * @deprecated Since 2.3.0, will be removed in 4.0.0. Keep here only for rolling upgrading, now we
111 *             use the new region based procedure store.
112 */
113@Deprecated
114@InterfaceAudience.Private
115public class WALProcedureStore extends ProcedureStoreBase {
116  private static final Logger LOG = LoggerFactory.getLogger(WALProcedureStore.class);
117  public static final String LOG_PREFIX = "pv2-";
118  /** Used to construct the name of the log directory for master procedures */
119  public static final String MASTER_PROCEDURE_LOGDIR = "MasterProcWALs";
120
121
122  public static final String WAL_COUNT_WARN_THRESHOLD_CONF_KEY =
123    "hbase.procedure.store.wal.warn.threshold";
124  private static final int DEFAULT_WAL_COUNT_WARN_THRESHOLD = 10;
125
126  public static final String EXEC_WAL_CLEANUP_ON_LOAD_CONF_KEY =
127    "hbase.procedure.store.wal.exec.cleanup.on.load";
128  private static final boolean DEFAULT_EXEC_WAL_CLEANUP_ON_LOAD_CONF_KEY = true;
129
130  public static final String MAX_RETRIES_BEFORE_ROLL_CONF_KEY =
131    "hbase.procedure.store.wal.max.retries.before.roll";
132  private static final int DEFAULT_MAX_RETRIES_BEFORE_ROLL = 3;
133
134  public static final String WAIT_BEFORE_ROLL_CONF_KEY =
135    "hbase.procedure.store.wal.wait.before.roll";
136  private static final int DEFAULT_WAIT_BEFORE_ROLL = 500;
137
138  public static final String ROLL_RETRIES_CONF_KEY =
139    "hbase.procedure.store.wal.max.roll.retries";
140  private static final int DEFAULT_ROLL_RETRIES = 3;
141
142  public static final String MAX_SYNC_FAILURE_ROLL_CONF_KEY =
143    "hbase.procedure.store.wal.sync.failure.roll.max";
144  private static final int DEFAULT_MAX_SYNC_FAILURE_ROLL = 3;
145
146  public static final String PERIODIC_ROLL_CONF_KEY =
147    "hbase.procedure.store.wal.periodic.roll.msec";
148  private static final int DEFAULT_PERIODIC_ROLL = 60 * 60 * 1000; // 1h
149
150  public static final String SYNC_WAIT_MSEC_CONF_KEY = "hbase.procedure.store.wal.sync.wait.msec";
151  private static final int DEFAULT_SYNC_WAIT_MSEC = 100;
152
153  public static final String USE_HSYNC_CONF_KEY = "hbase.procedure.store.wal.use.hsync";
154  private static final boolean DEFAULT_USE_HSYNC = true;
155
156  public static final String ROLL_THRESHOLD_CONF_KEY = "hbase.procedure.store.wal.roll.threshold";
157  private static final long DEFAULT_ROLL_THRESHOLD = 32 * 1024 * 1024; // 32M
158
159  public static final String STORE_WAL_SYNC_STATS_COUNT =
160      "hbase.procedure.store.wal.sync.stats.count";
161  private static final int DEFAULT_SYNC_STATS_COUNT = 10;
162
163  private final LinkedList<ProcedureWALFile> logs = new LinkedList<>();
164  private final ProcedureStoreTracker holdingCleanupTracker = new ProcedureStoreTracker();
165  private final ProcedureStoreTracker storeTracker = new ProcedureStoreTracker();
166  private final ReentrantLock lock = new ReentrantLock();
167  private final Condition waitCond = lock.newCondition();
168  private final Condition slotCond = lock.newCondition();
169  private final Condition syncCond = lock.newCondition();
170
171  private final LeaseRecovery leaseRecovery;
172  private final Configuration conf;
173  private final FileSystem fs;
174  private final Path walDir;
175  private final Path walArchiveDir;
176  private final boolean enforceStreamCapability;
177
178  private final AtomicReference<Throwable> syncException = new AtomicReference<>();
179  private final AtomicBoolean loading = new AtomicBoolean(true);
180  private final AtomicBoolean inSync = new AtomicBoolean(false);
181  private final AtomicLong totalSynced = new AtomicLong(0);
182  private final AtomicLong lastRollTs = new AtomicLong(0);
183  private final AtomicLong syncId = new AtomicLong(0);
184
185  private LinkedTransferQueue<ByteSlot> slotsCache = null;
186  private Set<ProcedureWALFile> corruptedLogs = null;
187  private FSDataOutputStream stream = null;
188  private int runningProcCount = 1;
189  private long flushLogId = 0;
190  private int syncMaxSlot = 1;
191  private int slotIndex = 0;
192  private Thread syncThread;
193  private ByteSlot[] slots;
194
195  private int walCountWarnThreshold;
196  private int maxRetriesBeforeRoll;
197  private int maxSyncFailureRoll;
198  private int waitBeforeRoll;
199  private int rollRetries;
200  private int periodicRollMsec;
201  private long rollThreshold;
202  private boolean useHsync;
203  private int syncWaitMsec;
204
205  // Variables used for UI display
206  private CircularFifoQueue<SyncMetrics> syncMetricsQueue;
207
208  public static class SyncMetrics {
209    private long timestamp;
210    private long syncWaitMs;
211    private long totalSyncedBytes;
212    private int syncedEntries;
213    private float syncedPerSec;
214
215    public long getTimestamp() {
216      return timestamp;
217    }
218
219    public long getSyncWaitMs() {
220      return syncWaitMs;
221    }
222
223    public long getTotalSyncedBytes() {
224      return totalSyncedBytes;
225    }
226
227    public long getSyncedEntries() {
228      return syncedEntries;
229    }
230
231    public float getSyncedPerSec() {
232      return syncedPerSec;
233    }
234  }
235
236  public WALProcedureStore(Configuration conf, LeaseRecovery leaseRecovery) throws IOException {
237    this(conf, new Path(CommonFSUtils.getWALRootDir(conf), MASTER_PROCEDURE_LOGDIR),
238      new Path(CommonFSUtils.getWALRootDir(conf), HConstants.HREGION_OLDLOGDIR_NAME),
239      leaseRecovery);
240  }
241
242  public WALProcedureStore(final Configuration conf, final Path walDir, final Path walArchiveDir,
243      final LeaseRecovery leaseRecovery) throws IOException {
244    this.conf = conf;
245    this.leaseRecovery = leaseRecovery;
246    this.walDir = walDir;
247    this.walArchiveDir = walArchiveDir;
248    this.fs = CommonFSUtils.getWALFileSystem(conf);
249    this.enforceStreamCapability = conf.getBoolean(CommonFSUtils.UNSAFE_STREAM_CAPABILITY_ENFORCE,
250        true);
251
252    // Create the log directory for the procedure store
253    if (!fs.exists(walDir)) {
254      if (!fs.mkdirs(walDir)) {
255        throw new IOException("Unable to mkdir " + walDir);
256      }
257    }
258    // Now that it exists, set the log policy
259    String storagePolicy =
260        conf.get(HConstants.WAL_STORAGE_POLICY, HConstants.DEFAULT_WAL_STORAGE_POLICY);
261    CommonFSUtils.setStoragePolicy(fs, walDir, storagePolicy);
262
263    // Create archive dir up front. Rename won't work w/o it up on HDFS.
264    if (this.walArchiveDir != null && !this.fs.exists(this.walArchiveDir)) {
265      if (this.fs.mkdirs(this.walArchiveDir)) {
266        LOG.debug("Created Procedure Store WAL archive dir {}", this.walArchiveDir);
267      } else {
268        LOG.warn("Failed create of {}", this.walArchiveDir);
269      }
270    }
271  }
272
273  @Override
274  public void start(int numSlots) throws IOException {
275    if (!setRunning(true)) {
276      return;
277    }
278
279    // Init buffer slots
280    loading.set(true);
281    runningProcCount = numSlots;
282    syncMaxSlot = numSlots;
283    slots = new ByteSlot[numSlots];
284    slotsCache = new LinkedTransferQueue<>();
285    while (slotsCache.size() < numSlots) {
286      slotsCache.offer(new ByteSlot());
287    }
288
289    // Tunings
290    walCountWarnThreshold =
291      conf.getInt(WAL_COUNT_WARN_THRESHOLD_CONF_KEY, DEFAULT_WAL_COUNT_WARN_THRESHOLD);
292    maxRetriesBeforeRoll =
293      conf.getInt(MAX_RETRIES_BEFORE_ROLL_CONF_KEY, DEFAULT_MAX_RETRIES_BEFORE_ROLL);
294    maxSyncFailureRoll = conf.getInt(MAX_SYNC_FAILURE_ROLL_CONF_KEY, DEFAULT_MAX_SYNC_FAILURE_ROLL);
295    waitBeforeRoll = conf.getInt(WAIT_BEFORE_ROLL_CONF_KEY, DEFAULT_WAIT_BEFORE_ROLL);
296    rollRetries = conf.getInt(ROLL_RETRIES_CONF_KEY, DEFAULT_ROLL_RETRIES);
297    rollThreshold = conf.getLong(ROLL_THRESHOLD_CONF_KEY, DEFAULT_ROLL_THRESHOLD);
298    periodicRollMsec = conf.getInt(PERIODIC_ROLL_CONF_KEY, DEFAULT_PERIODIC_ROLL);
299    syncWaitMsec = conf.getInt(SYNC_WAIT_MSEC_CONF_KEY, DEFAULT_SYNC_WAIT_MSEC);
300    useHsync = conf.getBoolean(USE_HSYNC_CONF_KEY, DEFAULT_USE_HSYNC);
301
302    // WebUI
303    syncMetricsQueue = new CircularFifoQueue<>(
304      conf.getInt(STORE_WAL_SYNC_STATS_COUNT, DEFAULT_SYNC_STATS_COUNT));
305
306    // Init sync thread
307    syncThread = new Thread("WALProcedureStoreSyncThread") {
308      @Override
309      public void run() {
310        try {
311          syncLoop();
312        } catch (Throwable e) {
313          LOG.error("Got an exception from the sync-loop", e);
314          if (!isSyncAborted()) {
315            sendAbortProcessSignal();
316          }
317        }
318      }
319    };
320    syncThread.start();
321  }
322
323  @Override
324  public void stop(final boolean abort) {
325    if (!setRunning(false)) {
326      return;
327    }
328
329    LOG.info("Stopping the WAL Procedure Store, isAbort=" + abort +
330      (isSyncAborted() ? " (self aborting)" : ""));
331    sendStopSignal();
332    if (!isSyncAborted()) {
333      try {
334        while (syncThread.isAlive()) {
335          sendStopSignal();
336          syncThread.join(250);
337        }
338      } catch (InterruptedException e) {
339        LOG.warn("join interrupted", e);
340        Thread.currentThread().interrupt();
341      }
342    }
343
344    // Close the writer
345    closeCurrentLogStream(abort);
346
347    // Close the old logs
348    // they should be already closed, this is just in case the load fails
349    // and we call start() and then stop()
350    for (ProcedureWALFile log: logs) {
351      log.close();
352    }
353    logs.clear();
354    loading.set(true);
355  }
356
357  private void sendStopSignal() {
358    if (lock.tryLock()) {
359      try {
360        waitCond.signalAll();
361        syncCond.signalAll();
362      } finally {
363        lock.unlock();
364      }
365    }
366  }
367
368  @Override
369  public int getNumThreads() {
370    return slots == null ? 0 : slots.length;
371  }
372
373  @Override
374  public int setRunningProcedureCount(final int count) {
375    this.runningProcCount = count > 0 ? Math.min(count, slots.length) : slots.length;
376    return this.runningProcCount;
377  }
378
379  public ProcedureStoreTracker getStoreTracker() {
380    return storeTracker;
381  }
382
383  public ArrayList<ProcedureWALFile> getActiveLogs() {
384    lock.lock();
385    try {
386      return new ArrayList<>(logs);
387    } finally {
388      lock.unlock();
389    }
390  }
391
392  public Set<ProcedureWALFile> getCorruptedLogs() {
393    return corruptedLogs;
394  }
395
396  @Override
397  public void recoverLease() throws IOException {
398    lock.lock();
399    try {
400      LOG.debug("Starting WAL Procedure Store lease recovery");
401      boolean afterFirstAttempt = false;
402      while (isRunning()) {
403        // Don't sleep before first attempt
404        if (afterFirstAttempt) {
405          LOG.trace("Sleep {} ms after first lease recovery attempt.",
406              waitBeforeRoll);
407          Threads.sleepWithoutInterrupt(waitBeforeRoll);
408        } else {
409          afterFirstAttempt = true;
410        }
411        FileStatus[] oldLogs = getLogFiles();
412        // Get Log-MaxID and recover lease on old logs
413        try {
414          flushLogId = initOldLogs(oldLogs);
415        } catch (FileNotFoundException e) {
416          LOG.warn("Someone else is active and deleted logs. retrying.", e);
417          continue;
418        }
419
420        // Create new state-log
421        if (!rollWriter(flushLogId + 1)) {
422          // someone else has already created this log
423          LOG.debug("Someone else has already created log {}. Retrying.", flushLogId);
424          continue;
425        }
426
427        // We have the lease on the log
428        oldLogs = getLogFiles();
429        if (getMaxLogId(oldLogs) > flushLogId) {
430          LOG.debug("Someone else created new logs. Expected maxLogId < {}", flushLogId);
431          logs.getLast().removeFile(this.walArchiveDir);
432          continue;
433        }
434
435        LOG.debug("Lease acquired for flushLogId={}", flushLogId);
436        break;
437      }
438    } finally {
439      lock.unlock();
440    }
441  }
442
443  @Override
444  public void load(ProcedureLoader loader) throws IOException {
445    lock.lock();
446    try {
447      if (logs.isEmpty()) {
448        throw new IllegalStateException("recoverLease() must be called before loading data");
449      }
450
451      // Nothing to do, If we have only the current log.
452      if (logs.size() == 1) {
453        LOG.debug("No state logs to replay.");
454        loader.setMaxProcId(0);
455        loading.set(false);
456        return;
457      }
458
459      // Load the old logs
460      Iterator<ProcedureWALFile> it = logs.descendingIterator();
461      it.next(); // Skip the current log
462
463      ProcedureWALFormat.load(it, storeTracker, new ProcedureWALFormat.Loader() {
464
465        @Override
466        public void setMaxProcId(long maxProcId) {
467          loader.setMaxProcId(maxProcId);
468        }
469
470        @Override
471        public void load(ProcedureIterator procIter) throws IOException {
472          loader.load(procIter);
473        }
474
475        @Override
476        public void handleCorrupted(ProcedureIterator procIter) throws IOException {
477          loader.handleCorrupted(procIter);
478        }
479
480        @Override
481        public void markCorruptedWAL(ProcedureWALFile log, IOException e) {
482          if (corruptedLogs == null) {
483            corruptedLogs = new HashSet<>();
484          }
485          corruptedLogs.add(log);
486          // TODO: sideline corrupted log
487        }
488      });
489      // if we fail when loading, we should prevent persisting the storeTracker later in the stop
490      // method. As it may happen that, we have finished constructing the modified and deleted bits,
491      // but before we call resetModified, we fail, then if we persist the storeTracker then when
492      // restarting, we will consider that all procedures have been included in this file and delete
493      // all the previous files. Obviously this not correct. So here we will only set loading to
494      // false when we successfully loaded all the procedures, and when closing we will skip
495      // persisting the store tracker. And also, this will prevent the sync thread to do
496      // periodicRoll, where we may also clean old logs.
497      loading.set(false);
498      // try to cleanup inactive wals and complete the operation
499      buildHoldingCleanupTracker();
500      tryCleanupLogsOnLoad();
501    } finally {
502      lock.unlock();
503    }
504  }
505
506  private void tryCleanupLogsOnLoad() {
507    // nothing to cleanup.
508    if (logs.size() <= 1) {
509      return;
510    }
511
512    // the config says to not cleanup wals on load.
513    if (!conf.getBoolean(EXEC_WAL_CLEANUP_ON_LOAD_CONF_KEY,
514      DEFAULT_EXEC_WAL_CLEANUP_ON_LOAD_CONF_KEY)) {
515      LOG.debug("WALs cleanup on load is not enabled: " + getActiveLogs());
516      return;
517    }
518
519    try {
520      periodicRoll();
521    } catch (IOException e) {
522      LOG.warn("Unable to cleanup logs on load: " + e.getMessage(), e);
523    }
524  }
525
526  @Override
527  public void insert(Procedure<?> proc, Procedure<?>[] subprocs) {
528    if (LOG.isTraceEnabled()) {
529      LOG.trace("Insert " + proc + ", subproc=" + Arrays.toString(subprocs));
530    }
531
532    ByteSlot slot = acquireSlot();
533    try {
534      // Serialize the insert
535      long[] subProcIds = null;
536      if (subprocs != null) {
537        ProcedureWALFormat.writeInsert(slot, proc, subprocs);
538        subProcIds = new long[subprocs.length];
539        for (int i = 0; i < subprocs.length; ++i) {
540          subProcIds[i] = subprocs[i].getProcId();
541        }
542      } else {
543        assert !proc.hasParent();
544        ProcedureWALFormat.writeInsert(slot, proc);
545      }
546
547      // Push the transaction data and wait until it is persisted
548      pushData(PushType.INSERT, slot, proc.getProcId(), subProcIds);
549    } catch (IOException e) {
550      // We are not able to serialize the procedure.
551      // this is a code error, and we are not able to go on.
552      LOG.error(HBaseMarkers.FATAL, "Unable to serialize one of the procedure: proc=" +
553          proc + ", subprocs=" + Arrays.toString(subprocs), e);
554      throw new RuntimeException(e);
555    } finally {
556      releaseSlot(slot);
557    }
558  }
559
560  @Override
561  public void insert(Procedure<?>[] procs) {
562    if (LOG.isTraceEnabled()) {
563      LOG.trace("Insert " + Arrays.toString(procs));
564    }
565
566    ByteSlot slot = acquireSlot();
567    try {
568      // Serialize the insert
569      long[] procIds = new long[procs.length];
570      for (int i = 0; i < procs.length; ++i) {
571        assert !procs[i].hasParent();
572        procIds[i] = procs[i].getProcId();
573        ProcedureWALFormat.writeInsert(slot, procs[i]);
574      }
575
576      // Push the transaction data and wait until it is persisted
577      pushData(PushType.INSERT, slot, Procedure.NO_PROC_ID, procIds);
578    } catch (IOException e) {
579      // We are not able to serialize the procedure.
580      // this is a code error, and we are not able to go on.
581      LOG.error(HBaseMarkers.FATAL, "Unable to serialize one of the procedure: " +
582          Arrays.toString(procs), e);
583      throw new RuntimeException(e);
584    } finally {
585      releaseSlot(slot);
586    }
587  }
588
589  @Override
590  public void update(Procedure<?> proc) {
591    if (LOG.isTraceEnabled()) {
592      LOG.trace("Update " + proc);
593    }
594
595    ByteSlot slot = acquireSlot();
596    try {
597      // Serialize the update
598      ProcedureWALFormat.writeUpdate(slot, proc);
599
600      // Push the transaction data and wait until it is persisted
601      pushData(PushType.UPDATE, slot, proc.getProcId(), null);
602    } catch (IOException e) {
603      // We are not able to serialize the procedure.
604      // this is a code error, and we are not able to go on.
605      LOG.error(HBaseMarkers.FATAL, "Unable to serialize the procedure: " + proc, e);
606      throw new RuntimeException(e);
607    } finally {
608      releaseSlot(slot);
609    }
610  }
611
612  @Override
613  public void delete(long procId) {
614    LOG.trace("Delete {}", procId);
615    ByteSlot slot = acquireSlot();
616    try {
617      // Serialize the delete
618      ProcedureWALFormat.writeDelete(slot, procId);
619
620      // Push the transaction data and wait until it is persisted
621      pushData(PushType.DELETE, slot, procId, null);
622    } catch (IOException e) {
623      // We are not able to serialize the procedure.
624      // this is a code error, and we are not able to go on.
625      LOG.error(HBaseMarkers.FATAL, "Unable to serialize the procedure: " + procId, e);
626      throw new RuntimeException(e);
627    } finally {
628      releaseSlot(slot);
629    }
630  }
631
632  @Override
633  public void delete(Procedure<?> proc, long[] subProcIds) {
634    assert proc != null : "expected a non-null procedure";
635    assert subProcIds != null && subProcIds.length > 0 : "expected subProcIds";
636    if (LOG.isTraceEnabled()) {
637      LOG.trace("Update " + proc + " and Delete " + Arrays.toString(subProcIds));
638    }
639
640    ByteSlot slot = acquireSlot();
641    try {
642      // Serialize the delete
643      ProcedureWALFormat.writeDelete(slot, proc, subProcIds);
644
645      // Push the transaction data and wait until it is persisted
646      pushData(PushType.DELETE, slot, proc.getProcId(), subProcIds);
647    } catch (IOException e) {
648      // We are not able to serialize the procedure.
649      // this is a code error, and we are not able to go on.
650      LOG.error(HBaseMarkers.FATAL, "Unable to serialize the procedure: " + proc, e);
651      throw new RuntimeException(e);
652    } finally {
653      releaseSlot(slot);
654    }
655  }
656
657  @Override
658  public void delete(final long[] procIds, final int offset, final int count) {
659    if (count == 0) {
660      return;
661    }
662
663    if (offset == 0 && count == procIds.length) {
664      delete(procIds);
665    } else if (count == 1) {
666      delete(procIds[offset]);
667    } else {
668      delete(Arrays.copyOfRange(procIds, offset, offset + count));
669    }
670  }
671
672  private void delete(long[] procIds) {
673    if (LOG.isTraceEnabled()) {
674      LOG.trace("Delete " + Arrays.toString(procIds));
675    }
676
677    final ByteSlot slot = acquireSlot();
678    try {
679      // Serialize the delete
680      for (int i = 0; i < procIds.length; ++i) {
681        ProcedureWALFormat.writeDelete(slot, procIds[i]);
682      }
683
684      // Push the transaction data and wait until it is persisted
685      pushData(PushType.DELETE, slot, Procedure.NO_PROC_ID, procIds);
686    } catch (IOException e) {
687      // We are not able to serialize the procedure.
688      // this is a code error, and we are not able to go on.
689      LOG.error("Unable to serialize the procedures: " + Arrays.toString(procIds), e);
690      throw new RuntimeException(e);
691    } finally {
692      releaseSlot(slot);
693    }
694  }
695
696  private ByteSlot acquireSlot() {
697    ByteSlot slot = slotsCache.poll();
698    return slot != null ? slot : new ByteSlot();
699  }
700
701  private void releaseSlot(final ByteSlot slot) {
702    slot.reset();
703    slotsCache.offer(slot);
704  }
705
706  private enum PushType { INSERT, UPDATE, DELETE }
707
708  private long pushData(final PushType type, final ByteSlot slot,
709      final long procId, final long[] subProcIds) {
710    if (!isRunning()) {
711      throw new RuntimeException("the store must be running before inserting data");
712    }
713    if (logs.isEmpty()) {
714      throw new RuntimeException("recoverLease() must be called before inserting data");
715    }
716
717    long logId = -1;
718    lock.lock();
719    try {
720      // Wait for the sync to be completed
721      while (true) {
722        if (!isRunning()) {
723          throw new RuntimeException("store no longer running");
724        } else if (isSyncAborted()) {
725          throw new RuntimeException("sync aborted", syncException.get());
726        } else if (inSync.get()) {
727          syncCond.await();
728        } else if (slotIndex >= syncMaxSlot) {
729          slotCond.signal();
730          syncCond.await();
731        } else {
732          break;
733        }
734      }
735
736      final long pushSyncId = syncId.get();
737      updateStoreTracker(type, procId, subProcIds);
738      slots[slotIndex++] = slot;
739      logId = flushLogId;
740
741      // Notify that there is new data
742      if (slotIndex == 1) {
743        waitCond.signal();
744      }
745
746      // Notify that the slots are full
747      if (slotIndex == syncMaxSlot) {
748        waitCond.signal();
749        slotCond.signal();
750      }
751
752      while (pushSyncId == syncId.get() && isRunning()) {
753        syncCond.await();
754      }
755    } catch (InterruptedException e) {
756      Thread.currentThread().interrupt();
757      sendAbortProcessSignal();
758      throw new RuntimeException(e);
759    } finally {
760      lock.unlock();
761      if (isSyncAborted()) {
762        throw new RuntimeException("sync aborted", syncException.get());
763      }
764    }
765    return logId;
766  }
767
768  private void updateStoreTracker(final PushType type,
769      final long procId, final long[] subProcIds) {
770    switch (type) {
771      case INSERT:
772        if (subProcIds == null) {
773          storeTracker.insert(procId);
774        } else if (procId == Procedure.NO_PROC_ID) {
775          storeTracker.insert(subProcIds);
776        } else {
777          storeTracker.insert(procId, subProcIds);
778          holdingCleanupTracker.setDeletedIfModified(procId);
779        }
780        break;
781      case UPDATE:
782        storeTracker.update(procId);
783        holdingCleanupTracker.setDeletedIfModified(procId);
784        break;
785      case DELETE:
786        if (subProcIds != null && subProcIds.length > 0) {
787          storeTracker.delete(subProcIds);
788          holdingCleanupTracker.setDeletedIfModified(subProcIds);
789        } else {
790          storeTracker.delete(procId);
791          holdingCleanupTracker.setDeletedIfModified(procId);
792        }
793        break;
794      default:
795        throw new RuntimeException("invalid push type " + type);
796    }
797  }
798
799  private boolean isSyncAborted() {
800    return syncException.get() != null;
801  }
802
803  private void syncLoop() throws Throwable {
804    long totalSyncedToStore = 0;
805    inSync.set(false);
806    lock.lock();
807    try {
808      while (isRunning()) {
809        try {
810          // Wait until new data is available
811          if (slotIndex == 0) {
812            if (!loading.get()) {
813              periodicRoll();
814            }
815
816            if (LOG.isTraceEnabled()) {
817              float rollTsSec = getMillisFromLastRoll() / 1000.0f;
818              LOG.trace(String.format("Waiting for data. flushed=%s (%s/sec)",
819                        StringUtils.humanSize(totalSynced.get()),
820                        StringUtils.humanSize(totalSynced.get() / rollTsSec)));
821            }
822
823            waitCond.await(getMillisToNextPeriodicRoll(), TimeUnit.MILLISECONDS);
824            if (slotIndex == 0) {
825              // no data.. probably a stop() or a periodic roll
826              continue;
827            }
828          }
829          // Wait SYNC_WAIT_MSEC or the signal of "slots full" before flushing
830          syncMaxSlot = runningProcCount;
831          assert syncMaxSlot > 0 : "unexpected syncMaxSlot=" + syncMaxSlot;
832          final long syncWaitSt = System.currentTimeMillis();
833          if (slotIndex != syncMaxSlot) {
834            slotCond.await(syncWaitMsec, TimeUnit.MILLISECONDS);
835          }
836
837          final long currentTs = System.currentTimeMillis();
838          final long syncWaitMs = currentTs - syncWaitSt;
839          final float rollSec = getMillisFromLastRoll() / 1000.0f;
840          final float syncedPerSec = totalSyncedToStore / rollSec;
841          if (LOG.isTraceEnabled() && (syncWaitMs > 10 || slotIndex < syncMaxSlot)) {
842            LOG.trace(String.format("Sync wait %s, slotIndex=%s , totalSynced=%s (%s/sec)",
843                      StringUtils.humanTimeDiff(syncWaitMs), slotIndex,
844                      StringUtils.humanSize(totalSyncedToStore),
845                      StringUtils.humanSize(syncedPerSec)));
846          }
847
848          // update webui circular buffers (TODO: get rid of allocations)
849          final SyncMetrics syncMetrics = new SyncMetrics();
850          syncMetrics.timestamp = currentTs;
851          syncMetrics.syncWaitMs = syncWaitMs;
852          syncMetrics.syncedEntries = slotIndex;
853          syncMetrics.totalSyncedBytes = totalSyncedToStore;
854          syncMetrics.syncedPerSec = syncedPerSec;
855          syncMetricsQueue.add(syncMetrics);
856
857          // sync
858          inSync.set(true);
859          long slotSize = syncSlots();
860          logs.getLast().addToSize(slotSize);
861          totalSyncedToStore = totalSynced.addAndGet(slotSize);
862          slotIndex = 0;
863          inSync.set(false);
864          syncId.incrementAndGet();
865        } catch (InterruptedException e) {
866          Thread.currentThread().interrupt();
867          syncException.compareAndSet(null, e);
868          sendAbortProcessSignal();
869          throw e;
870        } catch (Throwable t) {
871          syncException.compareAndSet(null, t);
872          sendAbortProcessSignal();
873          throw t;
874        } finally {
875          syncCond.signalAll();
876        }
877      }
878    } finally {
879      lock.unlock();
880    }
881  }
882
883  public ArrayList<SyncMetrics> getSyncMetrics() {
884    lock.lock();
885    try {
886      return new ArrayList<>(syncMetricsQueue);
887    } finally {
888      lock.unlock();
889    }
890  }
891
892  private long syncSlots() throws Throwable {
893    int retry = 0;
894    int logRolled = 0;
895    long totalSynced = 0;
896    do {
897      try {
898        totalSynced = syncSlots(stream, slots, 0, slotIndex);
899        break;
900      } catch (Throwable e) {
901        LOG.warn("unable to sync slots, retry=" + retry);
902        if (++retry >= maxRetriesBeforeRoll) {
903          if (logRolled >= maxSyncFailureRoll && isRunning()) {
904            LOG.error("Sync slots after log roll failed, abort.", e);
905            throw e;
906          }
907
908          if (!rollWriterWithRetries()) {
909            throw e;
910          }
911
912          logRolled++;
913          retry = 0;
914        }
915      }
916    } while (isRunning());
917    return totalSynced;
918  }
919
920  protected long syncSlots(final FSDataOutputStream stream, final ByteSlot[] slots,
921      final int offset, final int count) throws IOException {
922    long totalSynced = 0;
923    for (int i = 0; i < count; ++i) {
924      final ByteSlot data = slots[offset + i];
925      data.writeTo(stream);
926      totalSynced += data.size();
927    }
928
929    syncStream(stream);
930    sendPostSyncSignal();
931
932    if (LOG.isTraceEnabled()) {
933      LOG.trace("Sync slots=" + count + '/' + syncMaxSlot +
934                ", flushed=" + StringUtils.humanSize(totalSynced));
935    }
936    return totalSynced;
937  }
938
939  protected void syncStream(final FSDataOutputStream stream) throws IOException {
940    if (useHsync) {
941      stream.hsync();
942    } else {
943      stream.hflush();
944    }
945  }
946
947  private boolean rollWriterWithRetries() {
948    for (int i = 0; i < rollRetries && isRunning(); ++i) {
949      if (i > 0) {
950        Threads.sleepWithoutInterrupt(waitBeforeRoll * i);
951      }
952
953      try {
954        if (rollWriter()) {
955          return true;
956        }
957      } catch (IOException e) {
958        LOG.warn("Unable to roll the log, attempt=" + (i + 1), e);
959      }
960    }
961    LOG.error(HBaseMarkers.FATAL, "Unable to roll the log");
962    return false;
963  }
964
965  private boolean tryRollWriter() {
966    try {
967      return rollWriter();
968    } catch (IOException e) {
969      LOG.warn("Unable to roll the log", e);
970      return false;
971    }
972  }
973
974  public long getMillisToNextPeriodicRoll() {
975    if (lastRollTs.get() > 0 && periodicRollMsec > 0) {
976      return periodicRollMsec - getMillisFromLastRoll();
977    }
978    return Long.MAX_VALUE;
979  }
980
981  public long getMillisFromLastRoll() {
982    return (System.currentTimeMillis() - lastRollTs.get());
983  }
984
985  void periodicRollForTesting() throws IOException {
986    lock.lock();
987    try {
988      periodicRoll();
989    } finally {
990      lock.unlock();
991    }
992  }
993
994  public boolean rollWriterForTesting() throws IOException {
995    lock.lock();
996    try {
997      return rollWriter();
998    } finally {
999      lock.unlock();
1000    }
1001  }
1002
1003  void removeInactiveLogsForTesting() throws Exception {
1004    lock.lock();
1005    try {
1006      removeInactiveLogs();
1007    } finally  {
1008      lock.unlock();
1009    }
1010  }
1011
1012  private void periodicRoll() throws IOException {
1013    if (storeTracker.isEmpty()) {
1014      LOG.trace("no active procedures");
1015      tryRollWriter();
1016      removeAllLogs(flushLogId - 1, "no active procedures");
1017    } else {
1018      if (storeTracker.isAllModified()) {
1019        LOG.trace("all the active procedures are in the latest log");
1020        removeAllLogs(flushLogId - 1, "all the active procedures are in the latest log");
1021      }
1022
1023      // if the log size has exceeded the roll threshold
1024      // or the periodic roll timeout is expired, try to roll the wal.
1025      if (totalSynced.get() > rollThreshold || getMillisToNextPeriodicRoll() <= 0) {
1026        tryRollWriter();
1027      }
1028
1029      removeInactiveLogs();
1030    }
1031  }
1032
1033  private boolean rollWriter() throws IOException {
1034    if (!isRunning()) {
1035      return false;
1036    }
1037
1038    // Create new state-log
1039    if (!rollWriter(flushLogId + 1)) {
1040      LOG.warn("someone else has already created log {}", flushLogId);
1041      return false;
1042    }
1043
1044    // We have the lease on the log,
1045    // but we should check if someone else has created new files
1046    if (getMaxLogId(getLogFiles()) > flushLogId) {
1047      LOG.warn("Someone else created new logs. Expected maxLogId < {}", flushLogId);
1048      logs.getLast().removeFile(this.walArchiveDir);
1049      return false;
1050    }
1051
1052    // We have the lease on the log
1053    return true;
1054  }
1055
1056  boolean rollWriter(long logId) throws IOException {
1057    assert logId > flushLogId : "logId=" + logId + " flushLogId=" + flushLogId;
1058    assert lock.isHeldByCurrentThread() : "expected to be the lock owner. " + lock.isLocked();
1059
1060    ProcedureWALHeader header = ProcedureWALHeader.newBuilder()
1061      .setVersion(ProcedureWALFormat.HEADER_VERSION)
1062      .setType(ProcedureWALFormat.LOG_TYPE_STREAM)
1063      .setMinProcId(storeTracker.getActiveMinProcId())
1064      .setLogId(logId)
1065      .build();
1066
1067    FSDataOutputStream newStream = null;
1068    Path newLogFile = null;
1069    long startPos = -1;
1070    newLogFile = getLogFilePath(logId);
1071    try {
1072      newStream = CommonFSUtils.createForWal(fs, newLogFile, false);
1073    } catch (FileAlreadyExistsException e) {
1074      LOG.error("Log file with id={} already exists", logId, e);
1075      return false;
1076    } catch (RemoteException re) {
1077      LOG.warn("failed to create log file with id={}", logId, re);
1078      return false;
1079    }
1080    // After we create the stream but before we attempt to use it at all
1081    // ensure that we can provide the level of data safety we're configured
1082    // to provide.
1083    final String durability = useHsync ? StreamCapabilities.HSYNC : StreamCapabilities.HFLUSH;
1084    if (enforceStreamCapability && !newStream.hasCapability(durability)) {
1085      throw new IllegalStateException("The procedure WAL relies on the ability to " + durability +
1086          " for proper operation during component failures, but the underlying filesystem does " +
1087          "not support doing so. Please check the config value of '" + USE_HSYNC_CONF_KEY +
1088          "' to set the desired level of robustness and ensure the config value of '" +
1089          CommonFSUtils.HBASE_WAL_DIR + "' points to a FileSystem mount that can provide it.");
1090    }
1091    try {
1092      ProcedureWALFormat.writeHeader(newStream, header);
1093      startPos = newStream.getPos();
1094    } catch (IOException ioe) {
1095      LOG.warn("Encountered exception writing header", ioe);
1096      newStream.close();
1097      return false;
1098    }
1099
1100    closeCurrentLogStream(false);
1101
1102    storeTracker.resetModified();
1103    stream = newStream;
1104    flushLogId = logId;
1105    totalSynced.set(0);
1106    long rollTs = System.currentTimeMillis();
1107    lastRollTs.set(rollTs);
1108    logs.add(new ProcedureWALFile(fs, newLogFile, header, startPos, rollTs));
1109
1110    // if it's the first next WAL being added, build the holding cleanup tracker
1111    if (logs.size() == 2) {
1112      buildHoldingCleanupTracker();
1113    } else if (logs.size() > walCountWarnThreshold) {
1114      LOG.warn("procedure WALs count={} above the warning threshold {}. check running procedures" +
1115        " to see if something is stuck.", logs.size(), walCountWarnThreshold);
1116      // This is just like what we have done at RS side when there are too many wal files. For RS,
1117      // if there are too many wal files, we will find out the wal entries in the oldest file, and
1118      // tell the upper layer to flush these regions so the wal entries will be useless and then we
1119      // can delete the wal file. For WALProcedureStore, the assumption is that, if all the
1120      // procedures recorded in a proc wal file are modified or deleted in a new proc wal file, then
1121      // we are safe to delete it. So here if there are too many proc wal files, we will find out
1122      // the procedure ids in the oldest file, which are neither modified nor deleted in newer proc
1123      // wal files, and tell upper layer to update the state of these procedures to the newest proc
1124      // wal file(by calling ProcedureStore.update), then we are safe to delete the oldest proc wal
1125      // file.
1126      sendForceUpdateSignal(holdingCleanupTracker.getAllActiveProcIds());
1127    }
1128
1129    LOG.info("Rolled new Procedure Store WAL, id={}", logId);
1130    return true;
1131  }
1132
1133  private void closeCurrentLogStream(boolean abort) {
1134    if (stream == null || logs.isEmpty()) {
1135      return;
1136    }
1137
1138    try {
1139      ProcedureWALFile log = logs.getLast();
1140      // If the loading flag is true, it usually means that we fail when loading procedures, so we
1141      // should not persist the store tracker, as its state may not be correct.
1142      if (!loading.get()) {
1143        log.setProcIds(storeTracker.getModifiedMinProcId(), storeTracker.getModifiedMaxProcId());
1144        log.updateLocalTracker(storeTracker);
1145        if (!abort) {
1146          long trailerSize = ProcedureWALFormat.writeTrailer(stream, storeTracker);
1147          log.addToSize(trailerSize);
1148        }
1149      }
1150    } catch (IOException | FSError e) {
1151      LOG.warn("Unable to write the trailer", e);
1152    }
1153    try {
1154      stream.close();
1155    } catch (IOException | FSError e) {
1156      LOG.error("Unable to close the stream", e);
1157    }
1158    stream = null;
1159  }
1160
1161  // ==========================================================================
1162  //  Log Files cleaner helpers
1163  // ==========================================================================
1164  private void removeInactiveLogs() throws IOException {
1165    // We keep track of which procedures are holding the oldest WAL in 'holdingCleanupTracker'.
1166    // once there is nothing olding the oldest WAL we can remove it.
1167    while (logs.size() > 1 && holdingCleanupTracker.isEmpty()) {
1168      LOG.info("Remove the oldest log {}", logs.getFirst());
1169      removeLogFile(logs.getFirst(), walArchiveDir);
1170      buildHoldingCleanupTracker();
1171    }
1172
1173    // TODO: In case we are holding up a lot of logs for long time we should
1174    // rewrite old procedures (in theory parent procs) to the new WAL.
1175  }
1176
1177  private void buildHoldingCleanupTracker() {
1178    if (logs.size() <= 1) {
1179      // we only have one wal, so nothing to do
1180      holdingCleanupTracker.reset();
1181      return;
1182    }
1183
1184    // compute the holding tracker.
1185    // - the first WAL is used for the 'updates'
1186    // - the global tracker will be used to determine whether a procedure has been deleted
1187    // - other trackers will be used to determine whether a procedure has been updated, as a deleted
1188    // procedure can always be detected by checking the global tracker, we can save the deleted
1189    // checks when applying other trackers
1190    holdingCleanupTracker.resetTo(logs.getFirst().getTracker(), true);
1191    holdingCleanupTracker.setDeletedIfDeletedByThem(storeTracker);
1192    // the logs is a linked list, so avoid calling get(index) on it.
1193    Iterator<ProcedureWALFile> iter = logs.iterator();
1194    // skip the tracker for the first file when creating the iterator.
1195    iter.next();
1196    ProcedureStoreTracker tracker = iter.next().getTracker();
1197    // testing iter.hasNext after calling iter.next to skip applying the tracker for last file,
1198    // which is just the storeTracker above.
1199    while (iter.hasNext()) {
1200      holdingCleanupTracker.setDeletedIfModifiedInBoth(tracker);
1201      if (holdingCleanupTracker.isEmpty()) {
1202        break;
1203      }
1204      tracker = iter.next().getTracker();
1205    }
1206  }
1207
1208  /**
1209   * Remove all logs with logId <= {@code lastLogId}.
1210   */
1211  private void removeAllLogs(long lastLogId, String why) {
1212    if (logs.size() <= 1) {
1213      return;
1214    }
1215
1216    LOG.info("Remove all state logs with ID less than {}, since {}", lastLogId, why);
1217
1218    boolean removed = false;
1219    while (logs.size() > 1) {
1220      ProcedureWALFile log = logs.getFirst();
1221      if (lastLogId < log.getLogId()) {
1222        break;
1223      }
1224      removeLogFile(log, walArchiveDir);
1225      removed = true;
1226    }
1227
1228    if (removed) {
1229      buildHoldingCleanupTracker();
1230    }
1231  }
1232
1233  private boolean removeLogFile(final ProcedureWALFile log, final Path walArchiveDir) {
1234    try {
1235      LOG.trace("Removing log={}", log);
1236      log.removeFile(walArchiveDir);
1237      logs.remove(log);
1238      LOG.debug("Removed log={}, activeLogs={}", log, logs);
1239      assert logs.size() > 0 : "expected at least one log";
1240    } catch (IOException e) {
1241      LOG.error("Unable to remove log: " + log, e);
1242      return false;
1243    }
1244    return true;
1245  }
1246
1247  // ==========================================================================
1248  //  FileSystem Log Files helpers
1249  // ==========================================================================
1250  public Path getWALDir() {
1251    return this.walDir;
1252  }
1253
1254  Path getWalArchiveDir() {
1255    return this.walArchiveDir;
1256  }
1257
1258  public FileSystem getFileSystem() {
1259    return this.fs;
1260  }
1261
1262  protected Path getLogFilePath(final long logId) throws IOException {
1263    return new Path(walDir, String.format(LOG_PREFIX + "%020d.log", logId));
1264  }
1265
1266  private static long getLogIdFromName(final String name) {
1267    int end = name.lastIndexOf(".log");
1268    int start = name.lastIndexOf('-') + 1;
1269    return Long.parseLong(name.substring(start, end));
1270  }
1271
1272  private static final PathFilter WALS_PATH_FILTER = new PathFilter() {
1273    @Override
1274    public boolean accept(Path path) {
1275      String name = path.getName();
1276      return name.startsWith(LOG_PREFIX) && name.endsWith(".log");
1277    }
1278  };
1279
1280  private static final Comparator<FileStatus> FILE_STATUS_ID_COMPARATOR =
1281      new Comparator<FileStatus>() {
1282    @Override
1283    public int compare(FileStatus a, FileStatus b) {
1284      final long aId = getLogIdFromName(a.getPath().getName());
1285      final long bId = getLogIdFromName(b.getPath().getName());
1286      return Long.compare(aId, bId);
1287    }
1288  };
1289
1290  private FileStatus[] getLogFiles() throws IOException {
1291    try {
1292      FileStatus[] files = fs.listStatus(walDir, WALS_PATH_FILTER);
1293      Arrays.sort(files, FILE_STATUS_ID_COMPARATOR);
1294      return files;
1295    } catch (FileNotFoundException e) {
1296      LOG.warn("Log directory not found: " + e.getMessage());
1297      return null;
1298    }
1299  }
1300
1301  /**
1302   * Make sure that the file set are gotten by calling {@link #getLogFiles()}, where we will sort
1303   * the file set by log id.
1304   * @return Max-LogID of the specified log file set
1305   */
1306  private static long getMaxLogId(FileStatus[] logFiles) {
1307    if (logFiles == null || logFiles.length == 0) {
1308      return 0L;
1309    }
1310    return getLogIdFromName(logFiles[logFiles.length - 1].getPath().getName());
1311  }
1312
1313  /**
1314   * Make sure that the file set are gotten by calling {@link #getLogFiles()}, where we will sort
1315   * the file set by log id.
1316   * @return Max-LogID of the specified log file set
1317   */
1318  private long initOldLogs(FileStatus[] logFiles) throws IOException {
1319    if (logFiles == null || logFiles.length == 0) {
1320      return 0L;
1321    }
1322    long maxLogId = 0;
1323    for (int i = 0; i < logFiles.length; ++i) {
1324      final Path logPath = logFiles[i].getPath();
1325      leaseRecovery.recoverFileLease(fs, logPath);
1326      if (!isRunning()) {
1327        throw new IOException("wal aborting");
1328      }
1329
1330      maxLogId = Math.max(maxLogId, getLogIdFromName(logPath.getName()));
1331      ProcedureWALFile log = initOldLog(logFiles[i], this.walArchiveDir);
1332      if (log != null) {
1333        this.logs.add(log);
1334      }
1335    }
1336    initTrackerFromOldLogs();
1337    return maxLogId;
1338  }
1339
1340  /**
1341   * If last log's tracker is not null, use it as {@link #storeTracker}. Otherwise, set storeTracker
1342   * as partial, and let {@link ProcedureWALFormatReader} rebuild it using entries in the log.
1343   */
1344  private void initTrackerFromOldLogs() {
1345    if (logs.isEmpty() || !isRunning()) {
1346      return;
1347    }
1348    ProcedureWALFile log = logs.getLast();
1349    if (!log.getTracker().isPartial()) {
1350      storeTracker.resetTo(log.getTracker());
1351    } else {
1352      storeTracker.reset();
1353      storeTracker.setPartialFlag(true);
1354    }
1355  }
1356
1357  /**
1358   * Loads given log file and it's tracker.
1359   */
1360  private ProcedureWALFile initOldLog(final FileStatus logFile, final Path walArchiveDir)
1361      throws IOException {
1362    final ProcedureWALFile log = new ProcedureWALFile(fs, logFile);
1363    if (logFile.getLen() == 0) {
1364      LOG.warn("Remove uninitialized log: {}", logFile);
1365      log.removeFile(walArchiveDir);
1366      return null;
1367    }
1368    LOG.debug("Opening Pv2 {}", logFile);
1369    try {
1370      log.open();
1371    } catch (ProcedureWALFormat.InvalidWALDataException e) {
1372      LOG.warn("Remove uninitialized log: {}", logFile, e);
1373      log.removeFile(walArchiveDir);
1374      return null;
1375    } catch (IOException e) {
1376      String msg = "Unable to read state log: " + logFile;
1377      LOG.error(msg, e);
1378      throw new IOException(msg, e);
1379    }
1380
1381    try {
1382      log.readTracker();
1383    } catch (IOException e) {
1384      log.getTracker().reset();
1385      log.getTracker().setPartialFlag(true);
1386      LOG.warn("Unable to read tracker for {}", log, e);
1387    }
1388
1389    log.close();
1390    return log;
1391  }
1392
1393  /**
1394   * Parses a directory of WALs building up ProcedureState.
1395   * For testing parse and profiling.
1396   * @param args Include pointer to directory of WAL files for a store instance to parse & load.
1397   */
1398  public static void main(String [] args) throws IOException {
1399    Configuration conf = HBaseConfiguration.create();
1400    if (args == null || args.length != 1) {
1401      System.out.println("ERROR: Empty arguments list; pass path to MASTERPROCWALS_DIR.");
1402      System.out.println("Usage: WALProcedureStore MASTERPROCWALS_DIR");
1403      System.exit(-1);
1404    }
1405    WALProcedureStore store = new WALProcedureStore(conf, new Path(args[0]), null,
1406      new LeaseRecovery() {
1407        @Override
1408        public void recoverFileLease(FileSystem fs, Path path) throws IOException {
1409          // no-op
1410        }
1411      });
1412    try {
1413      store.start(16);
1414      ProcedureExecutor<?> pe = new ProcedureExecutor<>(conf, new Object()/*Pass anything*/, store);
1415      pe.init(1, true);
1416    } finally {
1417      store.stop(true);
1418    }
1419  }
1420}