From a00adb0b45f9411be2002a1d9b0ea20a046d52a3 Mon Sep 17 00:00:00 2001 From: Zach York Date: Wed, 27 Jun 2018 16:18:53 -0700 Subject: [PATCH] HBASE-20734 Colocate recovered edits directory with hbase.wal.dir Signed-off-by: Andrew Purtell Signed-off-by: Reid Chan --- .../hadoop/hbase/regionserver/HRegion.java | 212 +++++++++++------- .../org/apache/hadoop/hbase/util/FSUtils.java | 27 +++ .../apache/hadoop/hbase/wal/WALSplitter.java | 166 +++++++------- .../hbase/regionserver/TestHRegion.java | 8 +- .../regionserver/TestRecoveredEdits.java | 2 +- .../hbase/regionserver/wal/TestWALReplay.java | 6 +- .../hadoop/hbase/wal/TestWALFactory.java | 2 +- .../apache/hadoop/hbase/wal/TestWALSplit.java | 50 ++--- 8 files changed, 274 insertions(+), 199 deletions(-) diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java index b682b507967..06056636c85 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java @@ -58,6 +58,7 @@ import java.util.NavigableSet; import java.util.RandomAccess; import java.util.Set; import java.util.TreeMap; +import java.util.TreeSet; import java.util.concurrent.Callable; import java.util.concurrent.CompletionService; import java.util.concurrent.ConcurrentHashMap; @@ -326,6 +327,9 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi private final int rowLockWaitDuration; static final int DEFAULT_ROWLOCK_WAIT_DURATION = 30000; + private Path regionDir; + private FileSystem walFS; + // The internal wait duration to acquire a lock before read/update // from the region. It is not per row. The purpose of this wait time // is to avoid waiting a long time while the region is busy, so that @@ -921,7 +925,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi if (ServerRegionReplicaUtil.shouldReplayRecoveredEdits(this)) { // Recover any edits if available. maxSeqId = Math.max(maxSeqId, - replayRecoveredEditsIfAny(this.fs.getRegionDir(), maxSeqIdInStores, reporter, status)); + replayRecoveredEditsIfAny(maxSeqIdInStores, reporter, status)); // Make sure mvcc is up to max. this.mvcc.advanceTo(maxSeqId); } @@ -965,8 +969,8 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi // is opened before recovery completes. So we add a safety bumper to avoid new sequence number // overlaps used sequence numbers if (this.writestate.writesEnabled) { - nextSeqid = WALSplitter.writeRegionSequenceIdFile(this.fs.getFileSystem(), this.fs - .getRegionDir(), nextSeqid, (this.recovering ? (this.flushPerChanges + 10000000) : 1)); + nextSeqid = WALSplitter.writeRegionSequenceIdFile(getWalFileSystem(), getWALRegionDir(), + nextSeqid, (this.recovering ? (this.flushPerChanges + 10000000) : 1)); } else { nextSeqid++; } @@ -1104,11 +1108,11 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi getRegionServerServices().getServerName(), storeFiles); WALUtil.writeRegionEventMarker(wal, getTableDesc(), getRegionInfo(), regionEventDesc, mvcc); - // Store SeqId in HDFS when a region closes + // Store SeqId in WAL FileSystem when a region closes // checking region folder exists is due to many tests which delete the table folder while a // table is still online - if (this.fs.getFileSystem().exists(this.fs.getRegionDir())) { - WALSplitter.writeRegionSequenceIdFile(this.fs.getFileSystem(), this.fs.getRegionDir(), + if (getWalFileSystem().exists(getWALRegionDir())) { + WALSplitter.writeRegionSequenceIdFile(getWalFileSystem(), getWALRegionDir(), mvcc.getReadPoint(), 0); } } @@ -1797,6 +1801,32 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi return this.fs; } + /** @return the WAL {@link HRegionFileSystem} used by this region */ + HRegionFileSystem getRegionWALFileSystem() throws IOException { + return new HRegionFileSystem(conf, getWalFileSystem(), + FSUtils.getWALTableDir(conf, htableDescriptor.getTableName()), fs.getRegionInfo()); + } + + /** @return WAL {@link FileSystem} being used by this region */ + FileSystem getWalFileSystem() throws IOException { + if (walFS == null) { + walFS = FSUtils.getWALFileSystem(conf); + } + return walFS; + } + + /** + * @return the Region Directory under the WALRootDir + * @throws IOException if there is an error getting WALRootDir + */ + @VisibleForTesting + public Path getWALRegionDir() throws IOException { + if (regionDir == null) { + regionDir = FSUtils.getWALRegionDir(conf, fs.getRegionInfo()); + } + return regionDir; + } + @Override public long getEarliestFlushTimeForAllStores() { return Collections.min(lastStoreFlushTimeMap.values()); @@ -4247,8 +4277,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi * recovered edits log or minSeqId if nothing added from editlogs. * @throws IOException */ - protected long replayRecoveredEditsIfAny(final Path regiondir, - Map maxSeqIdInStores, + protected long replayRecoveredEditsIfAny(Map maxSeqIdInStores, final CancelableProgressable reporter, final MonitoredTask status) throws IOException { long minSeqIdForTheRegion = -1; @@ -4259,65 +4288,25 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi } long seqid = minSeqIdForTheRegion; - FileSystem fs = this.fs.getFileSystem(); - NavigableSet files = WALSplitter.getSplitEditFilesSorted(fs, regiondir); - if (LOG.isDebugEnabled()) { - LOG.debug("Found " + (files == null ? 0 : files.size()) - + " recovered edits file(s) under " + regiondir); + FileSystem walFS = getWalFileSystem(); + Path regionDir = getWALRegionDir(); + FileSystem rootFS = getFilesystem(); + Path defaultRegionDir = getRegionDir(FSUtils.getRootDir(conf), getRegionInfo()); + + // This is to ensure backwards compatability with HBASE-20723 where recovered edits can appear + // under the root dir even if walDir is set. + NavigableSet filesUnderRootDir = null; + if (!regionDir.equals(defaultRegionDir)) { + filesUnderRootDir = + WALSplitter.getSplitEditFilesSorted(rootFS, defaultRegionDir); + seqid = Math.max(seqid, + replayRecoveredEditsForPaths(minSeqIdForTheRegion, rootFS, filesUnderRootDir, reporter, + defaultRegionDir)); } + NavigableSet files = WALSplitter.getSplitEditFilesSorted(walFS, regionDir); + seqid = Math.max(seqid, replayRecoveredEditsForPaths(minSeqIdForTheRegion, walFS, + files, reporter, regionDir)); - if (files == null || files.isEmpty()) return seqid; - - for (Path edits: files) { - if (edits == null || !fs.exists(edits)) { - LOG.warn("Null or non-existent edits file: " + edits); - continue; - } - if (isZeroLengthThenDelete(fs, edits)) continue; - - long maxSeqId; - String fileName = edits.getName(); - maxSeqId = Math.abs(Long.parseLong(fileName)); - if (maxSeqId <= minSeqIdForTheRegion) { - if (LOG.isDebugEnabled()) { - String msg = "Maximum sequenceid for this wal is " + maxSeqId - + " and minimum sequenceid for the region is " + minSeqIdForTheRegion - + ", skipped the whole file, path=" + edits; - LOG.debug(msg); - } - continue; - } - - try { - // replay the edits. Replay can return -1 if everything is skipped, only update - // if seqId is greater - seqid = Math.max(seqid, replayRecoveredEdits(edits, maxSeqIdInStores, reporter)); - } catch (IOException e) { - boolean skipErrors = conf.getBoolean( - HConstants.HREGION_EDITS_REPLAY_SKIP_ERRORS, - conf.getBoolean( - "hbase.skip.errors", - HConstants.DEFAULT_HREGION_EDITS_REPLAY_SKIP_ERRORS)); - if (conf.get("hbase.skip.errors") != null) { - LOG.warn( - "The property 'hbase.skip.errors' has been deprecated. Please use " + - HConstants.HREGION_EDITS_REPLAY_SKIP_ERRORS + " instead."); - } - if (skipErrors) { - Path p = WALSplitter.moveAsideBadEditsFile(fs, edits); - LOG.error(HConstants.HREGION_EDITS_REPLAY_SKIP_ERRORS - + "=true so continuing. Renamed " + edits + - " as " + p, e); - } else { - throw e; - } - } - } - // The edits size added into rsAccounting during this replaying will not - // be required any more. So just clear it. - if (this.rsAccounting != null) { - this.rsAccounting.clearRegionReplayEditsSize(getRegionInfo().getRegionName()); - } if (seqid > minSeqIdForTheRegion) { // Then we added some edits to memory. Flush and cleanup split edit files. internalFlushcache(null, seqid, stores.values(), status, false); @@ -4327,16 +4316,25 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi // For debugging data loss issues! // If this flag is set, make use of the hfile archiving by making recovered.edits a fake // column family. Have to fake out file type too by casting our recovered.edits as storefiles - String fakeFamilyName = WALSplitter.getRegionDirRecoveredEditsDir(regiondir).getName(); - Set fakeStoreFiles = new HashSet(files.size()); + String fakeFamilyName = WALSplitter.getRegionDirRecoveredEditsDir(regionDir).getName(); + Set fakeStoreFiles = new HashSet<>(files.size()); for (Path file: files) { - fakeStoreFiles.add(new StoreFile(getRegionFileSystem().getFileSystem(), file, this.conf, - null, null)); + fakeStoreFiles.add( + new StoreFile(walFS, file, this.conf, null, null)); } - getRegionFileSystem().removeStoreFiles(fakeFamilyName, fakeStoreFiles); + getRegionWALFileSystem().removeStoreFiles(fakeFamilyName, fakeStoreFiles); } else { + if (filesUnderRootDir != null) { + for (Path file : filesUnderRootDir) { + if (!rootFS.delete(file, false)) { + LOG.error("Failed delete of {} under root directory." + file); + } else { + LOG.debug("Deleted recovered.edits root directory file=" + file); + } + } + } for (Path file: files) { - if (!fs.delete(file, false)) { + if (!walFS.delete(file, false)) { LOG.error("Failed delete of " + file); } else { LOG.debug("Deleted recovered.edits file=" + file); @@ -4346,22 +4344,80 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi return seqid; } - /* + private long replayRecoveredEditsForPaths(long minSeqIdForTheRegion, FileSystem fs, + final NavigableSet files, final CancelableProgressable reporter, final Path regionDir) + throws IOException { + long seqid = minSeqIdForTheRegion; + if (LOG.isDebugEnabled()) { + LOG.debug("Found " + (files == null ? 0 : files.size()) + + " recovered edits file(s) under " + regionDir); + } + + if (files == null || files.isEmpty()) { + return seqid; + } + + for (Path edits : files) { + if (edits == null || !walFS.exists(edits)) { + LOG.warn("Null or non-existent edits file: " + edits); + continue; + } + if (isZeroLengthThenDelete(walFS, edits)) { + continue; + } + + long maxSeqId; + String fileName = edits.getName(); + maxSeqId = Math.abs(Long.parseLong(fileName)); + if (maxSeqId <= minSeqIdForTheRegion) { + if (LOG.isDebugEnabled()) { + String msg = "Maximum sequenceid for this wal is " + maxSeqId + + " and minimum sequenceid for the region is " + minSeqIdForTheRegion + + ", skipped the whole file, path=" + edits; + LOG.debug(msg); + } + continue; + } + + try { + // replay the edits. Replay can return -1 if everything is skipped, only update + // if seqId is greater + seqid = Math.max(seqid, replayRecoveredEdits(edits, maxSeqIdInStores, reporter, fs)); + } catch (IOException e) { + boolean skipErrors = conf.getBoolean(HConstants.HREGION_EDITS_REPLAY_SKIP_ERRORS, + conf.getBoolean("hbase.skip.errors", + HConstants.DEFAULT_HREGION_EDITS_REPLAY_SKIP_ERRORS)); + if (conf.get("hbase.skip.errors") != null) { + LOG.warn("The property 'hbase.skip.errors' has been deprecated. Please use " + + HConstants.HREGION_EDITS_REPLAY_SKIP_ERRORS + " instead."); + } + if (skipErrors) { + Path p = WALSplitter.moveAsideBadEditsFile(walFS, edits); + LOG.error(HConstants.HREGION_EDITS_REPLAY_SKIP_ERRORS + + "=true so continuing. Renamed " + edits + " as " + p, e); + } else { + throw e; + } + } + } + return seqid; + } + + /** * @param edits File of recovered edits. * @param maxSeqIdInStores Maximum sequenceid found in each store. Edits in wal - * must be larger than this to be replayed for each store. - * @param reporter + * must be larger than this to be replayed for each store. + * @param reporter CacelableProgressable reporter * @return the sequence id of the last edit added to this region out of the * recovered edits log or minSeqId if nothing added from editlogs. * @throws IOException */ - private long replayRecoveredEdits(final Path edits, - Map maxSeqIdInStores, final CancelableProgressable reporter) + private long replayRecoveredEdits(final Path edits, Map maxSeqIdInStores, + final CancelableProgressable reporter, final FileSystem fs) throws IOException { String msg = "Replaying edits from " + edits; LOG.info(msg); MonitoredTask status = TaskMonitor.get().createStatus(msg); - FileSystem fs = this.fs.getFileSystem(); status.setStatus("Opening recovered edits"); WAL.Reader reader = null; @@ -8316,7 +8372,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi public static final long FIXED_OVERHEAD = ClassSize.align( ClassSize.OBJECT + ClassSize.ARRAY + - 48 * ClassSize.REFERENCE + 3 * Bytes.SIZEOF_INT + + 50 * ClassSize.REFERENCE + 3 * Bytes.SIZEOF_INT + (15 * Bytes.SIZEOF_LONG) + 5 * Bytes.SIZEOF_BOOLEAN); diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/util/FSUtils.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/util/FSUtils.java index a6fccfe319f..f558618de2b 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/util/FSUtils.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/util/FSUtils.java @@ -1163,6 +1163,20 @@ public abstract class FSUtils { return true; } + /** + * Returns the WAL region directory based on the region info + * @param conf configuration to determine WALRootDir + * @param regionInfo used to get region and table + * @return the region directory used to store WALs under the WALRootDir + * @throws IOException if there is an exception determining the WALRootDir + */ + public static Path getWALRegionDir(final Configuration conf, + final HRegionInfo regionInfo) + throws IOException { + return new Path(getWALTableDir(conf, regionInfo.getTable()), + regionInfo.getEncodedName()); + } + /** * Checks if meta region exists * @@ -1313,6 +1327,19 @@ public abstract class FSUtils { tableName.getQualifierAsString()); } + /** + * Returns the Table directory under the WALRootDir for the specified table name + * @param conf configuration used to get the WALRootDir + * @param tableName Table to get the directory for + * @return a path to the WAL table directory for the specified table + * @throws IOException if there is an exception determining the WALRootDir + */ + public static Path getWALTableDir(final Configuration conf, final TableName tableName) + throws IOException { + return new Path(new Path(getWALRootDir(conf), tableName.getNamespaceAsString()), + tableName.getQualifierAsString()); + } + /** * Returns the {@link org.apache.hadoop.hbase.TableName} object representing * the table directory under diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALSplitter.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALSplitter.java index 735549bb54d..6a6ceaa0b00 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALSplitter.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALSplitter.java @@ -143,7 +143,7 @@ public class WALSplitter { // Parameters for split process protected final Path walDir; - protected final FileSystem fs; + protected final FileSystem walFS; protected final Configuration conf; // Major subcomponents of the split process. @@ -190,14 +190,14 @@ public class WALSplitter { @VisibleForTesting WALSplitter(final WALFactory factory, Configuration conf, Path walDir, - FileSystem fs, LastSequenceId idChecker, + FileSystem walFS, LastSequenceId idChecker, CoordinatedStateManager csm, RecoveryMode mode) { this.conf = HBaseConfiguration.create(conf); String codecClassName = conf .get(WALCellCodec.WAL_CELL_CODEC_CLASS_KEY, WALCellCodec.class.getName()); this.conf.set(HConstants.RPC_CODEC_CONF_KEY, codecClassName); this.walDir = walDir; - this.fs = fs; + this.walFS = walFS; this.sequenceIdChecker = idChecker; this.csm = (BaseCoordinatedStateManager)csm; this.walFactory = factory; @@ -238,7 +238,7 @@ public class WALSplitter { *

