hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALSplitUtil.java

   1 /*
   2  * Licensed to the Apache Software Foundation (ASF) under one
   3  * or more contributor license agreements.  See the NOTICE file
   4  * distributed with this work for additional information
   5  * regarding copyright ownership.  The ASF licenses this file
   6  * to you under the Apache License, Version 2.0 (the
   7  * "License"); you may not use this file except in compliance
   8  * with the License.  You may obtain a copy of the License at
   9  *
  10  *     http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software
  13  * distributed under the License is distributed on an "AS IS" BASIS,
  14  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  * See the License for the specific language governing permissions and
  16  * limitations under the License.
  17  */
  18 package org.apache.hadoop.hbase.wal;
  19
  20 import java.io.FileNotFoundException;
  21 import java.io.IOException;
  22 import java.util.ArrayList;
  23 import java.util.Arrays;
  24 import java.util.Collections;
  25 import java.util.List;
  26 import java.util.NavigableSet;
  27 import java.util.TreeSet;
  28 import java.util.UUID;
  29 import java.util.regex.Matcher;
  30 import java.util.regex.Pattern;
  31 import org.apache.commons.lang3.ArrayUtils;
  32 import org.apache.hadoop.conf.Configuration;
  33 import org.apache.hadoop.fs.FileAlreadyExistsException;
  34 import org.apache.hadoop.fs.FileStatus;
  35 import org.apache.hadoop.fs.FileSystem;
  36 import org.apache.hadoop.fs.Path;
  37 import org.apache.hadoop.fs.PathFilter;
  38 import org.apache.hadoop.hbase.Cell;
  39 import org.apache.hadoop.hbase.CellScanner;
  40 import org.apache.hadoop.hbase.CellUtil;
  41 import org.apache.hadoop.hbase.HConstants;
  42 import org.apache.hadoop.hbase.TableName;
  43 import org.apache.hadoop.hbase.client.Delete;
  44 import org.apache.hadoop.hbase.client.Durability;
  45 import org.apache.hadoop.hbase.client.Mutation;
  46 import org.apache.hadoop.hbase.client.Put;
  47 import org.apache.hadoop.hbase.client.RegionInfo;
  48 import org.apache.hadoop.hbase.client.Row;
  49 import org.apache.hadoop.hbase.regionserver.HRegion;
  50 import org.apache.hadoop.hbase.regionserver.wal.AbstractFSWAL;
  51 import org.apache.hadoop.hbase.util.Bytes;
  52 import org.apache.hadoop.hbase.util.CommonFSUtils;
  53 import org.apache.hadoop.hbase.util.ConcurrentMapUtils.IOExceptionSupplier;
  54 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
  55 import org.apache.hadoop.hbase.util.FSUtils;
  56 import org.apache.hadoop.hbase.util.Pair;
  57 import org.apache.hadoop.hbase.zookeeper.ZKSplitLog;
  58 import org.apache.yetus.audience.InterfaceAudience;
  59 import org.slf4j.Logger;
  60 import org.slf4j.LoggerFactory;
  61
  62 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos;
  63 import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos;
  64 import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos;
  65
  66 /**
  67  * This class provides static methods to support WAL splitting related works
  68  */
  69 @InterfaceAudience.Private
  70 public final class WALSplitUtil {
  71   private static final Logger LOG = LoggerFactory.getLogger(WALSplitUtil.class);
  72
  73   private static final Pattern EDITFILES_NAME_PATTERN = Pattern.compile("-?[0-9]+");
  74   private static final String RECOVERED_LOG_TMPFILE_SUFFIX = ".temp";
  75   private static final String SEQUENCE_ID_FILE_SUFFIX = ".seqid";
  76   private static final String OLD_SEQUENCE_ID_FILE_SUFFIX = "_seqid";
  77   private static final int SEQUENCE_ID_FILE_SUFFIX_LENGTH = SEQUENCE_ID_FILE_SUFFIX.length();
  78
  79   private WALSplitUtil() {
  80   }
  81
  82   /**
  83    * Completes the work done by splitLogFile by archiving logs
  84    * <p>
  85    * It is invoked by SplitLogManager once it knows that one of the SplitLogWorkers have completed
  86    * the splitLogFile() part. If the master crashes then this function might get called multiple
  87    * times.
  88    * <p>
  89    */
  90   public static void finishSplitLogFile(String logfile, Configuration conf) throws IOException {
  91     Path walDir = CommonFSUtils.getWALRootDir(conf);
  92     Path oldLogDir = new Path(walDir, HConstants.HREGION_OLDLOGDIR_NAME);
  93     Path walPath;
  94     if (CommonFSUtils.isStartingWithPath(walDir, logfile)) {
  95       walPath = new Path(logfile);
  96     } else {
  97       walPath = new Path(walDir, logfile);
  98     }
  99     FileSystem walFS = walDir.getFileSystem(conf);
 100     boolean corrupt = ZKSplitLog.isCorrupted(walDir, walPath.getName(), walFS);
 101     archive(walPath, corrupt, oldLogDir, walFS, conf);
 102     Path stagingDir = ZKSplitLog.getSplitLogDir(walDir, walPath.getName());
 103     walFS.delete(stagingDir, true);
 104   }
 105
 106   /**
 107    * Moves processed logs to a oldLogDir after successful processing Moves corrupted logs (any log
 108    * that couldn't be successfully parsed to corruptDir (.corrupt) for later investigation
 109    */
 110   static void archive(final Path wal, final boolean corrupt, final Path oldWALDir,
 111       final FileSystem walFS, final Configuration conf) throws IOException {
 112     Path dir;
 113     Path target;
 114     if (corrupt) {
 115       dir = new Path(CommonFSUtils.getWALRootDir(conf), HConstants.CORRUPT_DIR_NAME);
 116       if (conf.get("hbase.regionserver.hlog.splitlog.corrupt.dir") != null) {
 117         LOG.warn("hbase.regionserver.hlog.splitlog.corrupt.dir is deprecated. Default to {}", dir);
 118       }
 119       target = new Path(dir, wal.getName());
 120     } else {
 121       dir = oldWALDir;
 122       target = AbstractFSWAL.getWALArchivePath(oldWALDir, wal);
 123     }
 124     mkdir(walFS, dir);
 125     moveWAL(walFS, wal, target);
 126   }
 127
 128   private static void mkdir(FileSystem fs, Path dir) throws IOException {
 129     if (!fs.mkdirs(dir)) {
 130       LOG.warn("Failed mkdir {}", dir);
 131     }
 132   }
 133
 134   /**
 135    * Move WAL. Used to move processed WALs to archive or bad WALs to corrupt WAL dir.
 136    * WAL may have already been moved; makes allowance.
 137    */
 138   public static void moveWAL(FileSystem fs, Path p, Path targetDir) throws IOException {
 139     if (fs.exists(p)) {
 140       if (!CommonFSUtils.renameAndSetModifyTime(fs, p, targetDir)) {
 141         LOG.warn("Failed move of {} to {}", p, targetDir);
 142       } else {
 143         LOG.info("Moved {} to {}", p, targetDir);
 144       }
 145     }
 146   }
 147
 148   /**
 149    * Path to a file under RECOVERED_EDITS_DIR directory of the region found in <code>logEntry</code>
 150    * named for the sequenceid in the passed <code>logEntry</code>: e.g.
 151    * /hbase/some_table/2323432434/recovered.edits/2332. This method also ensures existence of
 152    * RECOVERED_EDITS_DIR under the region creating it if necessary.
 153    * And also set storage policy for RECOVERED_EDITS_DIR if WAL_STORAGE_POLICY is configured.
 154    * @param tableName the table name
 155    * @param encodedRegionName the encoded region name
 156    * @param seqId the sequence id which used to generate file name
 157    * @param fileNameBeingSplit the file being split currently. Used to generate tmp file name.
 158    * @param tmpDirName of the directory used to sideline old recovered edits file
 159    * @param conf configuration
 160    * @return Path to file into which to dump split log edits.
 161    */
 162   @SuppressWarnings("deprecation")
 163   static Path getRegionSplitEditsPath(TableName tableName, byte[] encodedRegionName, long seqId,
 164       String fileNameBeingSplit, String tmpDirName, Configuration conf) throws IOException {
 165     FileSystem walFS = CommonFSUtils.getWALFileSystem(conf);
 166     Path tableDir = CommonFSUtils.getWALTableDir(conf, tableName);
 167     String encodedRegionNameStr = Bytes.toString(encodedRegionName);
 168     Path regionDir = HRegion.getRegionDir(tableDir, encodedRegionNameStr);
 169     Path dir = getRegionDirRecoveredEditsDir(regionDir);
 170
 171     if (walFS.exists(dir) && walFS.isFile(dir)) {
 172       Path tmp = new Path(tmpDirName);
 173       if (!walFS.exists(tmp)) {
 174         walFS.mkdirs(tmp);
 175       }
 176       tmp = new Path(tmp, HConstants.RECOVERED_EDITS_DIR + "_" + encodedRegionNameStr);
 177       LOG.warn("Found existing old file: {}. It could be some "
 178           + "leftover of an old installation. It should be a folder instead. "
 179           + "So moving it to {}",
 180         dir, tmp);
 181       if (!walFS.rename(dir, tmp)) {
 182         LOG.warn("Failed to sideline old file {}", dir);
 183       }
 184     }
 185
 186     if (!walFS.exists(dir) && !walFS.mkdirs(dir)) {
 187       LOG.warn("mkdir failed on {}", dir);
 188     } else {
 189       String storagePolicy =
 190         conf.get(HConstants.WAL_STORAGE_POLICY, HConstants.DEFAULT_WAL_STORAGE_POLICY);
 191       CommonFSUtils.setStoragePolicy(walFS, dir, storagePolicy);
 192     }
 193     // Append fileBeingSplit to prevent name conflict since we may have duplicate wal entries now.
 194     // Append file name ends with RECOVERED_LOG_TMPFILE_SUFFIX to ensure
 195     // region's replayRecoveredEdits will not delete it
 196     String fileName = formatRecoveredEditsFileName(seqId);
 197     fileName = getTmpRecoveredEditsFileName(fileName + "-" + fileNameBeingSplit);
 198     return new Path(dir, fileName);
 199   }
 200
 201   private static String getTmpRecoveredEditsFileName(String fileName) {
 202     return fileName + RECOVERED_LOG_TMPFILE_SUFFIX;
 203   }
 204
 205   /**
 206    * Get the completed recovered edits file path, renaming it to be by last edit in the file from
 207    * its first edit. Then we could use the name to skip recovered edits when doing
 208    * HRegion#replayRecoveredEditsIfAny(Map, CancelableProgressable, MonitoredTask).
 209    * @return dstPath take file's last edit log seq num as the name
 210    */
 211   static Path getCompletedRecoveredEditsFilePath(Path srcPath, long maximumEditWALSeqNum) {
 212     String fileName = formatRecoveredEditsFileName(maximumEditWALSeqNum);
 213     return new Path(srcPath.getParent(), fileName);
 214   }
 215
 216   static String formatRecoveredEditsFileName(final long seqid) {
 217     return String.format("%019d", seqid);
 218   }
 219
 220   /**
 221    * @param regionDir This regions directory in the filesystem.
 222    * @return The directory that holds recovered edits files for the region <code>regionDir</code>
 223    */
 224   public static Path getRegionDirRecoveredEditsDir(final Path regionDir) {
 225     return new Path(regionDir, HConstants.RECOVERED_EDITS_DIR);
 226   }
 227
 228   /**
 229    * Check whether there is recovered.edits in the region dir
 230    * @param conf conf
 231    * @param regionInfo the region to check
 232    * @return true if recovered.edits exist in the region dir
 233    */
 234   public static boolean hasRecoveredEdits(final Configuration conf, final RegionInfo regionInfo)
 235       throws IOException {
 236     // No recovered.edits for non default replica regions
 237     if (regionInfo.getReplicaId() != RegionInfo.DEFAULT_REPLICA_ID) {
 238       return false;
 239     }
 240     // Only default replica region can reach here, so we can use regioninfo
 241     // directly without converting it to default replica's regioninfo.
 242     Path regionWALDir =
 243       CommonFSUtils.getWALRegionDir(conf, regionInfo.getTable(), regionInfo.getEncodedName());
 244     Path regionDir = FSUtils.getRegionDirFromRootDir(CommonFSUtils.getRootDir(conf), regionInfo);
 245     Path wrongRegionWALDir =
 246       CommonFSUtils.getWrongWALRegionDir(conf, regionInfo.getTable(), regionInfo.getEncodedName());
 247     FileSystem walFs = CommonFSUtils.getWALFileSystem(conf);
 248     FileSystem rootFs = CommonFSUtils.getRootDirFileSystem(conf);
 249     NavigableSet<Path> files = getSplitEditFilesSorted(walFs, regionWALDir);
 250     if (!files.isEmpty()) {
 251       return true;
 252     }
 253     files = getSplitEditFilesSorted(rootFs, regionDir);
 254     if (!files.isEmpty()) {
 255       return true;
 256     }
 257     files = getSplitEditFilesSorted(walFs, wrongRegionWALDir);
 258     return !files.isEmpty();
 259   }
 260
 261   /**
 262    * This method will check 3 places for finding the max sequence id file. One is the expected
 263    * place, another is the old place under the region directory, and the last one is the wrong one
 264    * we introduced in HBASE-20734. See HBASE-22617 for more details.
 265    * <p/>
 266    * Notice that, you should always call this method instead of
 267    * {@link #getMaxRegionSequenceId(FileSystem, Path)} until 4.0.0 release.
 268    * @deprecated Only for compatibility, will be removed in 4.0.0.
 269    */
 270   @Deprecated
 271   public static long getMaxRegionSequenceId(Configuration conf, RegionInfo region,
 272     IOExceptionSupplier<FileSystem> rootFsSupplier, IOExceptionSupplier<FileSystem> walFsSupplier)
 273     throws IOException {
 274     FileSystem rootFs = rootFsSupplier.get();
 275     FileSystem walFs = walFsSupplier.get();
 276     Path regionWALDir =
 277       CommonFSUtils.getWALRegionDir(conf, region.getTable(), region.getEncodedName());
 278     // This is the old place where we store max sequence id file
 279     Path regionDir = FSUtils.getRegionDirFromRootDir(CommonFSUtils.getRootDir(conf), region);
 280     // This is for HBASE-20734, where we use a wrong directory, see HBASE-22617 for more details.
 281     Path wrongRegionWALDir =
 282       CommonFSUtils.getWrongWALRegionDir(conf, region.getTable(), region.getEncodedName());
 283     long maxSeqId = getMaxRegionSequenceId(walFs, regionWALDir);
 284     maxSeqId = Math.max(maxSeqId, getMaxRegionSequenceId(rootFs, regionDir));
 285     maxSeqId = Math.max(maxSeqId, getMaxRegionSequenceId(walFs, wrongRegionWALDir));
 286     return maxSeqId;
 287   }
 288
 289   /**
 290    * Returns sorted set of edit files made by splitter, excluding files with '.temp' suffix.
 291    * @param walFS WAL FileSystem used to retrieving split edits files.
 292    * @param regionDir WAL region dir to look for recovered edits files under.
 293    * @return Files in passed <code>regionDir</code> as a sorted set.
 294    */
 295   public static NavigableSet<Path> getSplitEditFilesSorted(final FileSystem walFS,
 296       final Path regionDir) throws IOException {
 297     NavigableSet<Path> filesSorted = new TreeSet<>();
 298     Path editsdir = getRegionDirRecoveredEditsDir(regionDir);
 299     if (!walFS.exists(editsdir)) {
 300       return filesSorted;
 301     }
 302     FileStatus[] files = CommonFSUtils.listStatus(walFS, editsdir, new PathFilter() {
 303       @Override
 304       public boolean accept(Path p) {
 305         boolean result = false;
 306         try {
 307           // Return files and only files that match the editfile names pattern.
 308           // There can be other files in this directory other than edit files.
 309           // In particular, on error, we'll move aside the bad edit file giving
 310           // it a timestamp suffix. See moveAsideBadEditsFile.
 311           Matcher m = EDITFILES_NAME_PATTERN.matcher(p.getName());
 312           result = walFS.isFile(p) && m.matches();
 313           // Skip the file whose name ends with RECOVERED_LOG_TMPFILE_SUFFIX,
 314           // because it means splitwal thread is writting this file.
 315           if (p.getName().endsWith(RECOVERED_LOG_TMPFILE_SUFFIX)) {
 316             result = false;
 317           }
 318           // Skip SeqId Files
 319           if (isSequenceIdFile(p)) {
 320             result = false;
 321           }
 322         } catch (IOException e) {
 323           LOG.warn("Failed isFile check on {}", p, e);
 324         }
 325         return result;
 326       }
 327     });
 328     if (ArrayUtils.isNotEmpty(files)) {
 329       Arrays.asList(files).forEach(status -> filesSorted.add(status.getPath()));
 330     }
 331     return filesSorted;
 332   }
 333
 334   /**
 335    * Move aside a bad edits file.
 336    * @param fs the file system used to rename bad edits file.
 337    * @param edits Edits file to move aside.
 338    * @return The name of the moved aside file.
 339    */
 340   public static Path moveAsideBadEditsFile(final FileSystem fs, final Path edits)
 341       throws IOException {
 342     Path moveAsideName =
 343         new Path(edits.getParent(), edits.getName() + "." + EnvironmentEdgeManager.currentTime());
 344     if (!fs.rename(edits, moveAsideName)) {
 345       LOG.warn("Rename failed from {} to {}", edits, moveAsideName);
 346     }
 347     return moveAsideName;
 348   }
 349
 350   /**
 351    * Is the given file a region open sequence id file.
 352    */
 353   public static boolean isSequenceIdFile(final Path file) {
 354     return file.getName().endsWith(SEQUENCE_ID_FILE_SUFFIX)
 355         || file.getName().endsWith(OLD_SEQUENCE_ID_FILE_SUFFIX);
 356   }
 357
 358   private static FileStatus[] getSequenceIdFiles(FileSystem walFS, Path regionDir)
 359       throws IOException {
 360     // TODO: Why are we using a method in here as part of our normal region open where
 361     // there is no splitting involved? Fix. St.Ack 01/20/2017.
 362     Path editsDir = getRegionDirRecoveredEditsDir(regionDir);
 363     try {
 364       FileStatus[] files = walFS.listStatus(editsDir, WALSplitUtil::isSequenceIdFile);
 365       return files != null ? files : new FileStatus[0];
 366     } catch (FileNotFoundException e) {
 367       return new FileStatus[0];
 368     }
 369   }
 370
 371   private static long getMaxSequenceId(FileStatus[] files) {
 372     long maxSeqId = -1L;
 373     for (FileStatus file : files) {
 374       String fileName = file.getPath().getName();
 375       try {
 376         maxSeqId = Math.max(maxSeqId, Long
 377             .parseLong(fileName.substring(0, fileName.length() - SEQUENCE_ID_FILE_SUFFIX_LENGTH)));
 378       } catch (NumberFormatException ex) {
 379         LOG.warn("Invalid SeqId File Name={}", fileName);
 380       }
 381     }
 382     return maxSeqId;
 383   }
 384
 385   /**
 386    * Get the max sequence id which is stored in the region directory. -1 if none.
 387    */
 388   public static long getMaxRegionSequenceId(FileSystem walFS, Path regionDir) throws IOException {
 389     return getMaxSequenceId(getSequenceIdFiles(walFS, regionDir));
 390   }
 391
 392   /**
 393    * Create a file with name as region's max sequence id
 394    */
 395   public static void writeRegionSequenceIdFile(FileSystem walFS, Path regionDir, long newMaxSeqId)
 396       throws IOException {
 397     FileStatus[] files = getSequenceIdFiles(walFS, regionDir);
 398     long maxSeqId = getMaxSequenceId(files);
 399     if (maxSeqId > newMaxSeqId) {
 400       throw new IOException("The new max sequence id " + newMaxSeqId
 401           + " is less than the old max sequence id " + maxSeqId);
 402     }
 403     // write a new seqId file
 404     Path newSeqIdFile =
 405         new Path(getRegionDirRecoveredEditsDir(regionDir), newMaxSeqId + SEQUENCE_ID_FILE_SUFFIX);
 406     if (newMaxSeqId != maxSeqId) {
 407       try {
 408         if (!walFS.createNewFile(newSeqIdFile) && !walFS.exists(newSeqIdFile)) {
 409           throw new IOException("Failed to create SeqId file:" + newSeqIdFile);
 410         }
 411         LOG.debug("Wrote file={}, newMaxSeqId={}, maxSeqId={}", newSeqIdFile, newMaxSeqId,
 412           maxSeqId);
 413       } catch (FileAlreadyExistsException ignored) {
 414         // latest hdfs throws this exception. it's all right if newSeqIdFile already exists
 415       }
 416     }
 417     // remove old ones
 418     for (FileStatus status : files) {
 419       if (!newSeqIdFile.equals(status.getPath())) {
 420         walFS.delete(status.getPath(), false);
 421       }
 422     }
 423   }
 424
 425   /** A struct used by getMutationsFromWALEntry */
 426   public static class MutationReplay implements Comparable<MutationReplay> {
 427     public MutationReplay(ClientProtos.MutationProto.MutationType type, Mutation mutation,
 428         long nonceGroup, long nonce) {
 429       this.type = type;
 430       this.mutation = mutation;
 431       if (this.mutation.getDurability() != Durability.SKIP_WAL) {
 432         // using ASYNC_WAL for relay
 433         this.mutation.setDurability(Durability.ASYNC_WAL);
 434       }
 435       this.nonceGroup = nonceGroup;
 436       this.nonce = nonce;
 437     }
 438
 439     private final ClientProtos.MutationProto.MutationType type;
 440     @SuppressWarnings("checkstyle:VisibilityModifier") public final Mutation mutation;
 441     @SuppressWarnings("checkstyle:VisibilityModifier") public final long nonceGroup;
 442     @SuppressWarnings("checkstyle:VisibilityModifier") public final long nonce;
 443
 444     @Override
 445     public int compareTo(final MutationReplay d) {
 446       return Row.COMPARATOR.compare(mutation, d.mutation);
 447     }
 448
 449     @Override
 450     public boolean equals(Object obj) {
 451       if (!(obj instanceof MutationReplay)) {
 452         return false;
 453       } else {
 454         return this.compareTo((MutationReplay) obj) == 0;
 455       }
 456     }
 457
 458     @Override
 459     public int hashCode() {
 460       return this.mutation.hashCode();
 461     }
 462
 463     public ClientProtos.MutationProto.MutationType getType() {
 464       return type;
 465     }
 466   }
 467
 468   /**
 469    * This function is used to construct mutations from a WALEntry. It also reconstructs WALKey &amp;
 470    * WALEdit from the passed in WALEntry
 471    * @param logEntry pair of WALKey and WALEdit instance stores WALKey and WALEdit instances
 472    *          extracted from the passed in WALEntry.
 473    * @return list of Pair&lt;MutationType, Mutation&gt; to be replayed
 474    * @deprecated Since 3.0.0, will be removed in 4.0.0.
 475    */
 476   @Deprecated
 477   public static List<MutationReplay> getMutationsFromWALEntry(AdminProtos.WALEntry entry,
 478       CellScanner cells, Pair<WALKey, WALEdit> logEntry, Durability durability) throws IOException {
 479     if (entry == null) {
 480       // return an empty array
 481       return Collections.emptyList();
 482     }
 483
 484     long replaySeqId =
 485         (entry.getKey().hasOrigSequenceNumber()) ? entry.getKey().getOrigSequenceNumber()
 486             : entry.getKey().getLogSequenceNumber();
 487     int count = entry.getAssociatedCellCount();
 488     List<MutationReplay> mutations = new ArrayList<>();
 489     Cell previousCell = null;
 490     Mutation m = null;
 491     WALKeyImpl key = null;
 492     WALEdit val = null;
 493     if (logEntry != null) {
 494       val = new WALEdit();
 495     }
 496
 497     for (int i = 0; i < count; i++) {
 498       // Throw index out of bounds if our cell count is off
 499       if (!cells.advance()) {
 500         throw new ArrayIndexOutOfBoundsException("Expected=" + count + ", index=" + i);
 501       }
 502       Cell cell = cells.current();
 503       if (val != null) {
 504         val.add(cell);
 505       }
 506
 507       boolean isNewRowOrType =
 508           previousCell == null || previousCell.getTypeByte() != cell.getTypeByte()
 509               || !CellUtil.matchingRows(previousCell, cell);
 510       if (isNewRowOrType) {
 511         // Create new mutation
 512         if (CellUtil.isDelete(cell)) {
 513           m = new Delete(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
 514           // Deletes don't have nonces.
 515           mutations.add(new MutationReplay(ClientProtos.MutationProto.MutationType.DELETE, m,
 516               HConstants.NO_NONCE, HConstants.NO_NONCE));
 517         } else {
 518           m = new Put(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
 519           // Puts might come from increment or append, thus we need nonces.
 520           long nonceGroup =
 521               entry.getKey().hasNonceGroup() ? entry.getKey().getNonceGroup() : HConstants.NO_NONCE;
 522           long nonce = entry.getKey().hasNonce() ? entry.getKey().getNonce() : HConstants.NO_NONCE;
 523           mutations.add(
 524             new MutationReplay(ClientProtos.MutationProto.MutationType.PUT, m, nonceGroup, nonce));
 525         }
 526       }
 527       if (CellUtil.isDelete(cell)) {
 528         ((Delete) m).add(cell);
 529       } else {
 530         ((Put) m).add(cell);
 531       }
 532       m.setDurability(durability);
 533       previousCell = cell;
 534     }
 535
 536     // reconstruct WALKey
 537     if (logEntry != null) {
 538       org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos.WALKey walKeyProto =
 539           entry.getKey();
 540       List<UUID> clusterIds = new ArrayList<>(walKeyProto.getClusterIdsCount());
 541       for (HBaseProtos.UUID uuid : entry.getKey().getClusterIdsList()) {
 542         clusterIds.add(new UUID(uuid.getMostSigBits(), uuid.getLeastSigBits()));
 543       }
 544       key = new WALKeyImpl(walKeyProto.getEncodedRegionName().toByteArray(),
 545           TableName.valueOf(walKeyProto.getTableName().toByteArray()), replaySeqId,
 546           walKeyProto.getWriteTime(), clusterIds, walKeyProto.getNonceGroup(),
 547           walKeyProto.getNonce(), null);
 548       logEntry.setFirst(key);
 549       logEntry.setSecond(val);
 550     }
 551
 552     return mutations;
 553   }
 554
 555   /**
 556    * Return path to recovered.hfiles directory of the region's column family: e.g.
 557    * /hbase/some_table/2323432434/cf/recovered.hfiles/. This method also ensures existence of
 558    * recovered.hfiles directory under the region's column family, creating it if necessary.
 559    * @param rootFS the root file system
 560    * @param conf configuration
 561    * @param tableName the table name
 562    * @param encodedRegionName the encoded region name
 563    * @param familyName the column family name
 564    * @return Path to recovered.hfiles directory of the region's column family.
 565    */
 566   static Path tryCreateRecoveredHFilesDir(FileSystem rootFS, Configuration conf,
 567       TableName tableName, String encodedRegionName, String familyName) throws IOException {
 568     Path rootDir = CommonFSUtils.getRootDir(conf);
 569     Path regionDir = FSUtils.getRegionDirFromTableDir(CommonFSUtils.getTableDir(rootDir, tableName),
 570       encodedRegionName);
 571     Path dir = getRecoveredHFilesDir(regionDir, familyName);
 572     if (!rootFS.exists(dir) && !rootFS.mkdirs(dir)) {
 573       LOG.warn("mkdir failed on {}, region {}, column family {}", dir, encodedRegionName,
 574         familyName);
 575     }
 576     return dir;
 577   }
 578
 579   /**
 580    * @param regionDir  This regions directory in the filesystem
 581    * @param familyName The column family name
 582    * @return The directory that holds recovered hfiles for the region's column family
 583    */
 584   private static Path getRecoveredHFilesDir(final Path regionDir, String familyName) {
 585     return new Path(new Path(regionDir, familyName), HConstants.RECOVERED_HFILES_DIR);
 586   }
 587
 588   public static FileStatus[] getRecoveredHFiles(final FileSystem rootFS,
 589       final Path regionDir, String familyName) throws IOException {
 590     Path dir = getRecoveredHFilesDir(regionDir, familyName);
 591     return CommonFSUtils.listStatus(rootFS, dir);
 592   }
 593 }