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