* @param walDir * @param logfile - * @param fs + * @param walFS FileSystem to use for WAL reading and splitting * @param conf * @param reporter * @param idChecker @@ -246,10 +246,10 @@ public class WALSplitter { * @return false if it is interrupted by the progress-able. * @throws IOException */ - public static boolean splitLogFile(Path walDir, FileStatus logfile, FileSystem fs, + public static boolean splitLogFile(Path walDir, FileStatus logfile, FileSystem walFS, Configuration conf, CancelableProgressable reporter, LastSequenceId idChecker, CoordinatedStateManager cp, RecoveryMode mode, final WALFactory factory) throws IOException { - WALSplitter s = new WALSplitter(factory, conf, walDir, fs, idChecker, cp, mode); + WALSplitter s = new WALSplitter(factory, conf, walDir, walFS, idChecker, cp, mode); return s.splitLogFile(logfile, reporter); } @@ -317,7 +317,7 @@ public class WALSplitter { in = getReader(logfile, skipErrors, reporter); } catch (CorruptedLogFileException e) { LOG.warn("Could not get reader, corrupted log file " + logPath, e); - ZKSplitLog.markCorrupted(walDir, logfile.getPath().getName(), fs); + ZKSplitLog.markCorrupted(walDir, logfile.getPath().getName(), walFS); isCorrupted = true; } if (in == null) { @@ -409,7 +409,7 @@ public class WALSplitter { } catch (CorruptedLogFileException e) { LOG.warn("Could not parse, corrupted log file " + logPath, e); csm.getSplitLogWorkerCoordination().markCorrupted(walDir, - logfile.getPath().getName(), fs); + logfile.getPath().getName(), walFS); isCorrupted = true; } catch (IOException e) { e = RemoteExceptionHandler.checkIOException(e); @@ -457,31 +457,30 @@ public class WALSplitter { */ public static void finishSplitLogFile(String logfile, Configuration conf) throws IOException { - Path rootdir = FSUtils.getWALRootDir(conf); - Path oldLogDir = new Path(rootdir, HConstants.HREGION_OLDLOGDIR_NAME); + Path walDir = FSUtils.getWALRootDir(conf); + Path oldLogDir = new Path(walDir, HConstants.HREGION_OLDLOGDIR_NAME); Path logPath; - if (FSUtils.isStartingWithPath(rootdir, logfile)) { + if (FSUtils.isStartingWithPath(walDir, logfile)) { logPath = new Path(logfile); } else { - logPath = new Path(rootdir, logfile); + logPath = new Path(walDir, logfile); } - finishSplitLogFile(rootdir, oldLogDir, logPath, conf); + finishSplitLogFile(walDir, oldLogDir, logPath, conf); } - private static void finishSplitLogFile(Path rootdir, Path oldLogDir, + private static void finishSplitLogFile(Path walDir, Path oldLogDir, Path logPath, Configuration conf) throws IOException { List processedLogs = new ArrayList(); List corruptedLogs = new ArrayList(); - FileSystem fs; - fs = rootdir.getFileSystem(conf); - if (ZKSplitLog.isCorrupted(rootdir, logPath.getName(), fs)) { + FileSystem walFS = walDir.getFileSystem(conf); + if (ZKSplitLog.isCorrupted(walDir, logPath.getName(), walFS)) { corruptedLogs.add(logPath); } else { processedLogs.add(logPath); } - archiveLogs(corruptedLogs, processedLogs, oldLogDir, fs, conf); - Path stagingDir = ZKSplitLog.getSplitLogDir(rootdir, logPath.getName()); - fs.delete(stagingDir, true); + archiveLogs(corruptedLogs, processedLogs, oldLogDir, walFS, conf); + Path stagingDir = ZKSplitLog.getSplitLogDir(walDir, logPath.getName()); + walFS.delete(stagingDir, true); } /** @@ -492,28 +491,28 @@ public class WALSplitter { * @param corruptedLogs * @param processedLogs * @param oldLogDir - * @param fs + * @param walFS FileSystem to use for WAL archival * @param conf * @throws IOException */ private static void archiveLogs( final List corruptedLogs, final List processedLogs, final Path oldLogDir, - final FileSystem fs, final Configuration conf) throws IOException { + final FileSystem walFS, final Configuration conf) throws IOException { final Path corruptDir = new Path(FSUtils.getWALRootDir(conf), conf.get( "hbase.regionserver.hlog.splitlog.corrupt.dir", HConstants.CORRUPT_DIR_NAME)); - if (!fs.mkdirs(corruptDir)) { + if (!walFS.mkdirs(corruptDir)) { LOG.info("Unable to mkdir " + corruptDir); } - fs.mkdirs(oldLogDir); + walFS.mkdirs(oldLogDir); // this method can get restarted or called multiple times for archiving // the same log files. for (Path corrupted : corruptedLogs) { Path p = new Path(corruptDir, corrupted.getName()); - if (fs.exists(corrupted)) { - if (!fs.rename(corrupted, p)) { + if (walFS.exists(corrupted)) { + if (!walFS.rename(corrupted, p)) { LOG.warn("Unable to move corrupted log " + corrupted + " to " + p); } else { LOG.warn("Moved corrupted log " + corrupted + " to " + p); @@ -523,8 +522,8 @@ public class WALSplitter { for (Path p : processedLogs) { Path newPath = FSHLog.getWALArchivePath(oldLogDir, p); - if (fs.exists(p)) { - if (!FSUtils.renameAndSetModifyTime(fs, p, newPath)) { + if (walFS.exists(p)) { + if (!FSUtils.renameAndSetModifyTime(walFS, p, newPath)) { LOG.warn("Unable to move " + p + " to " + newPath); } else { LOG.info("Archived processed log " + p + " to " + newPath); @@ -550,35 +549,28 @@ public class WALSplitter { @VisibleForTesting static Path getRegionSplitEditsPath(final Entry logEntry, String fileNameBeingSplit, String tmpDirName, Configuration conf) throws IOException { - FileSystem fs = FileSystem.get(conf); - Path rootDir = FSUtils.getRootDir(conf); - Path tableDir = FSUtils.getTableDir(rootDir, logEntry.getKey().getTablename()); + FileSystem walFS = FSUtils.getWALFileSystem(conf); + Path tableDir = FSUtils.getWALTableDir(conf, logEntry.getKey().getTablename()); String encodedRegionName = Bytes.toString(logEntry.getKey().getEncodedRegionName()); - Path regiondir = HRegion.getRegionDir(tableDir, encodedRegionName); - Path dir = getRegionDirRecoveredEditsDir(regiondir); + Path regionDir = HRegion.getRegionDir(tableDir, encodedRegionName); + Path dir = getRegionDirRecoveredEditsDir(regionDir); - if (!fs.exists(regiondir)) { - LOG.info("This region's directory doesn't exist: " - + regiondir.toString() + ". It is very likely that it was" + - " already split so it's safe to discard those edits."); - return null; - } - if (fs.exists(dir) && fs.isFile(dir)) { + if (walFS.exists(dir) && walFS.isFile(dir)) { Path tmp = new Path(tmpDirName); - if (!fs.exists(tmp)) { - fs.mkdirs(tmp); + if (!walFS.exists(tmp)) { + walFS.mkdirs(tmp); } tmp = new Path(tmp, HConstants.RECOVERED_EDITS_DIR + "_" + encodedRegionName); LOG.warn("Found existing old file: " + dir + ". It could be some " + "leftover of an old installation. It should be a folder instead. " + "So moving it to " + tmp); - if (!fs.rename(dir, tmp)) { + if (!walFS.rename(dir, tmp)) { LOG.warn("Failed to sideline old file " + dir); } } - if (!fs.exists(dir) && !fs.mkdirs(dir)) { + if (!walFS.exists(dir) && !walFS.mkdirs(dir)) { LOG.warn("mkdir failed on " + dir); } // Append fileBeingSplit to prevent name conflict since we may have duplicate wal entries now. @@ -616,31 +608,32 @@ public class WALSplitter { private static final String RECOVERED_LOG_TMPFILE_SUFFIX = ".temp"; /** - * @param regiondir + * @param regionDir * This regions directory in the filesystem. * @return The directory that holds recovered edits files for the region - * regiondir + * regionDir */ - public static Path getRegionDirRecoveredEditsDir(final Path regiondir) { - return new Path(regiondir, HConstants.RECOVERED_EDITS_DIR); + public static Path getRegionDirRecoveredEditsDir(final Path regionDir) { + return new Path(regionDir, HConstants.RECOVERED_EDITS_DIR); } /** * Returns sorted set of edit files made by splitter, excluding files * with '.temp' suffix. * - * @param fs - * @param regiondir - * @return Files in passed regiondir as a sorted set. + * @param walFS FileSystem to use for reading Recovered edits files + * @param regionDir Directory where Recovered edits should reside + * @return Files in passed regionDir as a sorted set. * @throws IOException */ - public static NavigableSet getSplitEditFilesSorted(final FileSystem fs, - final Path regiondir) throws IOException { + public static NavigableSet getSplitEditFilesSorted(final FileSystem walFS, + final Path regionDir) throws IOException { NavigableSet filesSorted = new TreeSet(); - Path editsdir = getRegionDirRecoveredEditsDir(regiondir); - if (!fs.exists(editsdir)) + Path editsdir = getRegionDirRecoveredEditsDir(regionDir); + if (!walFS.exists(editsdir)) { return filesSorted; - FileStatus[] files = FSUtils.listStatus(fs, editsdir, new PathFilter() { + } + FileStatus[] files = FSUtils.listStatus(walFS, editsdir, new PathFilter() { @Override public boolean accept(Path p) { boolean result = false; @@ -650,7 +643,7 @@ public class WALSplitter { // In particular, on error, we'll move aside the bad edit file giving // it a timestamp suffix. See moveAsideBadEditsFile. Matcher m = EDITFILES_NAME_PATTERN.matcher(p.getName()); - result = fs.isFile(p) && m.matches(); + result = walFS.isFile(p) && m.matches(); // Skip the file whose name ends with RECOVERED_LOG_TMPFILE_SUFFIX, // because it means splitwal thread is writting this file. if (p.getName().endsWith(RECOVERED_LOG_TMPFILE_SUFFIX)) { @@ -678,17 +671,17 @@ public class WALSplitter { /** * Move aside a bad edits file. * - * @param fs + * @param walFS FileSystem to use for WAL operations * @param edits * Edits file to move aside. * @return The name of the moved aside file. * @throws IOException */ - public static Path moveAsideBadEditsFile(final FileSystem fs, final Path edits) + public static Path moveAsideBadEditsFile(final FileSystem walFS, final Path edits) throws IOException { Path moveAsideName = new Path(edits.getParent(), edits.getName() + "." + System.currentTimeMillis()); - if (!fs.rename(edits, moveAsideName)) { + if (!walFS.rename(edits, moveAsideName)) { LOG.warn("Rename failed from " + edits + " to " + moveAsideName); } return moveAsideName; @@ -709,21 +702,21 @@ public class WALSplitter { /** * Create a file with name as region open sequence id - * @param fs - * @param regiondir + * @param walFS FileSystem to write Sequence file to + * @param regionDir WALRegionDir used to determine where to write edits files * @param newSeqId * @param saftyBumper * @return long new sequence Id value * @throws IOException */ - public static long writeRegionSequenceIdFile(final FileSystem fs, final Path regiondir, + public static long writeRegionSequenceIdFile(final FileSystem walFS, final Path regionDir, long newSeqId, long saftyBumper) throws IOException { - Path editsdir = WALSplitter.getRegionDirRecoveredEditsDir(regiondir); + Path editsdir = WALSplitter.getRegionDirRecoveredEditsDir(regionDir); long maxSeqId = 0; FileStatus[] files = null; - if (fs.exists(editsdir)) { - files = FSUtils.listStatus(fs, editsdir, new PathFilter() { + if (walFS.exists(editsdir)) { + files = FSUtils.listStatus(walFS, editsdir, new PathFilter() { @Override public boolean accept(Path p) { return isSequenceIdFile(p); @@ -751,7 +744,7 @@ public class WALSplitter { Path newSeqIdFile = new Path(editsdir, newSeqId + SEQUENCE_ID_FILE_SUFFIX); if (newSeqId != maxSeqId) { try { - if (!fs.createNewFile(newSeqIdFile) && !fs.exists(newSeqIdFile)) { + if (!walFS.createNewFile(newSeqIdFile) && !walFS.exists(newSeqIdFile)) { throw new IOException("Failed to create SeqId file:" + newSeqIdFile); } if (LOG.isDebugEnabled()) { @@ -768,7 +761,7 @@ public class WALSplitter { if (newSeqIdFile.equals(status.getPath())) { continue; } - fs.delete(status.getPath(), false); + walFS.delete(status.getPath(), false); } } return newSeqId; @@ -796,7 +789,7 @@ public class WALSplitter { } try { - FSUtils.getInstance(fs, conf).recoverFileLease(fs, path, conf, reporter); + FSUtils.getInstance(walFS, conf).recoverFileLease(walFS, path, conf, reporter); try { in = getReader(path, reporter); } catch (EOFException e) { @@ -866,7 +859,7 @@ public class WALSplitter { */ protected Writer createWriter(Path logfile) throws IOException { - return walFactory.createRecoveredEditsWriter(fs, logfile); + return walFactory.createRecoveredEditsWriter(walFS, logfile); } /** @@ -874,7 +867,7 @@ public class WALSplitter { * @return new Reader instance, caller should close */ protected Reader getReader(Path curLogFile, CancelableProgressable reporter) throws IOException { - return walFactory.createReader(fs, curLogFile, reporter); + return walFactory.createReader(walFS, curLogFile, reporter); } /** @@ -1357,10 +1350,10 @@ public class WALSplitter { } // delete the one with fewer wal entries - void deleteOneWithFewerEntries(FileSystem rootFs, WriterAndPath wap, Path dst) + void deleteOneWithFewerEntries(WriterAndPath wap, Path dst) throws IOException { long dstMinLogSeqNum = -1L; - try (WAL.Reader reader = walFactory.createReader(fs, dst)) { + try (WAL.Reader reader = walFactory.createReader(walFS, dst)) { WAL.Entry entry = reader.next(); if (entry != null) { dstMinLogSeqNum = entry.getKey().getLogSeqNum(); @@ -1375,15 +1368,15 @@ public class WALSplitter { if (wap.minLogSeqNum < dstMinLogSeqNum) { LOG.warn("Found existing old edits file. It could be the result of a previous failed" + " split attempt or we have duplicated wal entries. Deleting " + dst + ", length=" - + fs.getFileStatus(dst).getLen()); - if (!fs.delete(dst, false)) { + + walFS.getFileStatus(dst).getLen()); + if (!walFS.delete(dst, false)) { LOG.warn("Failed deleting of old " + dst); throw new IOException("Failed deleting of old " + dst); } } else { LOG.warn("Found existing old edits file and we have less entries. Deleting " + wap.p - + ", length=" + rootFs.getFileStatus(wap.p).getLen()); - if (!rootFs.delete(wap.p, false)) { + + ", length=" + walFS.getFileStatus(wap.p).getLen()); + if (!walFS.delete(wap.p, false)) { LOG.warn("Failed deleting of " + wap.p); throw new IOException("Failed deleting of " + wap.p); } @@ -1467,7 +1460,7 @@ public class WALSplitter { if (LOG.isTraceEnabled()) { LOG.trace("Closing " + wap.p); } - FileSystem rootFs = FileSystem.get(conf); + try { wap.w.close(); } catch (IOException ioe) { @@ -1482,7 +1475,7 @@ public class WALSplitter { } if (wap.editsWritten == 0) { // just remove the empty recovered.edits file - if (rootFs.exists(wap.p) && !rootFs.delete(wap.p, false)) { + if (walFS.exists(wap.p) && !walFS.delete(wap.p, false)) { LOG.warn("Failed deleting empty " + wap.p); throw new IOException("Failed deleting empty " + wap.p); } @@ -1492,14 +1485,14 @@ public class WALSplitter { Path dst = getCompletedRecoveredEditsFilePath(wap.p, regionMaximumEditLogSeqNum.get(encodedRegionName)); try { - if (!dst.equals(wap.p) && rootFs.exists(dst)) { - deleteOneWithFewerEntries(rootFs, wap, dst); + if (!dst.equals(wap.p) && walFS.exists(dst)) { + deleteOneWithFewerEntries(wap, dst); } // Skip the unit tests which create a splitter that reads and // writes the data without touching disk. // TestHLogSplit#testThreading is an example. - if (rootFs.exists(wap.p)) { - if (!rootFs.rename(wap.p, dst)) { + if (walFS.exists(wap.p)) { + if (!walFS.rename(wap.p, dst)) { throw new IOException("Failed renaming " + wap.p + " to " + dst); } LOG.info("Rename " + wap.p + " to " + dst); @@ -1596,12 +1589,11 @@ public class WALSplitter { if (regionedits == null) { return null; } - FileSystem rootFs = FileSystem.get(conf); - if (rootFs.exists(regionedits)) { + if (walFS.exists(regionedits)) { LOG.warn("Found old edits file. It could be the " + "result of a previous failed split attempt. Deleting " + regionedits + ", length=" - + rootFs.getFileStatus(regionedits).getLen()); - if (!rootFs.delete(regionedits, false)) { + + walFS.getFileStatus(regionedits).getLen()); + if (!walFS.delete(regionedits, false)) { LOG.warn("Failed delete of old " + regionedits); } } diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java index 69db63c073d..48ee82545ca 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java @@ -698,7 +698,7 @@ public class TestHRegion { for (Store store : region.getStores()) { maxSeqIdInStores.put(store.getColumnFamilyName().getBytes(), minSeqId - 1); } - long seqId = region.replayRecoveredEditsIfAny(regiondir, maxSeqIdInStores, null, status); + long seqId = region.replayRecoveredEditsIfAny(maxSeqIdInStores, null, status); assertEquals(maxSeqId, seqId); region.getMVCC().advanceTo(seqId); Get get = new Get(row); @@ -752,7 +752,7 @@ public class TestHRegion { for (Store store : region.getStores()) { maxSeqIdInStores.put(store.getColumnFamilyName().getBytes(), recoverSeqId - 1); } - long seqId = region.replayRecoveredEditsIfAny(regiondir, maxSeqIdInStores, null, status); + long seqId = region.replayRecoveredEditsIfAny(maxSeqIdInStores, null, status); assertEquals(maxSeqId, seqId); region.getMVCC().advanceTo(seqId); Get get = new Get(row); @@ -796,7 +796,7 @@ public class TestHRegion { for (Store store : region.getStores()) { maxSeqIdInStores.put(store.getColumnFamilyName().getBytes(), minSeqId); } - long seqId = region.replayRecoveredEditsIfAny(regiondir, maxSeqIdInStores, null, null); + long seqId = region.replayRecoveredEditsIfAny(maxSeqIdInStores, null, null); assertEquals(minSeqId, seqId); } @@ -852,7 +852,7 @@ public class TestHRegion { for (Store store : region.getStores()) { maxSeqIdInStores.put(store.getColumnFamilyName().getBytes(), recoverSeqId - 1); } - long seqId = region.replayRecoveredEditsIfAny(regiondir, maxSeqIdInStores, null, status); + long seqId = region.replayRecoveredEditsIfAny(maxSeqIdInStores, null, status); assertEquals(maxSeqId, seqId); // assert that the files are flushed diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestRecoveredEdits.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestRecoveredEdits.java index bf723012913..c3f21f33e3a 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestRecoveredEdits.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestRecoveredEdits.java @@ -112,7 +112,7 @@ public class TestRecoveredEdits { // There should be no store files. assertTrue(storeFiles.isEmpty()); region.close(); - Path regionDir = region.getRegionDir(hbaseRootDir, hri); + Path regionDir = region.getWALRegionDir(); Path recoveredEditsDir = WALSplitter.getRegionDirRecoveredEditsDir(regionDir); // This is a little fragile getting this path to a file of 10M of edits. Path recoveredEditsFile = new Path( diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestWALReplay.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestWALReplay.java index c5cc3726981..333b6055a15 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestWALReplay.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestWALReplay.java @@ -288,7 +288,7 @@ public class TestWALReplay { MultiVersionConcurrencyControl mvcc = new MultiVersionConcurrencyControl(); HRegionInfo hri = createBasic3FamilyHRegionInfo(tableName); - Path basedir = FSUtils.getTableDir(hbaseRootDir, tableName); + Path basedir = FSUtils.getWALTableDir(conf, tableName); deleteDir(basedir); HTableDescriptor htd = createBasic3FamilyHTD(tableName); @@ -915,7 +915,7 @@ public class TestWALReplay { WALSplitter.splitLogFile(hbaseWALRootDir, listStatus[0], this.fs, this.conf, null, null, null, mode, wals); FileStatus[] listStatus1 = this.fs.listStatus( - new Path(FSUtils.getTableDir(hbaseRootDir, tableName), new Path(hri.getEncodedName(), + new Path(FSUtils.getWALTableDir(this.conf, tableName), new Path(hri.getEncodedName(), "recovered.edits")), new PathFilter() { @Override public boolean accept(Path p) { @@ -949,7 +949,7 @@ public class TestWALReplay { final int countPerFamily = 10; final HTableDescriptor htd = createBasic1FamilyHTD(tableName); HRegion region1 = HBaseTestingUtility.createRegionAndWAL(hri, hbaseRootDir, this.hbaseWALRootDir, this.conf, htd); - Path regionDir = region1.getRegionFileSystem().getRegionDir(); + Path regionDir = region1.getWALRegionDir(); HBaseTestingUtility.closeRegionAndWAL(region1); WAL wal = createWAL(this.conf); diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/wal/TestWALFactory.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/wal/TestWALFactory.java index d258d1ddc7a..4bf211dcfa6 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/wal/TestWALFactory.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/wal/TestWALFactory.java @@ -172,7 +172,7 @@ public class TestWALFactory { Path oldLogDir = new Path(hbaseWALDir, HConstants.HREGION_OLDLOGDIR_NAME); final int howmany = 3; HRegionInfo[] infos = new HRegionInfo[3]; - Path tabledir = FSUtils.getTableDir(hbaseDir, tableName); + Path tabledir = FSUtils.getWALTableDir(conf, tableName); fs.mkdirs(tabledir); for(int i = 0; i < howmany; i++) { infos[i] = new HRegionInfo(tableName, diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/wal/TestWALSplit.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/wal/TestWALSplit.java index b6cfdc8a0de..83022c73095 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/wal/TestWALSplit.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/wal/TestWALSplit.java @@ -247,9 +247,9 @@ public class TestWALSplit { } LOG.debug(ls); LOG.info("Splitting WALs out from under zombie. Expecting " + numWriters + " files."); - WALSplitter.split(HBASEDIR, WALDIR, OLDLOGDIR, fs, conf2, wals); + WALSplitter.split(HBASELOGDIR, WALDIR, OLDLOGDIR, fs, conf2, wals); LOG.info("Finished splitting out from under zombie."); - Path[] logfiles = getLogForRegion(HBASEDIR, TABLE_NAME, region); + Path[] logfiles = getLogForRegion(TABLE_NAME, region); assertEquals("wrong number of split files for region", numWriters, logfiles.length); int count = 0; for (Path logfile: logfiles) { @@ -435,9 +435,9 @@ public class TestWALSplit { generateWALs(1, 10, -1, 0); useDifferentDFSClient(); - WALSplitter.split(HBASEDIR, WALDIR, OLDLOGDIR, fs, conf, wals); + WALSplitter.split(HBASELOGDIR, WALDIR, OLDLOGDIR, fs, conf, wals); Path originalLog = (fs.listStatus(OLDLOGDIR))[0].getPath(); - Path[] splitLog = getLogForRegion(HBASEDIR, TABLE_NAME, REGION); + Path[] splitLog = getLogForRegion(TABLE_NAME, REGION); assertEquals(1, splitLog.length); assertTrue("edits differ after split", logsAreEqual(originalLog, splitLog[0])); @@ -451,9 +451,9 @@ public class TestWALSplit { generateWALs(1, 10, -1, 100); useDifferentDFSClient(); - WALSplitter.split(HBASEDIR, WALDIR, OLDLOGDIR, fs, conf, wals); + WALSplitter.split(HBASELOGDIR, WALDIR, OLDLOGDIR, fs, conf, wals); Path originalLog = (fs.listStatus(OLDLOGDIR))[0].getPath(); - Path[] splitLog = getLogForRegion(HBASEDIR, TABLE_NAME, REGION); + Path[] splitLog = getLogForRegion(TABLE_NAME, REGION); assertEquals(1, splitLog.length); assertFalse("edits differ after split", logsAreEqual(originalLog, splitLog[0])); @@ -478,13 +478,13 @@ public class TestWALSplit { writer.close(); useDifferentDFSClient(); - WALSplitter.split(HBASEDIR, WALDIR, OLDLOGDIR, fs, conf, wals); + WALSplitter.split(HBASELOGDIR, WALDIR, OLDLOGDIR, fs, conf, wals); Path originalLog = (fs.listStatus(OLDLOGDIR))[0].getPath(); // original log should have 10 test edits, 10 region markers, 1 compaction marker assertEquals(21, countWAL(originalLog)); - Path[] splitLog = getLogForRegion(HBASEDIR, TABLE_NAME, hri.getEncodedName()); + Path[] splitLog = getLogForRegion(TABLE_NAME, hri.getEncodedName()); assertEquals(1, splitLog.length); assertFalse("edits differ after split", logsAreEqual(originalLog, splitLog[0])); @@ -499,10 +499,10 @@ public class TestWALSplit { private int splitAndCount(final int expectedFiles, final int expectedEntries) throws IOException { useDifferentDFSClient(); - WALSplitter.split(HBASEDIR, WALDIR, OLDLOGDIR, fs, conf, wals); + WALSplitter.split(HBASELOGDIR, WALDIR, OLDLOGDIR, fs, conf, wals); int result = 0; for (String region : REGIONS) { - Path[] logfiles = getLogForRegion(HBASEDIR, TABLE_NAME, region); + Path[] logfiles = getLogForRegion(TABLE_NAME, region); assertEquals(expectedFiles, logfiles.length); int count = 0; for (Path logfile: logfiles) { @@ -633,7 +633,7 @@ public class TestWALSplit { walDirContents.add(status.getPath().getName()); } useDifferentDFSClient(); - WALSplitter.split(HBASEDIR, WALDIR, OLDLOGDIR, fs, conf, wals); + WALSplitter.split(HBASELOGDIR, WALDIR, OLDLOGDIR, fs, conf, wals); return walDirContents; } finally { conf.setClass("hbase.regionserver.hlog.reader.impl", backupClass, @@ -674,9 +674,9 @@ public class TestWALSplit { corruptWAL(c1, corruption, true); useDifferentDFSClient(); - WALSplitter.split(HBASEDIR, WALDIR, OLDLOGDIR, fs, conf, wals); + WALSplitter.split(HBASELOGDIR, WALDIR, OLDLOGDIR, fs, conf, wals); - Path[] splitLog = getLogForRegion(HBASEDIR, TABLE_NAME, REGION); + Path[] splitLog = getLogForRegion(TABLE_NAME, REGION); assertEquals(1, splitLog.length); int actualCount = 0; @@ -710,7 +710,7 @@ public class TestWALSplit { conf.setBoolean(HBASE_SKIP_ERRORS, false); generateWALs(-1); useDifferentDFSClient(); - WALSplitter.split(HBASEDIR, WALDIR, OLDLOGDIR, fs, conf, wals); + WALSplitter.split(HBASELOGDIR, WALDIR, OLDLOGDIR, fs, conf, wals); FileStatus[] archivedLogs = fs.listStatus(OLDLOGDIR); assertEquals("wrong number of files in the archive log", NUM_WRITERS, archivedLogs.length); } @@ -726,7 +726,7 @@ public class TestWALSplit { throws IOException { generateWALs(-1); useDifferentDFSClient(); - WALSplitter.split(HBASEDIR, WALDIR, OLDLOGDIR, fs, conf, wals); + WALSplitter.split(HBASELOGDIR, WALDIR, OLDLOGDIR, fs, conf, wals); FileStatus [] statuses = null; try { statuses = fs.listStatus(WALDIR); @@ -756,7 +756,7 @@ public class TestWALSplit { try { InstrumentedLogWriter.activateFailure = true; - WALSplitter.split(HBASEDIR, WALDIR, OLDLOGDIR, fs, conf, wals); + WALSplitter.split(HBASELOGDIR, WALDIR, OLDLOGDIR, fs, conf, wals); } catch (IOException e) { assertTrue(e.getMessage(). contains("This exception is instrumented and should only be thrown for testing")); @@ -777,7 +777,7 @@ public class TestWALSplit { Path regiondir = new Path(TABLEDIR, region); fs.delete(regiondir, true); - WALSplitter.split(HBASEDIR, WALDIR, OLDLOGDIR, fs, conf, wals); + WALSplitter.split(HBASELOGDIR, WALDIR, OLDLOGDIR, fs, conf, wals); assertFalse(fs.exists(regiondir)); } @@ -854,7 +854,7 @@ public class TestWALSplit { useDifferentDFSClient(); try { - WALSplitter.split(HBASEDIR, WALDIR, OLDLOGDIR, spiedFs, conf, wals); + WALSplitter.split(HBASELOGDIR, WALDIR, OLDLOGDIR, spiedFs, conf, wals); assertEquals(NUM_WRITERS, fs.listStatus(OLDLOGDIR).length); assertFalse(fs.exists(WALDIR)); } catch (IOException e) { @@ -1077,7 +1077,7 @@ public class TestWALSplit { Path regiondir = new Path(TABLEDIR, REGION); LOG.info("Region directory is" + regiondir); fs.delete(regiondir, true); - WALSplitter.split(HBASEDIR, WALDIR, OLDLOGDIR, fs, conf, wals); + WALSplitter.split(HBASELOGDIR, WALDIR, OLDLOGDIR, fs, conf, wals); assertFalse(fs.exists(regiondir)); } @@ -1090,7 +1090,7 @@ public class TestWALSplit { injectEmptyFile(".empty", true); useDifferentDFSClient(); - WALSplitter.split(HBASEDIR, WALDIR, OLDLOGDIR, fs, conf, wals); + WALSplitter.split(HBASELOGDIR, WALDIR, OLDLOGDIR, fs, conf, wals); Path tdir = FSUtils.getTableDir(HBASEDIR, TABLE_NAME); assertFalse(fs.exists(tdir)); @@ -1115,7 +1115,7 @@ public class TestWALSplit { Corruptions.INSERT_GARBAGE_ON_FIRST_LINE, true); useDifferentDFSClient(); - WALSplitter.split(HBASEDIR, WALDIR, OLDLOGDIR, fs, conf, wals); + WALSplitter.split(HBASELOGDIR, WALDIR, OLDLOGDIR, fs, conf, wals); final Path corruptDir = new Path(FSUtils.getWALRootDir(conf), HConstants.CORRUPT_DIR_NAME); assertEquals(1, fs.listStatus(corruptDir).length); @@ -1145,14 +1145,14 @@ public class TestWALSplit { @Override protected Writer createWriter(Path logfile) throws IOException { - Writer writer = wals.createRecoveredEditsWriter(this.fs, logfile); + Writer writer = wals.createRecoveredEditsWriter(this.walFS, logfile); // After creating writer, simulate region's // replayRecoveredEditsIfAny() which gets SplitEditFiles of this // region and delete them, excluding files with '.temp' suffix. NavigableSet files = WALSplitter.getSplitEditFilesSorted(fs, regiondir); if (files != null && !files.isEmpty()) { for (Path file : files) { - if (!this.fs.delete(file, false)) { + if (!this.walFS.delete(file, false)) { LOG.error("Failed delete of " + file); } else { LOG.debug("Deleted recovered.edits file=" + file); @@ -1231,9 +1231,9 @@ public class TestWALSplit { - private Path[] getLogForRegion(Path rootdir, TableName table, String region) + private Path[] getLogForRegion(TableName table, String region) throws IOException { - Path tdir = FSUtils.getTableDir(rootdir, table); + Path tdir = FSUtils.getWALTableDir(conf, table); @SuppressWarnings("deprecation") Path editsdir = WALSplitter.getRegionDirRecoveredEditsDir(HRegion.getRegionDir(tdir, Bytes.toString(region.getBytes())));