You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by ki...@apache.org on 2017/06/01 17:26:42 UTC
hadoop git commit: HDFS-5042. Completed files lost after power
failure. Contributed by Vinayakumar B.
Repository: hadoop
Updated Branches:
refs/heads/branch-2.8 0cba28226 -> 89d59c292
HDFS-5042. Completed files lost after power failure. Contributed by Vinayakumar B.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/89d59c29
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/89d59c29
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/89d59c29
Branch: refs/heads/branch-2.8
Commit: 89d59c292fb149d245c18d1f188cc113e849c087
Parents: 0cba282
Author: Kihwal Lee <ki...@apache.org>
Authored: Thu Jun 1 12:25:27 2017 -0500
Committer: Kihwal Lee <ki...@apache.org>
Committed: Thu Jun 1 12:25:27 2017 -0500
----------------------------------------------------------------------
.../main/java/org/apache/hadoop/io/IOUtils.java | 55 +++++++++++++++++++-
.../hdfs/server/datanode/BlockReceiver.java | 9 +++-
.../server/datanode/fsdataset/FsDatasetSpi.java | 4 +-
.../datanode/fsdataset/impl/FsDatasetImpl.java | 37 ++++++++++---
.../server/datanode/SimulatedFSDataset.java | 3 +-
.../datanode/TestDataNodeHotSwapVolumes.java | 6 ++-
.../server/datanode/TestSimulatedFSDataset.java | 4 +-
.../extdataset/ExternalDatasetImpl.java | 3 +-
.../fsdataset/impl/TestFsDatasetImpl.java | 2 +-
9 files changed, 104 insertions(+), 19 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/89d59c29/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/IOUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/IOUtils.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/IOUtils.java
index 2807307..7268469 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/IOUtils.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/IOUtils.java
@@ -27,6 +27,7 @@ import java.nio.file.DirectoryStream;
import java.nio.file.DirectoryIteratorException;
import java.nio.file.Files;
import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
import java.util.ArrayList;
import java.util.List;
@@ -35,7 +36,7 @@ import org.apache.commons.logging.Log;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.util.ChunkedArrayList;
+import org.apache.hadoop.util.Shell;
import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.IO_FILE_BUFFER_SIZE_DEFAULT;
import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.IO_FILE_BUFFER_SIZE_KEY;
@@ -352,4 +353,56 @@ public class IOUtils {
}
return list;
}
+
+ /**
+ * Ensure that any writes to the given file is written to the storage device
+ * that contains it. This method opens channel on given File and closes it
+ * once the sync is done.<br>
+ * Borrowed from Uwe Schindler in LUCENE-5588
+ * @param fileToSync the file to fsync
+ */
+ public static void fsync(File fileToSync) throws IOException {
+ if (!fileToSync.exists()) {
+ throw new FileNotFoundException(
+ "File/Directory " + fileToSync.getAbsolutePath() + " does not exist");
+ }
+ boolean isDir = fileToSync.isDirectory();
+ // If the file is a directory we have to open read-only, for regular files
+ // we must open r/w for the fsync to have an effect. See
+ // http://blog.httrack.com/blog/2013/11/15/
+ // everything-you-always-wanted-to-know-about-fsync/
+ try(FileChannel channel = FileChannel.open(fileToSync.toPath(),
+ isDir ? StandardOpenOption.READ : StandardOpenOption.WRITE)){
+ fsync(channel, isDir);
+ }
+ }
+
+ /**
+ * Ensure that any writes to the given file is written to the storage device
+ * that contains it. This method opens channel on given File and closes it
+ * once the sync is done.
+ * Borrowed from Uwe Schindler in LUCENE-5588
+ * @param channel Channel to sync
+ * @param isDir if true, the given file is a directory (Channel should be
+ * opened for read and ignore IOExceptions, because not all file
+ * systems and operating systems allow to fsync on a directory)
+ * @throws IOException
+ */
+ public static void fsync(FileChannel channel, boolean isDir)
+ throws IOException {
+ try {
+ channel.force(true);
+ } catch (IOException ioe) {
+ if (isDir) {
+ assert !(Shell.LINUX
+ || Shell.MAC) : "On Linux and MacOSX fsyncing a directory"
+ + " should not throw IOException, we just don't want to rely"
+ + " on that in production (undocumented)" + ". Got: " + ioe;
+ // Ignore exception if it is a directory
+ return;
+ }
+ // Throw original exception
+ throw ioe;
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/89d59c29/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BlockReceiver.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BlockReceiver.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BlockReceiver.java
index 6dd2f3c..512dbe2 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BlockReceiver.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BlockReceiver.java
@@ -125,6 +125,7 @@ class BlockReceiver implements Closeable {
private final boolean isTransfer;
private boolean syncOnClose;
+ private volatile boolean dirSyncOnFinalize;
private long restartBudget;
/** the reference of the volume where the block receiver writes to */
private ReplicaHandler replicaHandler;
@@ -547,6 +548,9 @@ class BlockReceiver implements Closeable {
// avoid double sync'ing on close
if (syncBlock && lastPacketInBlock) {
this.syncOnClose = false;
+ // sync directory for finalize irrespective of syncOnClose config since
+ // sync is requested.
+ this.dirSyncOnFinalize = true;
}
// update received bytes
@@ -900,6 +904,7 @@ class BlockReceiver implements Closeable {
boolean isReplaceBlock) throws IOException {
syncOnClose = datanode.getDnConf().syncOnClose;
+ dirSyncOnFinalize = syncOnClose;
boolean responderClosed = false;
mirrorOut = mirrOut;
mirrorAddr = mirrAddr;
@@ -941,7 +946,7 @@ class BlockReceiver implements Closeable {
} else {
// for isDatnode or TRANSFER_FINALIZED
// Finalize the block.
- datanode.data.finalizeBlock(block);
+ datanode.data.finalizeBlock(block, dirSyncOnFinalize);
}
}
datanode.metrics.incrBlocksWritten();
@@ -1433,7 +1438,7 @@ class BlockReceiver implements Closeable {
BlockReceiver.this.close();
endTime = ClientTraceLog.isInfoEnabled() ? System.nanoTime() : 0;
block.setNumBytes(replicaInfo.getNumBytes());
- datanode.data.finalizeBlock(block);
+ datanode.data.finalizeBlock(block, dirSyncOnFinalize);
}
if (pinning) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/89d59c29/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/FsDatasetSpi.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/FsDatasetSpi.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/FsDatasetSpi.java
index c7ccd4c..d055984 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/FsDatasetSpi.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/FsDatasetSpi.java
@@ -390,12 +390,14 @@ public interface FsDatasetSpi<V extends FsVolumeSpi> extends FSDatasetMBean {
* Finalizes the block previously opened for writing using writeToBlock.
* The block size is what is in the parameter b and it must match the amount
* of data written
+ * @param block Block to be finalized
+ * @param fsyncDir whether to sync the directory changes to durable device.
* @throws IOException
* @throws ReplicaNotFoundException if the replica can not be found when the
* block is been finalized. For instance, the block resides on an HDFS volume
* that has been removed.
*/
- void finalizeBlock(ExtendedBlock b) throws IOException;
+ void finalizeBlock(ExtendedBlock b, boolean fsyncDir) throws IOException;
/**
* Unfinalizes the block previously opened for writing using writeToBlock.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/89d59c29/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsDatasetImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsDatasetImpl.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsDatasetImpl.java
index 483efd7..bdabb45 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsDatasetImpl.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsDatasetImpl.java
@@ -903,6 +903,17 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
return dstfile;
}
+ private void fsyncDirectory(FsVolumeSpi volume, File... dirs)
+ throws IOException {
+ for (File dir : dirs) {
+ try {
+ IOUtils.fsync(dir);
+ } catch (IOException e) {
+ throw new IOException("Failed to sync " + dir, e);
+ }
+ }
+ }
+
/**
* Copy the block and meta files for the given block to the given destination.
* @return the new meta and block files.
@@ -997,7 +1008,8 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
targetVolume, blockFiles[0].getParentFile(), 0);
newReplicaInfo.setNumBytes(blockFiles[1].length());
// Finalize the copied files
- newReplicaInfo = finalizeReplica(block.getBlockPoolId(), newReplicaInfo);
+ newReplicaInfo = finalizeReplica(block.getBlockPoolId(), newReplicaInfo,
+ false);
try(AutoCloseableLock lock = datasetLock.acquire()) {
// Increment numBlocks here as this block moved without knowing to BPS
FsVolumeImpl volume = (FsVolumeImpl) newReplicaInfo.getVolume();
@@ -1358,7 +1370,7 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
bumpReplicaGS(replicaInfo, newGS);
// finalize the replica if RBW
if (replicaInfo.getState() == ReplicaState.RBW) {
- finalizeReplica(b.getBlockPoolId(), replicaInfo);
+ finalizeReplica(b.getBlockPoolId(), replicaInfo, false);
}
return replicaInfo;
}
@@ -1707,7 +1719,8 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
* Complete the block write!
*/
@Override // FsDatasetSpi
- public void finalizeBlock(ExtendedBlock b) throws IOException {
+ public void finalizeBlock(ExtendedBlock b, boolean fsyncDir)
+ throws IOException {
try(AutoCloseableLock lock = datasetLock.acquire()) {
if (Thread.interrupted()) {
// Don't allow data modifications from interrupted threads
@@ -1719,12 +1732,12 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
// been opened for append but never modified
return;
}
- finalizeReplica(b.getBlockPoolId(), replicaInfo);
+ finalizeReplica(b.getBlockPoolId(), replicaInfo, fsyncDir);
}
}
private FinalizedReplica finalizeReplica(String bpid,
- ReplicaInfo replicaInfo) throws IOException {
+ ReplicaInfo replicaInfo, boolean fsyncDir) throws IOException {
try(AutoCloseableLock lock = datasetLock.acquire()) {
FinalizedReplica newReplicaInfo = null;
if (replicaInfo.getState() == ReplicaState.RUR &&
@@ -1744,7 +1757,15 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
bpid, replicaInfo, f, replicaInfo.getBytesReserved());
newReplicaInfo =
new FinalizedReplica(replicaInfo, v, dest.getParentFile());
-
+ /*
+ * Sync the directory after rename from tmp/rbw to Finalized if
+ * configured. Though rename should be atomic operation, sync on both
+ * dest and src directories are done because IOUtils.fsync() calls
+ * directory's channel sync, not the journal itself.
+ */
+ if (fsyncDir) {
+ fsyncDirectory(v, dest.getParentFile(), f.getParentFile());
+ }
if (v.isTransientStorage()) {
releaseLockedMemory(
replicaInfo.getOriginalBytesReserved()
@@ -2718,12 +2739,12 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
// but it is immediately converted to finalized state within the same
// lock, so no need to update it.
volumeMap.add(bpid, newReplicaInfo);
- finalizeReplica(bpid, newReplicaInfo);
+ finalizeReplica(bpid, newReplicaInfo, false);
}
}
// finalize the block
- return finalizeReplica(bpid, rur);
+ return finalizeReplica(bpid, rur, false);
}
private File[] copyReplicaWithNewBlockIdAndGS(
http://git-wip-us.apache.org/repos/asf/hadoop/blob/89d59c29/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/SimulatedFSDataset.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/SimulatedFSDataset.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/SimulatedFSDataset.java
index 7bd799d..544c9ca 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/SimulatedFSDataset.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/SimulatedFSDataset.java
@@ -598,7 +598,8 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
}
@Override // FsDatasetSpi
- public synchronized void finalizeBlock(ExtendedBlock b) throws IOException {
+ public synchronized void finalizeBlock(ExtendedBlock b, boolean fsyncDir)
+ throws IOException {
final Map<Block, BInfo> map = getMap(b.getBlockPoolId());
BInfo binfo = map.get(b.getLocalBlock());
if (binfo == null) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/89d59c29/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeHotSwapVolumes.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeHotSwapVolumes.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeHotSwapVolumes.java
index e36b744c..08df71f 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeHotSwapVolumes.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeHotSwapVolumes.java
@@ -666,10 +666,12 @@ public class TestDataNodeHotSwapVolumes {
// Bypass the argument to FsDatasetImpl#finalizeBlock to verify that
// the block is not removed, since the volume reference should not
// be released at this point.
- data.finalizeBlock((ExtendedBlock) invocation.getArguments()[0]);
+ data.finalizeBlock((ExtendedBlock) invocation.getArguments()[0],
+ (boolean) invocation.getArguments()[1]);
return null;
}
- }).when(dn.data).finalizeBlock(any(ExtendedBlock.class));
+ }).when(dn.data).finalizeBlock(any(ExtendedBlock.class),
+ Mockito.anyBoolean());
final CyclicBarrier barrier = new CyclicBarrier(2);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/89d59c29/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestSimulatedFSDataset.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestSimulatedFSDataset.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestSimulatedFSDataset.java
index 283ad8b..9a77350 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestSimulatedFSDataset.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestSimulatedFSDataset.java
@@ -96,7 +96,7 @@ public class TestSimulatedFSDataset {
out.close();
}
b.setNumBytes(blockIdToLen(i));
- fsdataset.finalizeBlock(b);
+ fsdataset.finalizeBlock(b, false);
assertEquals(blockIdToLen(i), fsdataset.getLength(b));
}
return bytesAdded;
@@ -295,7 +295,7 @@ public class TestSimulatedFSDataset {
}
try {
- fsdataset.finalizeBlock(b);
+ fsdataset.finalizeBlock(b, false);
assertTrue("Expected an IO exception", false);
} catch (IOException e) {
// ok - as expected
http://git-wip-us.apache.org/repos/asf/hadoop/blob/89d59c29/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/extdataset/ExternalDatasetImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/extdataset/ExternalDatasetImpl.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/extdataset/ExternalDatasetImpl.java
index 0be6168..70ddcab 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/extdataset/ExternalDatasetImpl.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/extdataset/ExternalDatasetImpl.java
@@ -179,7 +179,8 @@ public class ExternalDatasetImpl implements FsDatasetSpi<ExternalVolumeImpl> {
}
@Override
- public void finalizeBlock(ExtendedBlock b) throws IOException {
+ public void finalizeBlock(ExtendedBlock b, boolean fsyncDir)
+ throws IOException {
}
@Override
http://git-wip-us.apache.org/repos/asf/hadoop/blob/89d59c29/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/TestFsDatasetImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/TestFsDatasetImpl.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/TestFsDatasetImpl.java
index fd606e0..9b80430 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/TestFsDatasetImpl.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/TestFsDatasetImpl.java
@@ -529,7 +529,7 @@ public class TestFsDatasetImpl {
// Lets wait for the other thread finish getting block report
blockReportReceivedLatch.await();
- dataset.finalizeBlock(eb);
+ dataset.finalizeBlock(eb, false);
LOG.info("FinalizeBlock finished");
} catch (Exception e) {
LOG.warn("Exception caught. This should not affect the test", e);
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org