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 ar...@apache.org on 2017/01/17 06:14:25 UTC
[1/5] hadoop git commit: HDFS-10959. Update DataNode to use
DatasetVolumeChecker. Contributed by Xiaoyu Yao.
Repository: hadoop
Updated Branches:
refs/heads/branch-2 2d6be7ea2 -> 8ec9dca2e
HDFS-10959. Update DataNode to use DatasetVolumeChecker. Contributed by Xiaoyu Yao.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/413dccaf
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/413dccaf
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/413dccaf
Branch: refs/heads/branch-2
Commit: 413dccaf784c0cc13cd323bc49ecbaee27023c2e
Parents: 2d6be7e
Author: Arpit Agarwal <ar...@apache.org>
Authored: Mon Jan 16 14:31:39 2017 -0800
Committer: Arpit Agarwal <ar...@apache.org>
Committed: Mon Jan 16 22:13:58 2017 -0800
----------------------------------------------------------------------
.../server/datanode/ProfilingFileIoEvents.java | 110 +++++++
.../fsdataset/DataNodeVolumeMetrics.java | 289 +++++++++++++++++++
.../server/datanode/fsdataset/FsVolumeSpi.java | 2 +
.../datanode/fsdataset/impl/FsVolumeImpl.java | 12 +
.../hdfs/server/datanode/DataNodeTestUtils.java | 39 +++
.../server/datanode/SimulatedFSDataset.java | 19 +-
.../TestDataNodeVolumeFailureReporting.java | 47 +--
.../datanode/TestDataNodeVolumeMetrics.java | 182 ++++++++++++
.../server/datanode/TestDirectoryScanner.java | 8 +
.../datanode/extdataset/ExternalVolumeImpl.java | 6 +
10 files changed, 672 insertions(+), 42 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/413dccaf/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/ProfilingFileIoEvents.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/ProfilingFileIoEvents.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/ProfilingFileIoEvents.java
new file mode 100644
index 0000000..5835fe8
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/ProfilingFileIoEvents.java
@@ -0,0 +1,110 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hdfs.server.datanode;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.hdfs.server.datanode.fsdataset.DataNodeVolumeMetrics;
+import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi;
+import org.apache.hadoop.util.Time;
+
+import javax.annotation.Nullable;
+
+/**
+ * {@link FileIoEvents} that profiles the performance of the metadata and data
+ * related operations on datanode volumes.
+ */
+@InterfaceAudience.Private
+class ProfilingFileIoEvents implements FileIoEvents {
+
+ @Override
+ public long beforeMetadataOp(@Nullable FsVolumeSpi volume,
+ FileIoProvider.OPERATION op) {
+ DataNodeVolumeMetrics metrics = getVolumeMetrics(volume);
+ if (metrics != null) {
+ return Time.monotonicNow();
+ }
+ return 0;
+ }
+
+ @Override
+ public void afterMetadataOp(@Nullable FsVolumeSpi volume,
+ FileIoProvider.OPERATION op, long begin) {
+ DataNodeVolumeMetrics metrics = getVolumeMetrics(volume);
+ if (metrics != null) {
+ metrics.addMetadastaOperationLatency(Time.monotonicNow() - begin);
+ }
+ }
+
+ @Override
+ public long beforeFileIo(@Nullable FsVolumeSpi volume,
+ FileIoProvider.OPERATION op, long len) {
+ DataNodeVolumeMetrics metrics = getVolumeMetrics(volume);
+ if (metrics != null) {
+ return Time.monotonicNow();
+ }
+ return 0;
+ }
+
+ @Override
+ public void afterFileIo(@Nullable FsVolumeSpi volume,
+ FileIoProvider.OPERATION op, long begin, long len) {
+ DataNodeVolumeMetrics metrics = getVolumeMetrics(volume);
+ if (metrics != null) {
+ long latency = Time.monotonicNow() - begin;
+ metrics.addDataFileIoLatency(latency);
+ switch (op) {
+ case SYNC:
+ metrics.addSyncIoLatency(latency);
+ break;
+ case FLUSH:
+ metrics.addFlushIoLatency(latency);
+ break;
+ case READ:
+ metrics.addReadIoLatency(latency);
+ break;
+ case WRITE:
+ metrics.addWriteIoLatency(latency);
+ break;
+ default:
+ }
+ }
+ }
+
+ @Override
+ public void onFailure(@Nullable FsVolumeSpi volume,
+ FileIoProvider.OPERATION op, Exception e, long begin) {
+ DataNodeVolumeMetrics metrics = getVolumeMetrics(volume);
+ if (metrics != null) {
+ metrics.addFileIoError(Time.monotonicNow() - begin);
+ }
+ }
+
+ @Nullable
+ @Override
+ public String getStatistics() {
+ return null;
+ }
+
+ private DataNodeVolumeMetrics getVolumeMetrics(final FsVolumeSpi volume) {
+ if (volume != null) {
+ return volume.getMetrics();
+ }
+ return null;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/413dccaf/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/DataNodeVolumeMetrics.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/DataNodeVolumeMetrics.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/DataNodeVolumeMetrics.java
new file mode 100644
index 0000000..e4d8707
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/DataNodeVolumeMetrics.java
@@ -0,0 +1,289 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.datanode.fsdataset;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.DFSConfigKeys;
+import org.apache.hadoop.metrics2.MetricsSystem;
+import org.apache.hadoop.metrics2.annotation.Metric;
+import org.apache.hadoop.metrics2.annotation.Metrics;
+import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
+import org.apache.hadoop.metrics2.lib.MetricsRegistry;
+import org.apache.hadoop.metrics2.lib.MutableCounterLong;
+import org.apache.hadoop.metrics2.lib.MutableQuantiles;
+import org.apache.hadoop.metrics2.lib.MutableRate;
+
+import java.util.concurrent.ThreadLocalRandom;
+
+/**
+ * This class is for maintaining Datanode Volume IO related statistics and
+ * publishing them through the metrics interfaces.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+@Metrics(name = "DataNodeVolume", about = "DataNode Volume metrics",
+ context = "dfs")
+public class DataNodeVolumeMetrics {
+ private final MetricsRegistry registry = new MetricsRegistry("FsVolume");
+
+ @Metric("number of metadata operations")
+ private MutableCounterLong totalMetadataOperations;
+ @Metric("metadata operation rate")
+ private MutableRate metadataOperationRate;
+ private MutableQuantiles[] metadataOperationLatencyQuantiles;
+
+ @Metric("number of data file io operations")
+ private MutableCounterLong totalDataFileIos;
+ @Metric("data file io operation rate")
+ private MutableRate dataFileIoRate;
+ private MutableQuantiles[] dataFileIoLatencyQuantiles;
+
+ @Metric("file io flush rate")
+ private MutableRate flushIoRate;
+ private MutableQuantiles[] flushIoLatencyQuantiles;
+
+ @Metric("file io sync rate")
+ private MutableRate syncIoRate;
+ private MutableQuantiles[] syncIoLatencyQuantiles;
+
+ @Metric("file io read rate")
+ private MutableRate readIoRate;
+ private MutableQuantiles[] readIoLatencyQuantiles;
+
+ @Metric("file io write rate")
+ private MutableRate writeIoRate;
+ private MutableQuantiles[] writeIoLatencyQuantiles;
+
+ @Metric("number of file io errors")
+ private MutableCounterLong totalFileIoErrors;
+ @Metric("file io error rate")
+ private MutableRate fileIoErrorRate;
+
+ public long getTotalMetadataOperations() {
+ return totalMetadataOperations.value();
+ }
+
+ // Based on metadataOperationRate
+ public long getMetadataOperationSampleCount() {
+ return metadataOperationRate.lastStat().numSamples();
+ }
+
+ public double getMetadataOperationMean() {
+ return metadataOperationRate.lastStat().mean();
+ }
+
+ public double getMetadataOperationStdDev() {
+ return metadataOperationRate.lastStat().stddev();
+ }
+
+ public long getTotalDataFileIos() {
+ return totalDataFileIos.value();
+ }
+
+ // Based on dataFileIoRate
+ public long getDataFileIoSampleCount() {
+ return dataFileIoRate.lastStat().numSamples();
+ }
+
+ public double getDataFileIoMean() {
+ return dataFileIoRate.lastStat().mean();
+ }
+
+ public double getDataFileIoStdDev() {
+ return dataFileIoRate.lastStat().stddev();
+ }
+
+ // Based on flushIoRate
+ public long getFlushIoSampleCount() {
+ return flushIoRate.lastStat().numSamples();
+ }
+
+ public double getFlushIoMean() {
+ return flushIoRate.lastStat().mean();
+ }
+
+ public double getFlushIoStdDev() {
+ return flushIoRate.lastStat().stddev();
+ }
+
+ // Based on syncIoRate
+ public long getSyncIoSampleCount() {
+ return syncIoRate.lastStat().numSamples();
+ }
+
+ public double getSyncIoMean() {
+ return syncIoRate.lastStat().mean();
+ }
+
+ public double getSyncIoStdDev() {
+ return syncIoRate.lastStat().stddev();
+ }
+
+ // Based on readIoRate
+ public long getReadIoSampleCount() {
+ return readIoRate.lastStat().numSamples();
+ }
+
+ public double getReadIoMean() {
+ return readIoRate.lastStat().mean();
+ }
+
+ public double getReadIoStdDev() {
+ return readIoRate.lastStat().stddev();
+ }
+
+ // Based on writeIoRate
+ public long getWriteIoSampleCount() {
+ return syncIoRate.lastStat().numSamples();
+ }
+
+ public double getWriteIoMean() {
+ return syncIoRate.lastStat().mean();
+ }
+
+ public double getWriteIoStdDev() {
+ return syncIoRate.lastStat().stddev();
+ }
+
+ public long getTotalFileIoErrors() {
+ return totalFileIoErrors.value();
+ }
+
+ // Based on fileIoErrorRate
+ public long getFileIoErrorSampleCount() {
+ return fileIoErrorRate.lastStat().numSamples();
+ }
+
+ public double getFileIoErrorMean() {
+ return fileIoErrorRate.lastStat().mean();
+ }
+
+ public double getFileIoErrorStdDev() {
+ return fileIoErrorRate.lastStat().stddev();
+ }
+
+ private final String name;
+ private final MetricsSystem ms;
+
+ public DataNodeVolumeMetrics(final MetricsSystem metricsSystem,
+ final String volumeName, final int[] intervals) {
+ this.ms = metricsSystem;
+ this.name = volumeName;
+ final int len = intervals.length;
+ metadataOperationLatencyQuantiles = new MutableQuantiles[len];
+ dataFileIoLatencyQuantiles = new MutableQuantiles[len];
+ flushIoLatencyQuantiles = new MutableQuantiles[len];
+ syncIoLatencyQuantiles = new MutableQuantiles[len];
+ readIoLatencyQuantiles = new MutableQuantiles[len];
+ writeIoLatencyQuantiles = new MutableQuantiles[len];
+ for (int i = 0; i < len; i++) {
+ int interval = intervals[i];
+ metadataOperationLatencyQuantiles[i] = registry.newQuantiles(
+ "metadataOperationLatency" + interval + "s",
+ "Meatadata Operation Latency in ms", "ops", "latency", interval);
+ dataFileIoLatencyQuantiles[i] = registry.newQuantiles(
+ "dataFileIoLatency" + interval + "s",
+ "Data File Io Latency in ms", "ops", "latency", interval);
+ flushIoLatencyQuantiles[i] = registry.newQuantiles(
+ "flushIoLatency" + interval + "s",
+ "Data flush Io Latency in ms", "ops", "latency", interval);
+ syncIoLatencyQuantiles[i] = registry.newQuantiles(
+ "syncIoLatency" + interval + "s",
+ "Data sync Io Latency in ms", "ops", "latency", interval);
+ readIoLatencyQuantiles[i] = registry.newQuantiles(
+ "readIoLatency" + interval + "s",
+ "Data read Io Latency in ms", "ops", "latency", interval);
+ writeIoLatencyQuantiles[i] = registry.newQuantiles(
+ "writeIoLatency" + interval + "s",
+ "Data write Io Latency in ms", "ops", "latency", interval);
+ }
+ }
+
+ public static DataNodeVolumeMetrics create(final Configuration conf,
+ final String volumeName) {
+ MetricsSystem ms = DefaultMetricsSystem.instance();
+ String name = "DataNodeVolume-"+ (volumeName.isEmpty()
+ ? "UndefinedDataNodeVolume"+ ThreadLocalRandom.current().nextInt()
+ : volumeName.replace(':', '-'));
+
+ // Percentile measurement is off by default, by watching no intervals
+ int[] intervals =
+ conf.getInts(DFSConfigKeys.DFS_METRICS_PERCENTILES_INTERVALS_KEY);
+ return ms.register(name, null, new DataNodeVolumeMetrics(ms, name,
+ intervals));
+ }
+
+ public String name() {
+ return name;
+ }
+
+ public void unRegister() {
+ ms.unregisterSource(name);
+ }
+
+ public void addMetadastaOperationLatency(final long latency) {
+ totalMetadataOperations.incr();
+ metadataOperationRate.add(latency);
+ for (MutableQuantiles q : metadataOperationLatencyQuantiles) {
+ q.add(latency);
+ }
+ }
+
+ public void addDataFileIoLatency(final long latency) {
+ totalDataFileIos.incr();
+ dataFileIoRate.add(latency);
+ for (MutableQuantiles q : dataFileIoLatencyQuantiles) {
+ q.add(latency);
+ }
+ }
+
+ public void addSyncIoLatency(final long latency) {
+ syncIoRate.add(latency);
+ for (MutableQuantiles q : syncIoLatencyQuantiles) {
+ q.add(latency);
+ }
+ }
+
+ public void addFlushIoLatency(final long latency) {
+ flushIoRate.add(latency);
+ for (MutableQuantiles q : flushIoLatencyQuantiles) {
+ q.add(latency);
+ }
+ }
+
+ public void addReadIoLatency(final long latency) {
+ readIoRate.add(latency);
+ for (MutableQuantiles q : readIoLatencyQuantiles) {
+ q.add(latency);
+ }
+ }
+
+ public void addWriteIoLatency(final long latency) {
+ writeIoRate.add(latency);
+ for (MutableQuantiles q: writeIoLatencyQuantiles) {
+ q.add(latency);
+ }
+ }
+
+ public void addFileIoError(final long latency) {
+ totalFileIoErrors.incr();
+ metadataOperationRate.add(latency);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/413dccaf/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/FsVolumeSpi.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/FsVolumeSpi.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/FsVolumeSpi.java
index 607aefd..fbef439 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/FsVolumeSpi.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/FsVolumeSpi.java
@@ -210,4 +210,6 @@ public interface FsVolumeSpi
}
FileIoProvider getFileIoProvider();
+
+ DataNodeVolumeMetrics getMetrics();
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/413dccaf/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeImpl.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeImpl.java
index 9955bf4..bc822c6 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeImpl.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeImpl.java
@@ -47,6 +47,7 @@ import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdfs.server.datanode.FileIoProvider;
import org.apache.hadoop.hdfs.server.datanode.BlockMetadataHeader;
import org.apache.hadoop.hdfs.server.datanode.checker.VolumeCheckResult;
+import org.apache.hadoop.hdfs.server.datanode.fsdataset.DataNodeVolumeMetrics;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsDatasetSpi;
import org.apache.hadoop.util.AutoCloseableLock;
import org.apache.hadoop.hdfs.DFSConfigKeys;
@@ -110,6 +111,7 @@ public class FsVolumeImpl implements FsVolumeSpi {
// query from the filesystem.
protected volatile long configuredCapacity;
private final FileIoProvider fileIoProvider;
+ private final DataNodeVolumeMetrics metrics;
/**
* Per-volume worker pool that processes new blocks to cache.
@@ -137,6 +139,7 @@ public class FsVolumeImpl implements FsVolumeSpi {
this.fileIoProvider = dataset.datanode != null ?
dataset.datanode.getFileIoProvider() : new FileIoProvider(conf);
cacheExecutor = initializeCacheExecutor(parent);
+ this.metrics = DataNodeVolumeMetrics.create(conf, parent.getAbsolutePath());
}
protected ThreadPoolExecutor initializeCacheExecutor(File parent) {
@@ -950,6 +953,9 @@ public class FsVolumeImpl implements FsVolumeSpi {
for (Entry<String, BlockPoolSlice> entry : set) {
entry.getValue().shutdown(null);
}
+ if (metrics != null) {
+ metrics.unRegister();
+ }
}
void addBlockPool(String bpid, Configuration conf) throws IOException {
@@ -1105,4 +1111,10 @@ public class FsVolumeImpl implements FsVolumeSpi {
public FileIoProvider getFileIoProvider() {
return fileIoProvider;
}
+
+
+ @Override
+ public DataNodeVolumeMetrics getMetrics() {
+ return metrics;
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/413dccaf/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/DataNodeTestUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/DataNodeTestUtils.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/DataNodeTestUtils.java
index e2755f9..42bcbf3 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/DataNodeTestUtils.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/DataNodeTestUtils.java
@@ -22,7 +22,11 @@ package org.apache.hadoop.hdfs.server.datanode;
import java.io.File;
import java.io.IOException;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.conf.ReconfigurationException;
+import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.MiniDFSCluster;
import org.apache.hadoop.hdfs.protocol.DatanodeID;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsDatasetSpi;
@@ -30,12 +34,17 @@ import org.apache.hadoop.hdfs.server.datanode.fsdataset.impl.FsDatasetTestUtil;
import org.apache.hadoop.hdfs.server.protocol.DatanodeRegistration;
import org.apache.hadoop.hdfs.server.protocol.InterDatanodeProtocol;
+import static org.hamcrest.core.Is.is;
+import static org.junit.Assert.assertThat;
+
/**
* Utility class for accessing package-private DataNode information during tests.
* Must not contain usage of classes that are not explicitly listed as
* dependencies to {@link MiniDFSCluster}.
*/
public class DataNodeTestUtils {
+ private static final Log LOG =
+ LogFactory.getLog(DataNodeTestUtils.class);
private static final String DIR_FAILURE_SUFFIX = ".origin";
public final static String TEST_CLUSTER_ID = "testClusterID";
@@ -175,4 +184,34 @@ public class DataNodeTestUtils {
dn.getDirectoryScanner().reconcile();
}
}
+
+ /**
+ * Reconfigure a DataNode by setting a new list of volumes.
+ *
+ * @param dn DataNode to reconfigure
+ * @param newVols new volumes to configure
+ * @throws Exception if there is any failure
+ */
+ public static void reconfigureDataNode(DataNode dn, File... newVols)
+ throws Exception {
+ StringBuilder dnNewDataDirs = new StringBuilder();
+ for (File newVol: newVols) {
+ if (dnNewDataDirs.length() > 0) {
+ dnNewDataDirs.append(',');
+ }
+ dnNewDataDirs.append(newVol.getAbsolutePath());
+ }
+ try {
+ assertThat(
+ dn.reconfigurePropertyImpl(
+ DFSConfigKeys.DFS_DATANODE_DATA_DIR_KEY,
+ dnNewDataDirs.toString()),
+ is(dn.getConf().get(DFSConfigKeys.DFS_DATANODE_DATA_DIR_KEY)));
+ } catch (ReconfigurationException e) {
+ // This can be thrown if reconfiguration tries to use a failed volume.
+ // We need to swallow the exception, because some of our tests want to
+ // cover this case.
+ LOG.warn("Could not reconfigure DataNode.", e);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/413dccaf/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 c10d84b..dff8482 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
@@ -45,6 +45,7 @@ import org.apache.hadoop.hdfs.protocol.BlockLocalPathInfo;
import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
import org.apache.hadoop.hdfs.protocol.HdfsBlocksMetadata;
import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.ReplicaState;
+import org.apache.hadoop.hdfs.server.datanode.fsdataset.DataNodeVolumeMetrics;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsDatasetSpi;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeReference;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi;
@@ -449,11 +450,14 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
static class SimulatedVolume implements FsVolumeSpi {
private final SimulatedStorage storage;
private final FileIoProvider fileIoProvider;
+ private final DataNodeVolumeMetrics metrics;
SimulatedVolume(final SimulatedStorage storage,
- final FileIoProvider fileIoProvider) {
+ final FileIoProvider fileIoProvider,
+ final DataNodeVolumeMetrics metrics) {
this.storage = storage;
this.fileIoProvider = fileIoProvider;
+ this.metrics = metrics;
}
@Override
@@ -541,6 +545,11 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
}
@Override
+ public DataNodeVolumeMetrics getMetrics() {
+ return metrics;
+ }
+
+ @Override
public VolumeCheckResult check(VolumeCheckContext context)
throws Exception {
return VolumeCheckResult.HEALTHY;
@@ -575,7 +584,12 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
this.storage = new SimulatedStorage(
conf.getLong(CONFIG_PROPERTY_CAPACITY, DEFAULT_CAPACITY),
conf.getEnum(CONFIG_PROPERTY_STATE, DEFAULT_STATE));
- this.volume = new SimulatedVolume(this.storage, this.fileIoProvider);
+
+ // TODO: per volume id or path
+ DataNodeVolumeMetrics volumeMetrics = DataNodeVolumeMetrics.create(conf,
+ datanodeUuid);
+ this.volume = new SimulatedVolume(this.storage, this.fileIoProvider,
+ volumeMetrics);
this.datasetLock = new AutoCloseableLock();
}
@@ -1393,4 +1407,3 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
return datasetLock.acquire();
}
}
-
http://git-wip-us.apache.org/repos/asf/hadoop/blob/413dccaf/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailureReporting.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailureReporting.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailureReporting.java
index 2a2fc4a..e6f1366 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailureReporting.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailureReporting.java
@@ -24,7 +24,6 @@ import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assume.assumeTrue;
@@ -34,7 +33,6 @@ import java.util.ArrayList;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.conf.ReconfigurationException;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.DFSConfigKeys;
@@ -59,7 +57,8 @@ import org.junit.Test;
*/
public class TestDataNodeVolumeFailureReporting {
- private static final Log LOG = LogFactory.getLog(TestDataNodeVolumeFailureReporting.class);
+ private static final Log LOG =
+ LogFactory.getLog(TestDataNodeVolumeFailureReporting.class);
{
GenericTestUtils.setLogLevel(TestDataNodeVolumeFailureReporting.LOG,
Level.ALL);
@@ -387,8 +386,8 @@ public class TestDataNodeVolumeFailureReporting {
checkFailuresAtNameNode(dm, dns.get(1), true, dn2Vol1.getAbsolutePath());
// Reconfigure again to try to add back the failed volumes.
- reconfigureDataNode(dns.get(0), dn1Vol1, dn1Vol2);
- reconfigureDataNode(dns.get(1), dn2Vol1, dn2Vol2);
+ DataNodeTestUtils.reconfigureDataNode(dns.get(0), dn1Vol1, dn1Vol2);
+ DataNodeTestUtils.reconfigureDataNode(dns.get(1), dn2Vol1, dn2Vol2);
DataNodeTestUtils.triggerHeartbeat(dns.get(0));
DataNodeTestUtils.triggerHeartbeat(dns.get(1));
@@ -408,8 +407,8 @@ public class TestDataNodeVolumeFailureReporting {
// Reconfigure a third time with the failed volumes. Afterwards, we expect
// the same volume failures to be reported. (No double-counting.)
- reconfigureDataNode(dns.get(0), dn1Vol1, dn1Vol2);
- reconfigureDataNode(dns.get(1), dn2Vol1, dn2Vol2);
+ DataNodeTestUtils.reconfigureDataNode(dns.get(0), dn1Vol1, dn1Vol2);
+ DataNodeTestUtils.reconfigureDataNode(dns.get(1), dn2Vol1, dn2Vol2);
DataNodeTestUtils.triggerHeartbeat(dns.get(0));
DataNodeTestUtils.triggerHeartbeat(dns.get(1));
@@ -430,8 +429,8 @@ public class TestDataNodeVolumeFailureReporting {
// Replace failed volume with healthy volume and run reconfigure DataNode.
// The failed volume information should be cleared.
DataNodeTestUtils.restoreDataDirFromFailure(dn1Vol1, dn2Vol1);
- reconfigureDataNode(dns.get(0), dn1Vol1, dn1Vol2);
- reconfigureDataNode(dns.get(1), dn2Vol1, dn2Vol2);
+ DataNodeTestUtils.reconfigureDataNode(dns.get(0), dn1Vol1, dn1Vol2);
+ DataNodeTestUtils.reconfigureDataNode(dns.get(1), dn2Vol1, dn2Vol2);
DataNodeTestUtils.triggerHeartbeat(dns.get(0));
DataNodeTestUtils.triggerHeartbeat(dns.get(1));
@@ -631,34 +630,4 @@ public class TestDataNodeVolumeFailureReporting {
cluster.getNamesystem().getBlockManager().getDatanodeManager(), 0);
volumeCapacity = dnCapacity / cluster.getStoragesPerDatanode();
}
-
- /**
- * Reconfigure a DataNode by setting a new list of volumes.
- *
- * @param dn DataNode to reconfigure
- * @param newVols new volumes to configure
- * @throws Exception if there is any failure
- */
- private static void reconfigureDataNode(DataNode dn, File... newVols)
- throws Exception {
- StringBuilder dnNewDataDirs = new StringBuilder();
- for (File newVol: newVols) {
- if (dnNewDataDirs.length() > 0) {
- dnNewDataDirs.append(',');
- }
- dnNewDataDirs.append(newVol.getAbsolutePath());
- }
- try {
- assertThat(
- dn.reconfigurePropertyImpl(
- DFSConfigKeys.DFS_DATANODE_DATA_DIR_KEY,
- dnNewDataDirs.toString()),
- is(dn.getConf().get(DFSConfigKeys.DFS_DATANODE_DATA_DIR_KEY)));
- } catch (ReconfigurationException e) {
- // This can be thrown if reconfiguration tries to use a failed volume.
- // We need to swallow the exception, because some of our tests want to
- // cover this case.
- LOG.warn("Could not reconfigure DataNode.", e);
- }
- }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/413dccaf/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeMetrics.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeMetrics.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeMetrics.java
new file mode 100644
index 0000000..407c3e9
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeMetrics.java
@@ -0,0 +1,182 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.datanode;
+
+import static org.apache.hadoop.test.MetricsAsserts.assertCounter;
+import static org.apache.hadoop.test.MetricsAsserts.getMetrics;
+import static org.junit.Assert.assertTrue;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.StorageType;
+import org.apache.hadoop.hdfs.DFSConfigKeys;
+import org.apache.hadoop.hdfs.DFSOutputStream;
+import org.apache.hadoop.hdfs.DFSTestUtil;
+import org.apache.hadoop.hdfs.HdfsConfiguration;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
+import org.apache.hadoop.hdfs.server.datanode.fsdataset.DataNodeVolumeMetrics;
+import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi;
+import org.apache.hadoop.metrics2.MetricsRecordBuilder;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.Timeout;
+
+/**
+ * Test class for DataNodeVolumeMetrics.
+ */
+public class TestDataNodeVolumeMetrics {
+ private static final Log LOG =
+ LogFactory.getLog(TestDataNodeVolumeMetrics.class);
+
+ private static final int BLOCK_SIZE = 1024;
+ private static final short REPL = 1;
+ private static final int NUM_DATANODES = 1;
+
+ @Rule
+ public Timeout timeout = new Timeout(300000);
+
+ @Test
+ public void testVolumeMetrics() throws Exception {
+ MiniDFSCluster cluster = setupClusterForVolumeMetrics();
+ try {
+ FileSystem fs = cluster.getFileSystem();
+ final Path fileName = new Path("/test.dat");
+ final long fileLen = Integer.MAX_VALUE + 1L;
+ DFSTestUtil.createFile(fs, fileName, false, BLOCK_SIZE, fileLen,
+ fs.getDefaultBlockSize(fileName),
+ REPL, 1L, true);
+
+ try (FSDataOutputStream out = fs.append(fileName)) {
+ out.writeBytes("hello world");
+ ((DFSOutputStream) out.getWrappedStream()).hsync();
+ }
+
+ verifyDataNodeVolumeMetrics(fs, cluster, fileName);
+ } finally {
+ if (cluster != null) {
+ cluster.shutdown();
+ }
+ }
+ }
+
+ @Test
+ public void testVolumeMetricsWithVolumeDepartureArrival() throws Exception {
+ MiniDFSCluster cluster = setupClusterForVolumeMetrics();
+ try {
+ FileSystem fs = cluster.getFileSystem();
+ final Path fileName = new Path("/test.dat");
+ final long fileLen = Integer.MAX_VALUE + 1L;
+ DFSTestUtil.createFile(fs, fileName, false, BLOCK_SIZE, fileLen,
+ fs.getDefaultBlockSize(fileName),
+ REPL, 1L, true);
+
+ try (FSDataOutputStream out = fs.append(fileName)) {
+ out.writeBytes("hello world");
+ ((DFSOutputStream) out.getWrappedStream()).hsync();
+ }
+
+ ArrayList<DataNode> dns = cluster.getDataNodes();
+ assertTrue("DN1 should be up", dns.get(0).isDatanodeUp());
+
+ final String dataDir = cluster.getDataDirectory();
+ final File dn1Vol2 = new File(dataDir, "data2");
+
+ DataNodeTestUtils.injectDataDirFailure(dn1Vol2);
+ verifyDataNodeVolumeMetrics(fs, cluster, fileName);
+
+ DataNodeTestUtils.restoreDataDirFromFailure(dn1Vol2);
+ DataNodeTestUtils.reconfigureDataNode(dns.get(0), dn1Vol2);
+ verifyDataNodeVolumeMetrics(fs, cluster, fileName);
+ } finally {
+ if (cluster != null) {
+ cluster.shutdown();
+ }
+ }
+ }
+
+ private MiniDFSCluster setupClusterForVolumeMetrics() throws IOException {
+ Configuration conf = new HdfsConfiguration();
+ conf.set(DFSConfigKeys.DFS_DATANODE_FILE_IO_EVENTS_CLASS_KEY,
+ "org.apache.hadoop.hdfs.server.datanode.ProfilingFileIoEvents");
+ SimulatedFSDataset.setFactory(conf);
+ return new MiniDFSCluster.Builder(conf)
+ .numDataNodes(NUM_DATANODES)
+ .storageTypes(new StorageType[]{StorageType.RAM_DISK, StorageType.DISK})
+ .storagesPerDatanode(2)
+ .build();
+ }
+
+ private void verifyDataNodeVolumeMetrics(final FileSystem fs,
+ final MiniDFSCluster cluster, final Path fileName) throws IOException {
+ List<DataNode> datanodes = cluster.getDataNodes();
+ DataNode datanode = datanodes.get(0);
+
+ final ExtendedBlock block = DFSTestUtil.getFirstBlock(fs, fileName);
+ final FsVolumeSpi volume = datanode.getFSDataset().getVolume(block);
+ DataNodeVolumeMetrics metrics = volume.getMetrics();
+
+ MetricsRecordBuilder rb = getMetrics(volume.getMetrics().name());
+ assertCounter("TotalDataFileIos", metrics.getTotalDataFileIos(), rb);
+
+ LOG.info("TotalMetadataOperations : " +
+ metrics.getTotalMetadataOperations());
+ LOG.info("TotalDataFileIos : " + metrics.getTotalDataFileIos());
+ LOG.info("TotalFileIoErrors : " + metrics.getTotalFileIoErrors());
+
+ LOG.info("MetadataOperationSampleCount : " +
+ metrics.getMetadataOperationSampleCount());
+ LOG.info("MetadataOperationMean : " + metrics.getMetadataOperationMean());
+ LOG.info("MetadataFileIoStdDev : " +
+ metrics.getMetadataOperationStdDev());
+
+ LOG.info("DataFileIoSampleCount : " + metrics.getDataFileIoSampleCount());
+ LOG.info("DataFileIoMean : " + metrics.getDataFileIoMean());
+ LOG.info("DataFileIoStdDev : " + metrics.getDataFileIoStdDev());
+
+ LOG.info("flushIoSampleCount : " + metrics.getFlushIoSampleCount());
+ LOG.info("flushIoMean : " + metrics.getFlushIoMean());
+ LOG.info("flushIoStdDev : " + metrics.getFlushIoStdDev());
+
+ LOG.info("syncIoSampleCount : " + metrics.getSyncIoSampleCount());
+ LOG.info("syncIoMean : " + metrics.getSyncIoMean());
+ LOG.info("syncIoStdDev : " + metrics.getSyncIoStdDev());
+
+ LOG.info("readIoSampleCount : " + metrics.getReadIoMean());
+ LOG.info("readIoMean : " + metrics.getReadIoMean());
+ LOG.info("readIoStdDev : " + metrics.getReadIoStdDev());
+
+ LOG.info("writeIoSampleCount : " + metrics.getWriteIoSampleCount());
+ LOG.info("writeIoMean : " + metrics.getWriteIoMean());
+ LOG.info("writeIoStdDev : " + metrics.getWriteIoStdDev());
+
+ LOG.info("fileIoErrorSampleCount : "
+ + metrics.getFileIoErrorSampleCount());
+ LOG.info("fileIoErrorMean : " + metrics.getFileIoErrorMean());
+ LOG.info("fileIoErrorStdDev : " + metrics.getFileIoErrorStdDev());
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/413dccaf/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDirectoryScanner.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDirectoryScanner.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDirectoryScanner.java
index cda6066..6f7a6fa 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDirectoryScanner.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDirectoryScanner.java
@@ -53,6 +53,9 @@ import org.apache.hadoop.hdfs.DFSTestUtil;
import org.apache.hadoop.hdfs.HdfsConfiguration;
import org.apache.hadoop.hdfs.MiniDFSCluster;
import org.apache.hadoop.hdfs.server.datanode.checker.VolumeCheckResult;
+import org.apache.hadoop.hdfs.server.datanode.fsdataset.DataNodeVolumeMetrics;
+import org.apache.hadoop.util.AutoCloseableLock;
+import org.apache.hadoop.util.AutoCloseableLock;
import org.apache.hadoop.hdfs.protocol.Block;
import org.apache.hadoop.hdfs.protocol.HdfsConstants;
import org.apache.hadoop.hdfs.protocol.LocatedBlock;
@@ -892,6 +895,11 @@ public class TestDirectoryScanner {
}
@Override
+ public DataNodeVolumeMetrics getMetrics() {
+ return null;
+ }
+
+ @Override
public VolumeCheckResult check(VolumeCheckContext context)
throws Exception {
return VolumeCheckResult.HEALTHY;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/413dccaf/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/extdataset/ExternalVolumeImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/extdataset/ExternalVolumeImpl.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/extdataset/ExternalVolumeImpl.java
index ff18e84..61c7480 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/extdataset/ExternalVolumeImpl.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/extdataset/ExternalVolumeImpl.java
@@ -25,6 +25,7 @@ import java.nio.channels.ClosedChannelException;
import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdfs.server.datanode.FileIoProvider;
import org.apache.hadoop.hdfs.server.datanode.checker.VolumeCheckResult;
+import org.apache.hadoop.hdfs.server.datanode.fsdataset.DataNodeVolumeMetrics;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsDatasetSpi;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeReference;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi;
@@ -116,6 +117,11 @@ public class ExternalVolumeImpl implements FsVolumeSpi {
}
@Override
+ public DataNodeVolumeMetrics getMetrics() {
+ return null;
+ }
+
+ @Override
public VolumeCheckResult check(VolumeCheckContext context)
throws Exception {
return VolumeCheckResult.HEALTHY;
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[3/5] hadoop git commit: HDFS-11182. Update DataNode to use
DatasetVolumeChecker.
Posted by ar...@apache.org.
HDFS-11182. Update DataNode to use DatasetVolumeChecker.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/ec80de3c
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/ec80de3c
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/ec80de3c
Branch: refs/heads/branch-2
Commit: ec80de3ccc67e258a51b52866c5c0456d706b8eb
Parents: 413dcca
Author: Arpit Agarwal <ar...@apache.org>
Authored: Mon Jan 16 14:40:37 2017 -0800
Committer: Arpit Agarwal <ar...@apache.org>
Committed: Mon Jan 16 22:14:00 2017 -0800
----------------------------------------------------------------------
.../hadoop/hdfs/server/datanode/DataNode.java | 412 +++++++++----------
.../hdfs/server/datanode/StorageLocation.java | 5 -
.../datanode/checker/DatasetVolumeChecker.java | 90 ++--
.../server/datanode/fsdataset/FsDatasetSpi.java | 51 ++-
.../datanode/fsdataset/impl/FsDatasetImpl.java | 106 ++---
.../datanode/fsdataset/impl/FsVolumeImpl.java | 7 -
.../datanode/fsdataset/impl/FsVolumeList.java | 43 +-
.../blockmanagement/TestBlockStatsMXBean.java | 4 +
.../server/datanode/SimulatedFSDataset.java | 175 ++++----
.../datanode/TestDataNodeHotSwapVolumes.java | 3 +
.../datanode/TestDataNodeVolumeFailure.java | 3 +
.../TestDataNodeVolumeFailureReporting.java | 3 +
.../TestDataNodeVolumeFailureToleration.java | 3 +
.../hdfs/server/datanode/TestDiskError.java | 26 +-
.../checker/TestDatasetVolumeChecker.java | 9 +-
.../TestDatasetVolumeCheckerFailures.java | 43 +-
.../extdataset/ExternalDatasetImpl.java | 7 +-
.../fsdataset/impl/TestFsDatasetImpl.java | 86 +---
.../fsdataset/impl/TestFsVolumeList.java | 32 --
19 files changed, 521 insertions(+), 587 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ec80de3c/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
index 0031bff..bddde1f 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
@@ -88,7 +88,6 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledThreadPoolExecutor;
-import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@@ -161,8 +160,9 @@ import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.StartupOption;
import org.apache.hadoop.hdfs.server.common.MetricsLoggerTask;
import org.apache.hadoop.hdfs.server.common.Storage;
import org.apache.hadoop.hdfs.server.common.StorageInfo;
-import org.apache.hadoop.hdfs.server.datanode.SecureDataNodeStarter.SecureResources;
+import org.apache.hadoop.hdfs.server.datanode.checker.DatasetVolumeChecker;
import org.apache.hadoop.hdfs.server.datanode.checker.StorageLocationChecker;
+import org.apache.hadoop.hdfs.server.datanode.SecureDataNodeStarter.SecureResources;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsDatasetSpi;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi;
import org.apache.hadoop.hdfs.server.datanode.metrics.DataNodeMetrics;
@@ -249,7 +249,7 @@ import org.slf4j.LoggerFactory;
* to a DataNode directly; a NameNode simply returns values from
* functions invoked by a DataNode.
*
- * DataNodes maintain an open server socket so that client code
+ * DataNodes maintain an open server socket so that client code
* or other DataNodes can read/write data. The host/port for
* this server is reported to the NameNode, which then sends that
* information to clients or other DataNodes that might be interested.
@@ -260,7 +260,7 @@ public class DataNode extends ReconfigurableBase
implements InterDatanodeProtocol, ClientDatanodeProtocol,
TraceAdminProtocol, DataNodeMXBean, ReconfigurationProtocol {
public static final Logger LOG = LoggerFactory.getLogger(DataNode.class);
-
+
static{
HdfsConfiguration.init();
}
@@ -275,10 +275,10 @@ public class DataNode extends ReconfigurableBase
", srvID: %s" + // DatanodeRegistration
", blockid: %s" + // block id
", duration(ns): %s"; // duration time
-
+
static final Log ClientTraceLog =
LogFactory.getLog(DataNode.class.getName() + ".clienttrace");
-
+
private static final String USAGE =
"Usage: hdfs datanode [-regular | -rollback | -rollingupgrade rollback" +
" ]\n" +
@@ -309,7 +309,7 @@ public class DataNode extends ReconfigurableBase
public static InetSocketAddress createSocketAddr(String target) {
return NetUtils.createSocketAddr(target);
}
-
+
volatile boolean shouldRun = true;
volatile boolean shutdownForUpgrade = false;
private boolean shutdownInProgress = false;
@@ -334,24 +334,24 @@ public class DataNode extends ReconfigurableBase
DataNodeMetrics metrics;
private InetSocketAddress streamingAddr;
-
+
// See the note below in incrDatanodeNetworkErrors re: concurrency.
private LoadingCache<String, Map<String, Long>> datanodeNetworkCounts;
private String hostName;
private DatanodeID id;
-
+
final private String fileDescriptorPassingDisabledReason;
boolean isBlockTokenEnabled;
BlockPoolTokenSecretManager blockPoolTokenSecretManager;
private boolean hasAnyBlockPoolRegistered = false;
-
+
private BlockScanner blockScanner;
private DirectoryScanner directoryScanner = null;
-
+
/** Activated plug-ins. */
private List<ServicePlugin> plugins;
-
+
// For InterDataNodeProtocol
public RPC.Server ipcServer;
@@ -371,11 +371,7 @@ public class DataNode extends ReconfigurableBase
SaslDataTransferServer saslServer;
private final boolean getHdfsBlockLocationsEnabled;
private ObjectName dataNodeInfoBeanName;
- private Thread checkDiskErrorThread = null;
- protected final int checkDiskErrorInterval;
- private boolean checkDiskErrorFlag = false;
- private Object checkDiskErrorMutex = new Object();
- private long lastDiskErrorCheck;
+ private volatile long lastDiskErrorCheck;
private String supergroup;
private boolean isPermissionEnabled;
private String dnUserName = null;
@@ -389,6 +385,8 @@ public class DataNode extends ReconfigurableBase
@Nullable
private final StorageLocationChecker storageLocationChecker;
+ private final DatasetVolumeChecker volumeChecker;
+
private static Tracer createTracer(Configuration conf) {
return new Tracer.Builder("DataNode").
conf(TraceUtils.wrapHadoopConf(DATANODE_HTRACE_PREFIX , conf)).
@@ -404,7 +402,7 @@ public class DataNode extends ReconfigurableBase
*/
@VisibleForTesting
@InterfaceAudience.LimitedPrivate("HDFS")
- DataNode(final Configuration conf) {
+ DataNode(final Configuration conf) throws DiskErrorException {
super(conf);
this.blockScanner = new BlockScanner(this, conf);
this.tracer = createTracer(conf);
@@ -418,10 +416,9 @@ public class DataNode extends ReconfigurableBase
this.connectToDnViaHostname = false;
this.getHdfsBlockLocationsEnabled = false;
this.pipelineSupportECN = false;
- this.checkDiskErrorInterval =
- ThreadLocalRandom.current().nextInt(5000, (int) (5000 * 1.25));
initOOBTimeout();
storageLocationChecker = null;
+ volumeChecker = new DatasetVolumeChecker(conf, new Timer());
}
/**
@@ -448,7 +445,7 @@ public class DataNode extends ReconfigurableBase
DFSConfigKeys.DFS_DATANODE_USE_DN_HOSTNAME,
DFSConfigKeys.DFS_DATANODE_USE_DN_HOSTNAME_DEFAULT);
this.getHdfsBlockLocationsEnabled = conf.getBoolean(
- DFSConfigKeys.DFS_HDFS_BLOCKS_METADATA_ENABLED,
+ DFSConfigKeys.DFS_HDFS_BLOCKS_METADATA_ENABLED,
DFSConfigKeys.DFS_HDFS_BLOCKS_METADATA_ENABLED_DEFAULT);
this.supergroup = conf.get(DFSConfigKeys.DFS_PERMISSIONS_SUPERUSERGROUP_KEY,
DFSConfigKeys.DFS_PERMISSIONS_SUPERUSERGROUP_DEFAULT);
@@ -464,8 +461,7 @@ public class DataNode extends ReconfigurableBase
",hdfs-" +
conf.get("hadoop.hdfs.configuration.version", "UNSPECIFIED");
- this.checkDiskErrorInterval =
- ThreadLocalRandom.current().nextInt(5000, (int) (5000 * 1.25));
+ this.volumeChecker = new DatasetVolumeChecker(conf, new Timer());
// Determine whether we should try to pass file descriptors to clients.
if (conf.getBoolean(HdfsClientConfigKeys.Read.ShortCircuit.KEY,
@@ -861,7 +857,7 @@ public class DataNode extends ReconfigurableBase
private synchronized void setClusterId(final String nsCid, final String bpid
) throws IOException {
if(clusterId != null && !clusterId.equals(nsCid)) {
- throw new IOException ("Cluster IDs not matched: dn cid=" + clusterId
+ throw new IOException ("Cluster IDs not matched: dn cid=" + clusterId
+ " but ns cid="+ nsCid + "; bpid=" + bpid);
}
// else
@@ -941,11 +937,11 @@ public class DataNode extends ReconfigurableBase
private void initIpcServer() throws IOException {
InetSocketAddress ipcAddr = NetUtils.createSocketAddr(
getConf().getTrimmed(DFS_DATANODE_IPC_ADDRESS_KEY));
-
- // Add all the RPC protocols that the Datanode implements
+
+ // Add all the RPC protocols that the Datanode implements
RPC.setProtocolEngine(getConf(), ClientDatanodeProtocolPB.class,
ProtobufRpcEngine.class);
- ClientDatanodeProtocolServerSideTranslatorPB clientDatanodeProtocolXlator =
+ ClientDatanodeProtocolServerSideTranslatorPB clientDatanodeProtocolXlator =
new ClientDatanodeProtocolServerSideTranslatorPB(this);
BlockingService service = ClientDatanodeProtocolService
.newReflectiveBlockingService(clientDatanodeProtocolXlator);
@@ -966,7 +962,7 @@ public class DataNode extends ReconfigurableBase
DFSUtil.addPBProtocol(getConf(), ReconfigurationProtocolPB.class, service,
ipcServer);
- InterDatanodeProtocolServerSideTranslatorPB interDatanodeProtocolXlator =
+ InterDatanodeProtocolServerSideTranslatorPB interDatanodeProtocolXlator =
new InterDatanodeProtocolServerSideTranslatorPB(this);
service = InterDatanodeProtocolService
.newReflectiveBlockingService(interDatanodeProtocolXlator);
@@ -1032,12 +1028,12 @@ public class DataNode extends ReconfigurableBase
return;
}
String reason = null;
- if (conf.getInt(DFS_DATANODE_DIRECTORYSCAN_INTERVAL_KEY,
+ if (conf.getInt(DFS_DATANODE_DIRECTORYSCAN_INTERVAL_KEY,
DFS_DATANODE_DIRECTORYSCAN_INTERVAL_DEFAULT) < 0) {
reason = "verification is turned off by configuration";
} else if ("SimulatedFSDataset".equals(data.getClass().getSimpleName())) {
reason = "verifcation is not supported by SimulatedFSDataset";
- }
+ }
if (reason == null) {
directoryScanner = new DirectoryScanner(this, data, conf);
directoryScanner.start();
@@ -1046,13 +1042,13 @@ public class DataNode extends ReconfigurableBase
reason);
}
}
-
+
private synchronized void shutdownDirectoryScanner() {
if (directoryScanner != null) {
directoryScanner.shutdown();
}
}
-
+
private void initDataXceiver() throws IOException {
// find free port or use privileged port provided
TcpPeerServer tcpPeerServer;
@@ -1127,7 +1123,7 @@ public class DataNode extends ReconfigurableBase
}
return domainPeerServer;
}
-
+
// calls specific to BP
public void notifyNamenodeReceivedBlock(ExtendedBlock block, String delHint,
String storageUuid, boolean isOnTransientStorage) {
@@ -1140,7 +1136,7 @@ public class DataNode extends ReconfigurableBase
+ block.getBlockPoolId());
}
}
-
+
// calls specific to BP
protected void notifyNamenodeReceivingBlock(
ExtendedBlock block, String storageUuid) {
@@ -1152,7 +1148,7 @@ public class DataNode extends ReconfigurableBase
+ block.getBlockPoolId());
}
}
-
+
/** Notify the corresponding namenode to delete the block. */
public void notifyNamenodeDeletedBlock(ExtendedBlock block, String storageUuid) {
BPOfferService bpos = blockPoolManager.get(block.getBlockPoolId());
@@ -1163,7 +1159,7 @@ public class DataNode extends ReconfigurableBase
+ block.getBlockPoolId());
}
}
-
+
/**
* Report a bad block which is hosted on the local DN.
*/
@@ -1202,7 +1198,7 @@ public class DataNode extends ReconfigurableBase
BPOfferService bpos = getBPOSForBlock(block);
bpos.reportRemoteBadBlock(srcDataNode, block);
}
-
+
/**
* Try to send an error report to the NNs associated with the given
* block pool.
@@ -1258,10 +1254,10 @@ public class DataNode extends ReconfigurableBase
/**
* This method starts the data node with the specified conf.
- *
+ *
* If conf's CONFIG_PROPERTY_SIMULATED property is set
* then a simulated storage based data node is created.
- *
+ *
* @param dataDirs - only for a non-simulated storage data node
* @throws IOException
*/
@@ -1312,7 +1308,7 @@ public class DataNode extends ReconfigurableBase
}
storage = new DataStorage();
-
+
// global DN settings
registerMXBean();
initDataXceiver();
@@ -1320,7 +1316,7 @@ public class DataNode extends ReconfigurableBase
pauseMonitor = new JvmPauseMonitor();
pauseMonitor.init(getConf());
pauseMonitor.start();
-
+
// BlockPoolTokenSecretManager is required to create ipc server.
this.blockPoolTokenSecretManager = new BlockPoolTokenSecretManager();
@@ -1404,7 +1400,7 @@ public class DataNode extends ReconfigurableBase
"protection and SSL for HTTP. Using privileged resources in " +
"combination with SASL RPC data transfer protection is not supported.");
}
-
+
public static String generateUuid() {
return UUID.randomUUID().toString();
}
@@ -1439,10 +1435,10 @@ public class DataNode extends ReconfigurableBase
}
DatanodeID dnId = new DatanodeID(
- streamingAddr.getAddress().getHostAddress(), hostName,
+ streamingAddr.getAddress().getHostAddress(), hostName,
storage.getDatanodeUuid(), getXferPort(), getInfoPort(),
infoSecurePort, getIpcPort());
- return new DatanodeRegistration(dnId, storageInfo,
+ return new DatanodeRegistration(dnId, storageInfo,
new ExportedBlockKeys(), VersionInfo.getVersion());
}
@@ -1461,10 +1457,10 @@ public class DataNode extends ReconfigurableBase
+ bpRegistration.getDatanodeUuid()
+ ". Expecting " + storage.getDatanodeUuid());
}
-
+
registerBlockPoolWithSecretManager(bpRegistration, blockPoolId);
}
-
+
/**
* After the block pool has contacted the NN, registers that block pool
* with the secret manager, updating it with the secrets provided by the NN.
@@ -1484,7 +1480,7 @@ public class DataNode extends ReconfigurableBase
}
}
if (!isBlockTokenEnabled) return;
-
+
if (!blockPoolTokenSecretManager.isBlockPoolRegistered(blockPoolId)) {
long blockKeyUpdateInterval = keys.getKeyUpdateInterval();
long blockTokenLifetime = keys.getTokenLifetime();
@@ -1493,7 +1489,7 @@ public class DataNode extends ReconfigurableBase
+ blockKeyUpdateInterval / (60 * 1000)
+ " min(s), tokenLifetime=" + blockTokenLifetime / (60 * 1000)
+ " min(s)");
- final BlockTokenSecretManager secretMgr =
+ final BlockTokenSecretManager secretMgr =
new BlockTokenSecretManager(0, blockTokenLifetime, blockPoolId,
dnConf.encryptionAlgorithm);
blockPoolTokenSecretManager.addBlockPool(blockPoolId, secretMgr);
@@ -1528,10 +1524,10 @@ public class DataNode extends ReconfigurableBase
* One of the Block Pools has successfully connected to its NN.
* This initializes the local storage for that block pool,
* checks consistency of the NN's cluster ID, etc.
- *
+ *
* If this is the first block pool to register, this also initializes
* the datanode-scoped storage.
- *
+ *
* @param bpos Block pool offer service
* @throws IOException if the NN is inconsistent with the local storage.
*/
@@ -1541,12 +1537,12 @@ public class DataNode extends ReconfigurableBase
throw new IOException("NamespaceInfo not found: Block pool " + bpos
+ " should have retrieved namespace info before initBlockPool.");
}
-
+
setClusterId(nsInfo.clusterID, nsInfo.getBlockPoolID());
// Register the new block pool with the BP manager.
blockPoolManager.addBlockPool(bpos);
-
+
// In the case that this is the first block pool to connect, initialize
// the dataset, block scanners, etc.
initStorage(nsInfo);
@@ -1567,11 +1563,11 @@ public class DataNode extends ReconfigurableBase
BPOfferService getBPOfferService(String bpid){
return blockPoolManager.get(bpid);
}
-
+
int getBpOsCount() {
return blockPoolManager.getAllNamenodeThreads().size();
}
-
+
/**
* Initializes the {@link #data}. The initialization is done only once, when
* handshake with the the first namenode is completed.
@@ -1579,7 +1575,7 @@ public class DataNode extends ReconfigurableBase
private void initStorage(final NamespaceInfo nsInfo) throws IOException {
final FsDatasetSpi.Factory<? extends FsDatasetSpi<?>> factory
= FsDatasetSpi.Factory.getFactory(getConf());
-
+
if (!factory.isSimulated()) {
final StartupOption startOpt = getStartupOption(getConf());
if (startOpt == null) {
@@ -1613,21 +1609,21 @@ public class DataNode extends ReconfigurableBase
return NetUtils.createSocketAddr(conf.getTrimmed(DFS_DATANODE_HTTP_ADDRESS_KEY,
DFS_DATANODE_HTTP_ADDRESS_DEFAULT));
}
-
+
private void registerMXBean() {
dataNodeInfoBeanName = MBeans.register("DataNode", "DataNodeInfo", this);
}
-
+
@VisibleForTesting
public DataXceiverServer getXferServer() {
- return xserver;
+ return xserver;
}
-
+
@VisibleForTesting
public int getXferPort() {
return streamingAddr.getPort();
}
-
+
/**
* @return name useful for logging
*/
@@ -1640,7 +1636,7 @@ public class DataNode extends ReconfigurableBase
* NB: The datanode can perform data transfer on the streaming
* address however clients are given the IPC IP address for data
* transfer, and that may be a different address.
- *
+ *
* @return socket address for data transfer
*/
public InetSocketAddress getXferAddress() {
@@ -1653,14 +1649,14 @@ public class DataNode extends ReconfigurableBase
public int getIpcPort() {
return ipcServer.getListenerAddress().getPort();
}
-
+
/**
* get BP registration by blockPool id
* @return BP registration object
* @throws IOException on error
*/
@VisibleForTesting
- public DatanodeRegistration getDNRegistrationForBP(String bpid)
+ public DatanodeRegistration getDNRegistrationForBP(String bpid)
throws IOException {
DataNodeFaultInjector.get().noRegistration();
BPOfferService bpos = blockPoolManager.get(bpid);
@@ -1669,13 +1665,13 @@ public class DataNode extends ReconfigurableBase
}
return bpos.bpRegistration;
}
-
+
/**
* Creates either NIO or regular depending on socketWriteTimeout.
*/
protected Socket newSocket() throws IOException {
- return (dnConf.socketWriteTimeout > 0) ?
- SocketChannel.open().socket() : new Socket();
+ return (dnConf.socketWriteTimeout > 0) ?
+ SocketChannel.open().socket() : new Socket();
}
/**
@@ -1721,24 +1717,24 @@ public class DataNode extends ReconfigurableBase
throw new IOException(ie.getMessage());
}
}
-
+
public DataNodeMetrics getMetrics() {
return metrics;
}
-
+
/** Ensure the authentication method is kerberos */
private void checkKerberosAuthMethod(String msg) throws IOException {
// User invoking the call must be same as the datanode user
if (!UserGroupInformation.isSecurityEnabled()) {
return;
}
- if (UserGroupInformation.getCurrentUser().getAuthenticationMethod() !=
+ if (UserGroupInformation.getCurrentUser().getAuthenticationMethod() !=
AuthenticationMethod.KERBEROS) {
throw new AccessControlException("Error in " + msg
+ "Only kerberos based authentication is allowed.");
}
}
-
+
private void checkBlockLocalPathAccess() throws IOException {
checkKerberosAuthMethod("getBlockLocalPathInfo()");
String currentUser = UserGroupInformation.getCurrentUser().getShortUserName();
@@ -1797,7 +1793,7 @@ public class DataNode extends ReconfigurableBase
}
FileInputStream[] requestShortCircuitFdsForRead(final ExtendedBlock blk,
- final Token<BlockTokenIdentifier> token, int maxVersion)
+ final Token<BlockTokenIdentifier> token, int maxVersion)
throws ShortCircuitFdsUnsupportedException,
ShortCircuitFdsVersionException, IOException {
if (fileDescriptorPassingDisabledReason != null) {
@@ -1807,13 +1803,13 @@ public class DataNode extends ReconfigurableBase
int blkVersion = CURRENT_BLOCK_FORMAT_VERSION;
if (maxVersion < blkVersion) {
throw new ShortCircuitFdsVersionException("Your client is too old " +
- "to read this block! Its format version is " +
+ "to read this block! Its format version is " +
blkVersion + ", but the highest format version you can read is " +
maxVersion);
}
metrics.incrBlocksGetLocalPathInfo();
FileInputStream fis[] = new FileInputStream[2];
-
+
try {
fis[0] = (FileInputStream)data.getBlockInputStream(blk, 0);
fis[1] = DatanodeUtil.getMetaDataInputStream(blk, data);
@@ -1828,7 +1824,7 @@ public class DataNode extends ReconfigurableBase
@Override
public HdfsBlocksMetadata getHdfsBlocksMetadata(
String bpId, long[] blockIds,
- List<Token<BlockTokenIdentifier>> tokens) throws IOException,
+ List<Token<BlockTokenIdentifier>> tokens) throws IOException,
UnsupportedOperationException {
if (!getHdfsBlockLocationsEnabled) {
throw new UnsupportedOperationException("Datanode#getHdfsBlocksMetadata "
@@ -1847,7 +1843,7 @@ public class DataNode extends ReconfigurableBase
return data.getHdfsBlocksMetadata(bpId, blockIds);
}
-
+
private void checkBlockToken(ExtendedBlock block, Token<BlockTokenIdentifier> token,
AccessMode accessMode) throws IOException {
if (isBlockTokenEnabled) {
@@ -1880,11 +1876,11 @@ public class DataNode extends ReconfigurableBase
}
}
}
-
+
List<BPOfferService> bposArray = (this.blockPoolManager == null)
? new ArrayList<BPOfferService>()
: this.blockPoolManager.getAllNamenodeThreads();
- // If shutdown is not for restart, set shouldRun to false early.
+ // If shutdown is not for restart, set shouldRun to false early.
if (!shutdownForUpgrade) {
shouldRun = false;
}
@@ -1903,11 +1899,6 @@ public class DataNode extends ReconfigurableBase
}
}
- // Interrupt the checkDiskErrorThread and terminate it.
- if(this.checkDiskErrorThread != null) {
- this.checkDiskErrorThread.interrupt();
- }
-
// Record the time of initial notification
long timeNotified = Time.monotonicNow();
@@ -1928,6 +1919,8 @@ public class DataNode extends ReconfigurableBase
}
}
+ volumeChecker.shutdownAndWait(1, TimeUnit.SECONDS);
+
if (storageLocationChecker != null) {
storageLocationChecker.shutdownAndWait(1, TimeUnit.SECONDS);
}
@@ -1939,7 +1932,7 @@ public class DataNode extends ReconfigurableBase
// shouldRun is set to false here to prevent certain threads from exiting
// before the restart prep is done.
this.shouldRun = false;
-
+
// wait reconfiguration thread, if any, to exit
shutdownReconfigurationTask();
@@ -1984,7 +1977,7 @@ public class DataNode extends ReconfigurableBase
} catch (InterruptedException ie) {
}
}
-
+
// IPC server needs to be shutdown late in the process, otherwise
// shutdown command response won't get sent.
if (ipcServer != null) {
@@ -1998,7 +1991,7 @@ public class DataNode extends ReconfigurableBase
LOG.warn("Received exception in BlockPoolManager#shutDownAll: ", ie);
}
}
-
+
if (storage != null) {
try {
this.storage.unlockAll();
@@ -2026,46 +2019,53 @@ public class DataNode extends ReconfigurableBase
}
tracer.close();
}
-
-
+
+
/**
* Check if there is a disk failure asynchronously and if so, handle the error
*/
public void checkDiskErrorAsync() {
- synchronized(checkDiskErrorMutex) {
- checkDiskErrorFlag = true;
- if(checkDiskErrorThread == null) {
- startCheckDiskErrorThread();
- checkDiskErrorThread.start();
- LOG.info("Starting CheckDiskError Thread");
- }
- }
+ volumeChecker.checkAllVolumesAsync(
+ data, new DatasetVolumeChecker.Callback() {
+ @Override
+ public void call(Set<FsVolumeSpi> healthyVolumes,
+ Set<FsVolumeSpi> failedVolumes) {
+ if (failedVolumes.size() > 0) {
+ LOG.warn("checkDiskErrorAsync callback got {} failed volumes: {}",
+ failedVolumes.size(), failedVolumes);
+ } else {
+ LOG.debug("checkDiskErrorAsync: no volume failures detected");
+ }
+ lastDiskErrorCheck = Time.monotonicNow();
+ DataNode.this.handleVolumeFailures(failedVolumes);
+ }
+ });
}
-
+
private void handleDiskError(String errMsgr) {
final boolean hasEnoughResources = data.hasEnoughResource();
LOG.warn("DataNode.handleDiskError: Keep Running: " + hasEnoughResources);
-
- // If we have enough active valid volumes then we do not want to
+
+ // If we have enough active valid volumes then we do not want to
// shutdown the DN completely.
- int dpError = hasEnoughResources ? DatanodeProtocol.DISK_ERROR
- : DatanodeProtocol.FATAL_DISK_ERROR;
+ int dpError = hasEnoughResources ? DatanodeProtocol.DISK_ERROR
+ : DatanodeProtocol.FATAL_DISK_ERROR;
metrics.incrVolumeFailures();
//inform NameNodes
for(BPOfferService bpos: blockPoolManager.getAllNamenodeThreads()) {
bpos.trySendErrorReport(dpError, errMsgr);
}
-
+
if(hasEnoughResources) {
scheduleAllBlockReport(0);
return; // do not shutdown
}
-
+
LOG.warn("DataNode is shutting down: " + errMsgr);
shouldRun = false;
}
-
+
/** Number of concurrent xceivers per node. */
@Override // DataNodeMXBean
public int getXceiverCount() {
@@ -2136,9 +2136,9 @@ public class DataNode extends ReconfigurableBase
lengthTooShort = true;
} catch (IOException e) {
// The IOException indicates not being able to access block file,
- // treat it the same here as blockFileNotExist, to trigger
+ // treat it the same here as blockFileNotExist, to trigger
// reporting it as a bad block
- blockFileNotExist = true;
+ blockFileNotExist = true;
}
if (replicaNotExist || replicaStateNotFinalized) {
@@ -2154,14 +2154,14 @@ public class DataNode extends ReconfigurableBase
return;
}
if (lengthTooShort) {
- // Check if NN recorded length matches on-disk length
+ // Check if NN recorded length matches on-disk length
// Shorter on-disk len indicates corruption so report NN the corrupt block
reportBadBlock(bpos, block, "Can't replicate block " + block
- + " because on-disk length " + data.getLength(block)
+ + " because on-disk length " + data.getLength(block)
+ " is shorter than NameNode recorded length " + block.getNumBytes());
return;
}
-
+
int numTargets = xferTargets.length;
if (numTargets > 0) {
StringBuilder xfersBuilder = new StringBuilder();
@@ -2169,8 +2169,8 @@ public class DataNode extends ReconfigurableBase
xfersBuilder.append(xferTargets[i]);
xfersBuilder.append(" ");
}
- LOG.info(bpReg + " Starting thread to transfer " +
- block + " to " + xfersBuilder);
+ LOG.info(bpReg + " Starting thread to transfer " +
+ block + " to " + xfersBuilder);
new Daemon(new DataTransfer(xferTargets, xferTargetStorageTypes, block,
BlockConstructionStage.PIPELINE_SETUP_CREATE, "")).start();
@@ -2191,27 +2191,27 @@ public class DataNode extends ReconfigurableBase
/* ********************************************************************
Protocol when a client reads data from Datanode (Cur Ver: 9):
-
+
Client's Request :
=================
-
+
Processed in DataXceiver:
+----------------------------------------------+
| Common Header | 1 byte OP == OP_READ_BLOCK |
+----------------------------------------------+
-
+
Processed in readBlock() :
+-------------------------------------------------------------------------+
| 8 byte Block ID | 8 byte genstamp | 8 byte start offset | 8 byte length |
+-------------------------------------------------------------------------+
| vInt length | <DFSClient id> |
+-----------------------------------+
-
+
Client sends optional response only at the end of receiving data.
-
+
DataNode Response :
===================
-
+
In readBlock() :
If there is an error while initializing BlockSender :
+---------------------------+
@@ -2221,21 +2221,21 @@ public class DataNode extends ReconfigurableBase
+---------------------------+
| 2 byte OP_STATUS_SUCCESS |
+---------------------------+
-
+
Actual data, sent by BlockSender.sendBlock() :
-
+
ChecksumHeader :
+--------------------------------------------------+
| 1 byte CHECKSUM_TYPE | 4 byte BYTES_PER_CHECKSUM |
+--------------------------------------------------+
- Followed by actual data in the form of PACKETS:
+ Followed by actual data in the form of PACKETS:
+------------------------------------+
| Sequence of data PACKETs .... |
+------------------------------------+
-
+
A "PACKET" is defined further below.
-
- The client reads data until it receives a packet with
+
+ The client reads data until it receives a packet with
"LastPacketInBlock" set to true or with a zero length. It then replies
to DataNode with one of the status codes:
- CHECKSUM_OK: All the chunk checksums have been verified
@@ -2245,16 +2245,16 @@ public class DataNode extends ReconfigurableBase
+---------------+
| 2 byte Status |
+---------------+
-
+
The DataNode expects all well behaved clients to send the 2 byte
status code. And if the the client doesn't, the DN will close the
connection. So the status code is optional in the sense that it
does not affect the correctness of the data. (And the client can
always reconnect.)
-
+
PACKET : Contains a packet header, checksum and data. Amount of data
======== carried is set by BUFFER_SIZE.
-
+
+-----------------------------------------------------+
| 4 byte packet length (excluding packet header) |
+-----------------------------------------------------+
@@ -2268,15 +2268,15 @@ public class DataNode extends ReconfigurableBase
+-----------------------------------------------------+
| actual data ...... |
+-----------------------------------------------------+
-
+
x = (length of data + BYTE_PER_CHECKSUM - 1)/BYTES_PER_CHECKSUM *
CHECKSUM_SIZE
-
+
CHECKSUM_SIZE depends on CHECKSUM_TYPE (usually, 4 for CRC32)
-
+
The above packet format is used while writing data to DFS also.
Not all the fields might be used while reading.
-
+
************************************************************************ */
/**
@@ -2293,7 +2293,7 @@ public class DataNode extends ReconfigurableBase
final CachingStrategy cachingStrategy;
/**
- * Connect to the first item in the target list. Pass along the
+ * Connect to the first item in the target list. Pass along the
* entire target list, the block, and the data.
*/
DataTransfer(DatanodeInfo targets[], StorageType[] targetStorageTypes,
@@ -2330,7 +2330,7 @@ public class DataNode extends ReconfigurableBase
DataInputStream in = null;
BlockSender blockSender = null;
final boolean isClient = clientname.length() > 0;
-
+
try {
final String dnAddr = targets[0].getXferAddr(connectToDnViaHostname);
InetSocketAddress curTarget = NetUtils.createSocketAddr(dnAddr);
@@ -2346,11 +2346,11 @@ public class DataNode extends ReconfigurableBase
//
Token<BlockTokenIdentifier> accessToken = BlockTokenSecretManager.DUMMY_TOKEN;
if (isBlockTokenEnabled) {
- accessToken = blockPoolTokenSecretManager.generateToken(b,
+ accessToken = blockPoolTokenSecretManager.generateToken(b,
EnumSet.of(BlockTokenIdentifier.AccessMode.WRITE));
}
- long writeTimeout = dnConf.socketWriteTimeout +
+ long writeTimeout = dnConf.socketWriteTimeout +
HdfsConstants.WRITE_TIMEOUT_EXTENSION * (targets.length-1);
OutputStream unbufOut = NetUtils.getOutputStream(sock, writeTimeout);
InputStream unbufIn = NetUtils.getInputStream(sock);
@@ -2360,11 +2360,11 @@ public class DataNode extends ReconfigurableBase
unbufIn, keyFactory, accessToken, bpReg);
unbufOut = saslStreams.out;
unbufIn = saslStreams.in;
-
+
out = new DataOutputStream(new BufferedOutputStream(unbufOut,
DFSUtilClient.getSmallBufferSize(getConf())));
in = new DataInputStream(unbufIn);
- blockSender = new BlockSender(b, 0, b.getNumBytes(),
+ blockSender = new BlockSender(b, 0, b.getNumBytes(),
false, false, true, DataNode.this, null, cachingStrategy);
DatanodeInfo srcNode = new DatanodeInfoBuilder().setNodeID(bpReg)
.build();
@@ -2486,28 +2486,28 @@ public class DataNode extends ReconfigurableBase
}
/** Instantiate a single datanode object. This must be run by invoking
- * {@link DataNode#runDatanodeDaemon()} subsequently.
+ * {@link DataNode#runDatanodeDaemon()} subsequently.
*/
public static DataNode instantiateDataNode(String args[],
Configuration conf) throws IOException {
return instantiateDataNode(args, conf, null);
}
-
- /** Instantiate a single datanode object, along with its secure resources.
- * This must be run by invoking{@link DataNode#runDatanodeDaemon()}
- * subsequently.
+
+ /** Instantiate a single datanode object, along with its secure resources.
+ * This must be run by invoking{@link DataNode#runDatanodeDaemon()}
+ * subsequently.
*/
public static DataNode instantiateDataNode(String args [], Configuration conf,
SecureResources resources) throws IOException {
if (conf == null)
conf = new HdfsConfiguration();
-
+
if (args != null) {
// parse generic hadoop options
GenericOptionsParser hParser = new GenericOptionsParser(conf, args);
args = hParser.getRemainingArgs();
}
-
+
if (!parseArguments(args, conf)) {
printUsage(System.err);
return null;
@@ -2555,7 +2555,7 @@ public class DataNode extends ReconfigurableBase
Configuration conf) throws IOException {
return createDataNode(args, conf, null);
}
-
+
/** Instantiate & Start a single datanode daemon and wait for it to finish.
* If this thread is specifically interrupted, it will stop waiting.
*/
@@ -2667,7 +2667,7 @@ public class DataNode extends ReconfigurableBase
}
/**
- * This methods arranges for the data node to send
+ * This methods arranges for the data node to send
* the block report at the next heartbeat.
*/
public void scheduleAllBlockReport(long delay) {
@@ -2679,7 +2679,7 @@ public class DataNode extends ReconfigurableBase
/**
* Examples are adding and deleting blocks directly.
* The most common usage will be when the data node's storage is simulated.
- *
+ *
* @return the fsdataset that stores the blocks
*/
@VisibleForTesting
@@ -2725,7 +2725,7 @@ public class DataNode extends ReconfigurableBase
terminate(errorCode);
}
}
-
+
public static void main(String args[]) {
if (DFSUtil.parseHelpArgument(args, DataNode.USAGE, System.out, true)) {
System.exit(0);
@@ -2742,7 +2742,7 @@ public class DataNode extends ReconfigurableBase
}
/**
- * Update replica with the new generation stamp and length.
+ * Update replica with the new generation stamp and length.
*/
@Override // InterDatanodeProtocol
public String updateReplicaUnderRecovery(final ExtendedBlock oldBlock,
@@ -2802,7 +2802,7 @@ public class DataNode extends ReconfigurableBase
* @param b the block to transfer.
* The corresponding replica must be an RBW or a Finalized.
* Its GS and numBytes will be set to
- * the stored GS and the visible length.
+ * the stored GS and the visible length.
* @param targets targets to transfer the block to
* @param client client name
*/
@@ -2857,12 +2857,12 @@ public class DataNode extends ReconfigurableBase
return NetUtils.createSocketAddr(
conf.getTrimmed(DFS_DATANODE_ADDRESS_KEY, DFS_DATANODE_ADDRESS_DEFAULT));
}
-
+
@Override // DataNodeMXBean
public String getVersion() {
return VersionInfo.getVersion();
}
-
+
@Override // DataNodeMXBean
public String getRpcPort(){
InetSocketAddress ipcAddr = NetUtils.createSocketAddr(
@@ -2881,7 +2881,7 @@ public class DataNode extends ReconfigurableBase
public String getHttpPort(){
return this.getConf().get("dfs.datanode.info.port");
}
-
+
/**
* @return the datanode's http port
*/
@@ -2897,7 +2897,7 @@ public class DataNode extends ReconfigurableBase
}
/**
- * Returned information is a JSON representation of a map with
+ * Returned information is a JSON representation of a map with
* name node host name as the key and block pool Id as the value.
* Note that, if there are multiple NNs in an NA nameservice,
* a given block pool may be represented twice.
@@ -2936,8 +2936,8 @@ public class DataNode extends ReconfigurableBase
}
/**
- * Returned information is a JSON representation of a map with
- * volume name as the key and value is a map of volume attribute
+ * Returned information is a JSON representation of a map with
+ * volume name as the key and value is a map of volume attribute
* keys to its values
*/
@Override // DataNodeMXBean
@@ -2945,13 +2945,12 @@ public class DataNode extends ReconfigurableBase
Preconditions.checkNotNull(data, "Storage not yet initialized");
return JSON.toString(data.getVolumeInfoMap());
}
-
+
@Override // DataNodeMXBean
public synchronized String getClusterId() {
return clusterId;
}
-
@Override // DataNodeMXBean
public String getFileIoProviderStatistics() {
return fileIoProvider.getStatistics();
@@ -2967,7 +2966,7 @@ public class DataNode extends ReconfigurableBase
setConf(new Configuration());
refreshNamenodes(getConf());
}
-
+
@Override // ClientDatanodeProtocol
public void deleteBlockPool(String blockPoolId, boolean force)
throws IOException {
@@ -2981,7 +2980,7 @@ public class DataNode extends ReconfigurableBase
"The block pool is still running. First do a refreshNamenodes to " +
"shutdown the block pool service");
}
-
+
data.deleteBlockPool(blockPoolId, force);
}
@@ -3077,7 +3076,7 @@ public class DataNode extends ReconfigurableBase
}
return false;
}
-
+
/**
* @param bpid block pool Id
* @return true - if BPOfferService thread is alive
@@ -3094,7 +3093,7 @@ public class DataNode extends ReconfigurableBase
/**
* A datanode is considered to be fully started if all the BP threads are
* alive and all the block pools are initialized.
- *
+ *
* @return true - if the data node is fully started
*/
public boolean isDatanodeFullyStarted() {
@@ -3105,12 +3104,12 @@ public class DataNode extends ReconfigurableBase
}
return true;
}
-
+
@VisibleForTesting
public DatanodeID getDatanodeId() {
return id;
}
-
+
@VisibleForTesting
public void clearAllBlockSecretKeys() {
blockPoolTokenSecretManager.clearAllKeysForTesting();
@@ -3122,7 +3121,7 @@ public class DataNode extends ReconfigurableBase
(DataXceiverServer) this.dataXceiverServer.getRunnable();
return dxcs.balanceThrottler.getBandwidth();
}
-
+
public DNConf getDnConf() {
return dnConf;
}
@@ -3143,70 +3142,55 @@ public class DataNode extends ReconfigurableBase
public ShortCircuitRegistry getShortCircuitRegistry() {
return shortCircuitRegistry;
}
-
+
/**
- * Check the disk error
+ * Check the disk error synchronously.
*/
- private void checkDiskError() {
- Set<File> unhealthyDataDirs = data.checkDataDir();
- if (unhealthyDataDirs != null && !unhealthyDataDirs.isEmpty()) {
+ @VisibleForTesting
+ public void checkDiskError() throws IOException {
+ Set<FsVolumeSpi> unhealthyVolumes;
+ try {
+ unhealthyVolumes = volumeChecker.checkAllVolumes(data);
+ lastDiskErrorCheck = Time.monotonicNow();
+ } catch (InterruptedException e) {
+ LOG.error("Interruped while running disk check", e);
+ throw new IOException("Interrupted while running disk check", e);
+ }
+
+ if (unhealthyVolumes.size() > 0) {
+ LOG.warn("checkDiskError got {} failed volumes - {}",
+ unhealthyVolumes.size(), unhealthyVolumes);
+ handleVolumeFailures(unhealthyVolumes);
+ } else {
+ LOG.debug("checkDiskError encountered no failures");
+ }
+ }
+
+ private void handleVolumeFailures(Set<FsVolumeSpi> unhealthyVolumes) {
+ data.handleVolumeFailures(unhealthyVolumes);
+ final Set<File> unhealthyDirs = new HashSet<>(unhealthyVolumes.size());
+
+ if (!unhealthyVolumes.isEmpty()) {
+ StringBuilder sb = new StringBuilder("DataNode failed volumes:");
+ for (FsVolumeSpi vol : unhealthyVolumes) {
+ unhealthyDirs.add(new File(vol.getBasePath()).getAbsoluteFile());
+ sb.append(vol).append(";");
+ }
+
try {
// Remove all unhealthy volumes from DataNode.
- removeVolumes(unhealthyDataDirs, false);
+ removeVolumes(unhealthyDirs, false);
} catch (IOException e) {
LOG.warn("Error occurred when removing unhealthy storage dirs: "
+ e.getMessage(), e);
}
- StringBuilder sb = new StringBuilder("DataNode failed volumes:");
- for (File dataDir : unhealthyDataDirs) {
- sb.append(dataDir.getAbsolutePath() + ";");
- }
+ LOG.info(sb.toString());
handleDiskError(sb.toString());
}
}
- /**
- * Starts a new thread which will check for disk error check request
- * every 5 sec
- */
- private void startCheckDiskErrorThread() {
- checkDiskErrorThread = new Thread(new Runnable() {
- @Override
- public void run() {
- while(shouldRun) {
- boolean tempFlag ;
- synchronized(checkDiskErrorMutex) {
- tempFlag = checkDiskErrorFlag;
- checkDiskErrorFlag = false;
- }
- if(tempFlag) {
- try {
- checkDiskError();
- } catch (Exception e) {
- LOG.warn("Unexpected exception occurred while checking disk error " + e);
- checkDiskErrorThread = null;
- return;
- }
- synchronized(checkDiskErrorMutex) {
- lastDiskErrorCheck = Time.monotonicNow();
- }
- }
- try {
- Thread.sleep(checkDiskErrorInterval);
- } catch (InterruptedException e) {
- LOG.debug("InterruptedException in check disk error thread", e);
- checkDiskErrorThread = null;
- return;
- }
- }
- }
- });
- }
-
public long getLastDiskErrorCheck() {
- synchronized(checkDiskErrorMutex) {
- return lastDiskErrorCheck;
- }
+ return lastDiskErrorCheck;
}
@Override
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ec80de3c/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/StorageLocation.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/StorageLocation.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/StorageLocation.java
index 7a91a82..3c7ba29 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/StorageLocation.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/StorageLocation.java
@@ -30,11 +30,6 @@ import org.apache.hadoop.fs.LocalFileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.fs.permission.FsPermission;
-import org.apache.hadoop.fs.StorageType;
-import org.apache.hadoop.hdfs.DFSConfigKeys;
-import org.apache.hadoop.hdfs.HdfsConfiguration;
-import org.apache.hadoop.hdfs.server.common.Storage.StorageDirectory;
-import org.apache.hadoop.hdfs.server.common.Storage;
import org.apache.hadoop.hdfs.server.datanode.checker.Checkable;
import org.apache.hadoop.hdfs.server.datanode.checker.VolumeCheckResult;
import org.apache.hadoop.util.DiskChecker;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ec80de3c/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/DatasetVolumeChecker.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/DatasetVolumeChecker.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/DatasetVolumeChecker.java
index e49ec7d..8f346dc 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/DatasetVolumeChecker.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/DatasetVolumeChecker.java
@@ -27,7 +27,6 @@ import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdfs.DFSConfigKeys;
-import org.apache.hadoop.hdfs.server.datanode.StorageLocation;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsDatasetSpi;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeReference;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi;
@@ -164,9 +163,21 @@ public class DatasetVolumeChecker {
public Set<FsVolumeSpi> checkAllVolumes(
final FsDatasetSpi<? extends FsVolumeSpi> dataset)
throws InterruptedException {
-
- if (timer.monotonicNow() - lastAllVolumesCheck < minDiskCheckGapMs) {
+ final long gap = timer.monotonicNow() - lastAllVolumesCheck;
+ if (gap < minDiskCheckGapMs) {
numSkippedChecks.incrementAndGet();
+ LOG.trace(
+ "Skipped checking all volumes, time since last check {} is less " +
+ "than the minimum gap between checks ({} ms).",
+ gap, minDiskCheckGapMs);
+ return Collections.emptySet();
+ }
+
+ final FsDatasetSpi.FsVolumeReferences references =
+ dataset.getFsVolumeReferences();
+
+ if (references.size() == 0) {
+ LOG.warn("checkAllVolumesAsync - no volumes can be referenced");
return Collections.emptySet();
}
@@ -175,9 +186,8 @@ public class DatasetVolumeChecker {
final Set<FsVolumeSpi> failedVolumes = new HashSet<>();
final Set<FsVolumeSpi> allVolumes = new HashSet<>();
- final FsDatasetSpi.FsVolumeReferences references =
- dataset.getFsVolumeReferences();
- final CountDownLatch resultsLatch = new CountDownLatch(references.size());
+ final AtomicLong numVolumes = new AtomicLong(references.size());
+ final CountDownLatch latch = new CountDownLatch(1);
for (int i = 0; i < references.size(); ++i) {
final FsVolumeReference reference = references.getReference(i);
@@ -186,12 +196,18 @@ public class DatasetVolumeChecker {
delegateChecker.schedule(reference.getVolume(), IGNORED_CONTEXT);
LOG.info("Scheduled health check for volume {}", reference.getVolume());
Futures.addCallback(future, new ResultHandler(
- reference, healthyVolumes, failedVolumes, resultsLatch, null));
+ reference, healthyVolumes, failedVolumes, numVolumes, new Callback() {
+ @Override
+ public void call(Set<FsVolumeSpi> ignored1,
+ Set<FsVolumeSpi> ignored2) {
+ latch.countDown();
+ }
+ }));
}
// Wait until our timeout elapses, after which we give up on
// the remaining volumes.
- if (!resultsLatch.await(maxAllowedTimeForCheckMs, TimeUnit.MILLISECONDS)) {
+ if (!latch.await(maxAllowedTimeForCheckMs, TimeUnit.MILLISECONDS)) {
LOG.warn("checkAllVolumes timed out after {} ms" +
maxAllowedTimeForCheckMs);
}
@@ -225,18 +241,28 @@ public class DatasetVolumeChecker {
public boolean checkAllVolumesAsync(
final FsDatasetSpi<? extends FsVolumeSpi> dataset,
Callback callback) {
-
- if (timer.monotonicNow() - lastAllVolumesCheck < minDiskCheckGapMs) {
+ final long gap = timer.monotonicNow() - lastAllVolumesCheck;
+ if (gap < minDiskCheckGapMs) {
numSkippedChecks.incrementAndGet();
+ LOG.trace(
+ "Skipped checking all volumes, time since last check {} is less " +
+ "than the minimum gap between checks ({} ms).",
+ gap, minDiskCheckGapMs);
+ return false;
+ }
+
+ final FsDatasetSpi.FsVolumeReferences references =
+ dataset.getFsVolumeReferences();
+
+ if (references.size() == 0) {
+ LOG.warn("checkAllVolumesAsync - no volumes can be referenced");
return false;
}
lastAllVolumesCheck = timer.monotonicNow();
final Set<FsVolumeSpi> healthyVolumes = new HashSet<>();
final Set<FsVolumeSpi> failedVolumes = new HashSet<>();
- final FsDatasetSpi.FsVolumeReferences references =
- dataset.getFsVolumeReferences();
- final CountDownLatch latch = new CountDownLatch(references.size());
+ final AtomicLong numVolumes = new AtomicLong(references.size());
LOG.info("Checking {} volumes", references.size());
for (int i = 0; i < references.size(); ++i) {
@@ -245,7 +271,7 @@ public class DatasetVolumeChecker {
ListenableFuture<VolumeCheckResult> future =
delegateChecker.schedule(reference.getVolume(), IGNORED_CONTEXT);
Futures.addCallback(future, new ResultHandler(
- reference, healthyVolumes, failedVolumes, latch, callback));
+ reference, healthyVolumes, failedVolumes, numVolumes, callback));
}
numAsyncDatasetChecks.incrementAndGet();
return true;
@@ -273,8 +299,10 @@ public class DatasetVolumeChecker {
*
* @param volume the volume that is to be checked.
* @param callback callback to be invoked when the volume check completes.
+ * @return true if the check was scheduled and the callback will be invoked.
+ * false otherwise.
*/
- public void checkVolume(
+ public boolean checkVolume(
final FsVolumeSpi volume,
Callback callback) {
FsVolumeReference volumeReference;
@@ -283,14 +311,15 @@ public class DatasetVolumeChecker {
} catch (ClosedChannelException e) {
// The volume has already been closed.
callback.call(new HashSet<FsVolumeSpi>(), new HashSet<FsVolumeSpi>());
- return;
+ return false;
}
ListenableFuture<VolumeCheckResult> future =
delegateChecker.schedule(volume, IGNORED_CONTEXT);
numVolumeChecks.incrementAndGet();
Futures.addCallback(future, new ResultHandler(
- volumeReference, new HashSet<FsVolumeSpi>(),
- new HashSet<FsVolumeSpi>(), new CountDownLatch(1), callback));
+ volumeReference, new HashSet<FsVolumeSpi>(), new HashSet<FsVolumeSpi>(),
+ new AtomicLong(1), callback));
+ return true;
}
/**
@@ -301,24 +330,33 @@ public class DatasetVolumeChecker {
private final FsVolumeReference reference;
private final Set<FsVolumeSpi> failedVolumes;
private final Set<FsVolumeSpi> healthyVolumes;
- private final CountDownLatch latch;
- private final AtomicLong numVolumes;
+ private final AtomicLong volumeCounter;
@Nullable
private final Callback callback;
+ /**
+ *
+ * @param reference FsVolumeReference to be released when the check is
+ * complete.
+ * @param healthyVolumes set of healthy volumes. If the disk check is
+ * successful, add the volume here.
+ * @param failedVolumes set of failed volumes. If the disk check fails,
+ * add the volume here.
+ * @param semaphore semaphore used to trigger callback invocation.
+ * @param callback invoked when the semaphore can be successfully acquired.
+ */
ResultHandler(FsVolumeReference reference,
Set<FsVolumeSpi> healthyVolumes,
Set<FsVolumeSpi> failedVolumes,
- CountDownLatch latch,
+ AtomicLong volumeCounter,
@Nullable Callback callback) {
Preconditions.checkState(reference != null);
this.reference = reference;
this.healthyVolumes = healthyVolumes;
this.failedVolumes = failedVolumes;
- this.latch = latch;
+ this.volumeCounter = volumeCounter;
this.callback = callback;
- numVolumes = new AtomicLong(latch.getCount());
}
@Override
@@ -372,10 +410,8 @@ public class DatasetVolumeChecker {
private void invokeCallback() {
try {
- latch.countDown();
-
- if (numVolumes.decrementAndGet() == 0 &&
- callback != null) {
+ final long remaining = volumeCounter.decrementAndGet();
+ if (callback != null && remaining == 0) {
callback.call(healthyVolumes, failedVolumes);
}
} catch(Exception e) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ec80de3c/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 8356331..4ce603f 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
@@ -65,7 +65,7 @@ import org.apache.hadoop.util.ReflectionUtils;
/**
* This is a service provider interface for the underlying storage that
* stores replicas for a data node.
- * The default implementation stores replicas on local drives.
+ * The default implementation stores replicas on local drives.
*/
@InterfaceAudience.Private
public interface FsDatasetSpi<V extends FsVolumeSpi> extends FSDatasetMBean {
@@ -273,7 +273,7 @@ public interface FsDatasetSpi<V extends FsVolumeSpi> extends FSDatasetMBean {
long getLength(ExtendedBlock b) throws IOException;
/**
- * Get reference to the replica meta info in the replicasMap.
+ * Get reference to the replica meta info in the replicasMap.
* To be called from methods that are synchronized on {@link FSDataset}
* @return replica from the replicas map
*/
@@ -314,7 +314,7 @@ public interface FsDatasetSpi<V extends FsVolumeSpi> extends FSDatasetMBean {
/**
* Creates a temporary replica and returns the meta information of the replica
* .
- *
+ *
* @param b block
* @return the meta info of the replica which is being written to
* @throws IOException if an error occurs
@@ -324,7 +324,7 @@ public interface FsDatasetSpi<V extends FsVolumeSpi> extends FSDatasetMBean {
/**
* Creates a RBW replica and returns the meta info of the replica
- *
+ *
* @param b block
* @return the meta info of the replica which is being written to
* @throws IOException if an error occurs
@@ -334,7 +334,7 @@ public interface FsDatasetSpi<V extends FsVolumeSpi> extends FSDatasetMBean {
/**
* Recovers a RBW replica and returns the meta info of the replica.
- *
+ *
* @param b block
* @param newGS the new generation stamp for the replica
* @param minBytesRcvd the minimum number of bytes that the replica could have
@@ -355,7 +355,7 @@ public interface FsDatasetSpi<V extends FsVolumeSpi> extends FSDatasetMBean {
/**
* Append to a finalized replica and returns the meta info of the replica.
- *
+ *
* @param b block
* @param newGS the new generation stamp for the replica
* @param expectedBlockLen the number of bytes the replica is expected to have
@@ -368,7 +368,7 @@ public interface FsDatasetSpi<V extends FsVolumeSpi> extends FSDatasetMBean {
/**
* Recover a failed append to a finalized replica and returns the meta
* info of the replica.
- *
+ *
* @param b block
* @param newGS the new generation stamp for the replica
* @param expectedBlockLen the number of bytes the replica is expected to have
@@ -377,11 +377,11 @@ public interface FsDatasetSpi<V extends FsVolumeSpi> extends FSDatasetMBean {
*/
ReplicaHandler recoverAppend(
ExtendedBlock b, long newGS, long expectedBlockLen) throws IOException;
-
+
/**
* Recover a failed pipeline close.
* It bumps the replica's generation stamp and finalize it if RBW replica
- *
+ *
* @param b block
* @param newGS the new generation stamp for the replica
* @param expectedBlockLen the number of bytes the replica is expected to have
@@ -390,7 +390,7 @@ public interface FsDatasetSpi<V extends FsVolumeSpi> extends FSDatasetMBean {
*/
Replica recoverClose(ExtendedBlock b, long newGS, long expectedBlockLen
) throws IOException;
-
+
/**
* 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
@@ -437,19 +437,19 @@ public interface FsDatasetSpi<V extends FsVolumeSpi> extends FSDatasetMBean {
*
* @throws ReplicaNotFoundException If the replica is not found
*
- * @throws UnexpectedReplicaStateException If the replica is not in the
+ * @throws UnexpectedReplicaStateException If the replica is not in the
* expected state.
- * @throws FileNotFoundException If the block file is not found or there
+ * @throws FileNotFoundException If the block file is not found or there
* was an error locating it.
* @throws EOFException If the replica length is too short.
- *
- * @throws IOException May be thrown from the methods called.
+ *
+ * @throws IOException May be thrown from the methods called.
*/
void checkBlock(ExtendedBlock b, long minLength, ReplicaState state)
throws ReplicaNotFoundException, UnexpectedReplicaStateException,
FileNotFoundException, EOFException, IOException;
-
-
+
+
/**
* Is the block valid?
* @return - true if the specified block is valid
@@ -493,10 +493,9 @@ public interface FsDatasetSpi<V extends FsVolumeSpi> extends FSDatasetMBean {
boolean isCached(String bpid, long blockId);
/**
- * Check if all the data directories are healthy
- * @return A set of unhealthy data directories.
+ * Handle volume failures by removing the failed volumes.
*/
- Set<File> checkDataDir();
+ void handleVolumeFailures(Set<FsVolumeSpi> failedVolumes);
/**
* Shutdown the FSDataset
@@ -516,7 +515,7 @@ public interface FsDatasetSpi<V extends FsVolumeSpi> extends FSDatasetMBean {
/**
* Checks how many valid storage volumes there are in the DataNode.
- * @return true if more than the minimum number of valid volumes are left
+ * @return true if more than the minimum number of valid volumes are left
* in the FSDataSet.
*/
boolean hasEnoughResource();
@@ -528,7 +527,7 @@ public interface FsDatasetSpi<V extends FsVolumeSpi> extends FSDatasetMBean {
/**
* Initialize a replica recovery.
- * @return actual state of the replica on this data-node or
+ * @return actual state of the replica on this data-node or
* null if data-node does not have the replica.
*/
ReplicaRecoveryInfo initReplicaRecovery(RecoveringBlock rBlock
@@ -555,13 +554,13 @@ public interface FsDatasetSpi<V extends FsVolumeSpi> extends FSDatasetMBean {
void shutdownBlockPool(String bpid) ;
/**
- * Deletes the block pool directories. If force is false, directories are
- * deleted only if no block files exist for the block pool. If force
+ * Deletes the block pool directories. If force is false, directories are
+ * deleted only if no block files exist for the block pool. If force
* is true entire directory for the blockpool is deleted along with its
* contents.
* @param bpid BlockPool Id to be deleted.
* @param force If force is false, directories are deleted only if no
- * block files exist for the block pool, otherwise entire
+ * block files exist for the block pool, otherwise entire
* directory for the blockpool is deleted along with its contents.
* @throws IOException
*/
@@ -574,9 +573,9 @@ public interface FsDatasetSpi<V extends FsVolumeSpi> extends FSDatasetMBean {
) throws IOException;
/**
- * Get a {@link HdfsBlocksMetadata} corresponding to the list of blocks in
+ * Get a {@link HdfsBlocksMetadata} corresponding to the list of blocks in
* <code>blocks</code>.
- *
+ *
* @param bpid pool to query
* @param blockIds List of block ids for which to return metadata
* @return metadata Metadata for the list of blocks
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ec80de3c/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 c821989..e37934e 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
@@ -231,7 +231,7 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
}
return null;
}
-
+
@Override // FsDatasetSpi
public LengthInputStream getMetaDataInputStream(ExtendedBlock b)
throws IOException {
@@ -291,7 +291,7 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
@VisibleForTesting
final AutoCloseableLock datasetLock;
private final Condition datasetLockCondition;
-
+
/**
* An FSDataset has a directory where it loads its data files.
*/
@@ -324,8 +324,8 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
if (volsFailed > volFailuresTolerated) {
throw new DiskErrorException("Too many failed volumes - "
- + "current valid volumes: " + storage.getNumStorageDirs()
- + ", volumes configured: " + volsConfigured
+ + "current valid volumes: " + storage.getNumStorageDirs()
+ + ", volumes configured: " + volsConfigured
+ ", volumes failed: " + volsFailed
+ ", volume failures tolerated: " + volFailuresTolerated);
}
@@ -614,9 +614,9 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
return volumes.getBlockPoolUsed(bpid);
}
}
-
+
/**
- * Return true - if there are still valid volumes on the DataNode.
+ * Return true - if there are still valid volumes on the DataNode.
*/
@Override // FsDatasetSpi
public boolean hasEnoughResource() {
@@ -762,7 +762,7 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
private File getBlockFile(ExtendedBlock b) throws IOException {
return getBlockFile(b.getBlockPoolId(), b.getBlockId());
}
-
+
/**
* Get File name for a given block.
*/
@@ -824,7 +824,7 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
* block pool Id, block Id and generation stamp must match.
* @param b extended block
* @return the meta replica information
- * @throws ReplicaNotFoundException if no entry is in the map or
+ * @throws ReplicaNotFoundException if no entry is in the map or
* there is a generation stamp mismatch
*/
ReplicaInfo getReplicaInfo(ExtendedBlock b)
@@ -849,7 +849,7 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
* @param bpid block pool Id
* @param blkid block Id
* @return the meta replica information; null if block was not found
- * @throws ReplicaNotFoundException if no entry is in the map or
+ * @throws ReplicaNotFoundException if no entry is in the map or
* there is a generation stamp mismatch
*/
private ReplicaInfo getReplicaInfo(String bpid, long blkid)
@@ -1190,15 +1190,15 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
}
/** Append to a finalized replica
- * Change a finalized replica to be a RBW replica and
+ * Change a finalized replica to be a RBW replica and
* bump its generation stamp to be the newGS
- *
+ *
* @param bpid block pool Id
* @param replicaInfo a finalized replica
* @param newGS new generation stamp
* @param estimateBlockLen estimate generation stamp
* @return a RBW replica
- * @throws IOException if moving the replica from finalized directory
+ * @throws IOException if moving the replica from finalized directory
* to rbw directory fails
*/
private ReplicaBeingWritten append(String bpid,
@@ -1287,10 +1287,10 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
}
}
- private ReplicaInfo recoverCheck(ExtendedBlock b, long newGS,
+ private ReplicaInfo recoverCheck(ExtendedBlock b, long newGS,
long expectedBlockLen) throws IOException, MustStopExistingWriter {
ReplicaInfo replicaInfo = getReplicaInfo(b.getBlockPoolId(), b.getBlockId());
-
+
// check state
if (replicaInfo.getState() != ReplicaState.FINALIZED &&
replicaInfo.getState() != ReplicaState.RBW) {
@@ -1304,10 +1304,10 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
replicaGenerationStamp > newGS) {
throw new ReplicaNotFoundException(
ReplicaNotFoundException.UNEXPECTED_GS_REPLICA + replicaGenerationStamp
- + ". Expected GS range is [" + b.getGenerationStamp() + ", " +
+ + ". Expected GS range is [" + b.getGenerationStamp() + ", " +
newGS + "].");
}
-
+
// stop the previous writer before check a replica's length
long replicaLen = replicaInfo.getNumBytes();
if (replicaInfo.getState() == ReplicaState.RBW) {
@@ -1316,22 +1316,22 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
throw new MustStopExistingWriter(rbw);
}
// check length: bytesRcvd, bytesOnDisk, and bytesAcked should be the same
- if (replicaLen != rbw.getBytesOnDisk()
+ if (replicaLen != rbw.getBytesOnDisk()
|| replicaLen != rbw.getBytesAcked()) {
- throw new ReplicaAlreadyExistsException("RBW replica " + replicaInfo +
- "bytesRcvd(" + rbw.getNumBytes() + "), bytesOnDisk(" +
+ throw new ReplicaAlreadyExistsException("RBW replica " + replicaInfo +
+ "bytesRcvd(" + rbw.getNumBytes() + "), bytesOnDisk(" +
rbw.getBytesOnDisk() + "), and bytesAcked(" + rbw.getBytesAcked() +
") are not the same.");
}
}
-
+
// check block length
if (replicaLen != expectedBlockLen) {
- throw new IOException("Corrupted replica " + replicaInfo +
- " with a length of " + replicaLen +
+ throw new IOException("Corrupted replica " + replicaInfo +
+ " with a length of " + replicaLen +
" expected length is " + expectedBlockLen);
}
-
+
return replicaInfo;
}
@@ -1390,7 +1390,7 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
}
}
}
-
+
/**
* Bump a replica's generation stamp to a new one.
* Its on-disk meta file name is renamed to be the new one too.
@@ -1559,7 +1559,7 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
return new ReplicaHandler(rbw, ref);
}
}
-
+
@Override // FsDatasetSpi
public ReplicaInPipeline convertTemporaryToRbw(
final ExtendedBlock b) throws IOException {
@@ -1673,7 +1673,7 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
// Hang too long, just bail out. This is not supposed to happen.
long writerStopMs = Time.monotonicNow() - startTimeMs;
if (writerStopMs > writerStopTimeoutMs) {
- LOG.warn("Unable to stop existing writer for block " + b + " after "
+ LOG.warn("Unable to stop existing writer for block " + b + " after "
+ writerStopMs + " miniseconds.");
throw new IOException("Unable to stop existing writer for block " + b
+ " after " + writerStopMs + " miniseconds.");
@@ -1690,7 +1690,7 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
* last checksum will be overwritten.
*/
@Override // FsDatasetSpi
- public void adjustCrcChannelPosition(ExtendedBlock b, ReplicaOutputStreams streams,
+ public void adjustCrcChannelPosition(ExtendedBlock b, ReplicaOutputStreams streams,
int checksumSize) throws IOException {
FileOutputStream file = (FileOutputStream)streams.getChecksumOut();
FileChannel channel = file.getChannel();
@@ -1807,7 +1807,7 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
LOG.warn("No file exists for block: " + b);
return true;
}
-
+
if (!blockFile.delete()) {
LOG.warn("Not able to delete the block file: " + blockFile);
return false;
@@ -1910,21 +1910,21 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
* @param state If this is null, it is ignored. If it is non-null, we
* will check that the replica has this state.
*
- * @throws ReplicaNotFoundException If the replica is not found
+ * @throws ReplicaNotFoundException If the replica is not found
*
- * @throws UnexpectedReplicaStateException If the replica is not in the
+ * @throws UnexpectedReplicaStateException If the replica is not in the
* expected state.
* @throws FileNotFoundException If the block file is not found or there
* was an error locating it.
* @throws EOFException If the replica length is too short.
- *
- * @throws IOException May be thrown from the methods called.
+ *
+ * @throws IOException May be thrown from the methods called.
*/
@Override // FsDatasetSpi
public void checkBlock(ExtendedBlock b, long minLength, ReplicaState state)
throws ReplicaNotFoundException, UnexpectedReplicaStateException,
FileNotFoundException, EOFException, IOException {
- final ReplicaInfo replicaInfo = volumeMap.get(b.getBlockPoolId(),
+ final ReplicaInfo replicaInfo = volumeMap.get(b.getBlockPoolId(),
b.getLocalBlock());
if (replicaInfo == null) {
throw new ReplicaNotFoundException(b);
@@ -1950,7 +1950,7 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
public boolean isValidBlock(ExtendedBlock b) {
return isValid(b, ReplicaState.FINALIZED);
}
-
+
/**
* Check whether the given block is a valid RBW.
*/
@@ -2182,7 +2182,7 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
genstamp = info.getGenerationStamp();
volumeExecutor = volume.getCacheExecutor();
}
- cacheManager.cacheBlock(blockId, bpid,
+ cacheManager.cacheBlock(blockId, bpid,
blockFileName, length, genstamp, volumeExecutor);
}
@@ -2237,12 +2237,13 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
* if some volumes failed - the caller must emove all the blocks that belong
* to these failed volumes.
* @return the failed volumes. Returns null if no volume failed.
+ * @param failedVolumes
*/
@Override // FsDatasetSpi
- public Set<File> checkDataDir() {
- return volumes.checkDirs();
+ public void handleVolumeFailures(Set<FsVolumeSpi> failedVolumes) {
+ volumes.handleVolumeFailures(failedVolumes);
}
-
+
@Override // FsDatasetSpi
public String toString() {
@@ -2250,14 +2251,14 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
}
private ObjectName mbeanName;
-
+
/**
* Register the FSDataset MBean using the name
* "hadoop:service=DataNode,name=FSDatasetState-<datanodeUuid>"
*/
void registerMBean(final String datanodeUuid) {
// We wrap to bypass standard mbean naming convetion.
- // This wraping can be removed in java 6 as it is more flexible in
+ // This wraping can be removed in java 6 as it is more flexible in
// package naming for mbeans and their impl.
try {
StandardMBean bean = new StandardMBean(this,FSDatasetMBean.class);
@@ -2280,7 +2281,7 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
if (mbeanName != null) {
MBeans.unregister(mbeanName);
}
-
+
if (asyncDiskService != null) {
asyncDiskService.shutdown();
}
@@ -2288,7 +2289,7 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
if (asyncLazyPersistService != null) {
asyncLazyPersistService.shutdown();
}
-
+
if(volumes != null) {
volumes.shutdown();
}
@@ -2497,7 +2498,7 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
return volumeMap.get(bpid, blockId);
}
- @Override
+ @Override
public String getReplicaString(String bpid, long blockId) {
try(AutoCloseableLock lock = datasetLock.acquire()) {
final Replica r = volumeMap.get(bpid, blockId);
@@ -2753,7 +2754,7 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
return replica.getVisibleLength();
}
}
-
+
@Override
public void addBlockPool(String bpid, Configuration conf)
throws IOException {
@@ -2775,7 +2776,7 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
volumes.removeBlockPool(bpid, blocksPerVolume);
}
}
-
+
/**
* Class for representing the Datanode volume information
*/
@@ -2796,7 +2797,7 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
this.reservedSpaceForReplicas = v.getReservedForReplicas();
this.numBlocks = v.getNumBlocks();
}
- }
+ }
private Collection<VolumeInfo> getVolumeInfo() {
Collection<VolumeInfo> info = new ArrayList<VolumeInfo>();
@@ -2813,7 +2814,7 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
used = 0;
free = 0;
}
-
+
info.add(new VolumeInfo(volume, used, free));
}
return info;
@@ -2863,7 +2864,7 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
}
}
}
-
+
@Override // FsDatasetSpi
public BlockLocalPathInfo getBlockLocalPathInfo(ExtendedBlock block)
throws IOException {
@@ -3237,7 +3238,7 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
shouldRun = false;
}
}
-
+
@Override
public void setPinning(ExtendedBlock block) throws IOException {
if (!blockPinningEnabled) {
@@ -3254,7 +3255,7 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
oldPermission.getGroupAction(), oldPermission.getOtherAction(), true);
localFS.setPermission(p, permission);
}
-
+
@Override
public boolean getPinning(ExtendedBlock block) throws IOException {
if (!blockPinningEnabled) {
@@ -3265,7 +3266,7 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
FileStatus fss = localFS.getFileStatus(new Path(f.getAbsolutePath()));
return fss.getPermission().getStickyBit();
}
-
+
@Override
public boolean isDeletingBlock(String bpid, long blockId) {
synchronized(deletingBlock) {
@@ -3289,7 +3290,7 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
}
}
}
-
+
private void addDeletingBlock(String bpid, Long blockId) {
synchronized(deletingBlock) {
Set<Long> s = deletingBlock.get(bpid);
@@ -3361,4 +3362,3 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
}
}
}
-
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ec80de3c/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeImpl.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeImpl.java
index bc822c6..b1ce931 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeImpl.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeImpl.java
@@ -901,13 +901,6 @@ public class FsVolumeImpl implements FsVolumeSpi {
return cacheExecutor;
}
- void checkDirs() throws DiskErrorException {
- // TODO:FEDERATION valid synchronization
- for(BlockPoolSlice s : bpSlices.values()) {
- s.checkDirs();
- }
- }
-
@Override
public VolumeCheckResult check(VolumeCheckContext ignored)
throws DiskErrorException {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ec80de3c/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeList.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeList.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeList.java
index 9ecb901..8f84bf2 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeList.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeList.java
@@ -24,7 +24,6 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -44,7 +43,6 @@ import org.apache.hadoop.hdfs.server.datanode.BlockScanner;
import org.apache.hadoop.hdfs.server.protocol.DatanodeStorage;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.AutoCloseableLock;
-import org.apache.hadoop.util.DiskChecker.DiskErrorException;
import org.apache.hadoop.util.Time;
class FsVolumeList {
@@ -96,11 +94,11 @@ class FsVolumeList {
}
}
- /**
+ /**
* Get next volume.
*
* @param blockSize free space needed on the volume
- * @param storageType the desired {@link StorageType}
+ * @param storageType the desired {@link StorageType}
* @return next volume to store the block in.
*/
FsVolumeReference getNextVolume(StorageType storageType, long blockSize)
@@ -167,7 +165,7 @@ class FsVolumeList {
}
return capacity;
}
-
+
long getRemaining() throws IOException {
long remaining = 0L;
for (FsVolumeSpi vol : volumes) {
@@ -179,7 +177,7 @@ class FsVolumeList {
}
return remaining;
}
-
+
void getAllVolumesMap(final String bpid,
final ReplicaMap volumeMap,
final RamDiskReplicaTracker ramDiskReplicaMap)
@@ -229,28 +227,17 @@ class FsVolumeList {
/**
* Calls {@link FsVolumeImpl#checkDirs()} on each volume.
- *
+ *
* Use {@link checkDirsLock} to allow only one instance of checkDirs() call.
*
* @return list of all the failed volumes.
+ * @param failedVolumes
*/
- Set<File> checkDirs() {
+ void handleVolumeFailures(Set<FsVolumeSpi> failedVolumes) {
try (AutoCloseableLock lock = checkDirsLock.acquire()) {
- Set<File> failedVols = null;
-
- // Make a copy of volumes for performing modification
- final List<FsVolumeImpl> volumeList = getVolumes();
-
- for(Iterator<FsVolumeImpl> i = volumeList.iterator(); i.hasNext(); ) {
- final FsVolumeImpl fsv = i.next();
+ for(FsVolumeSpi vol : failedVolumes) {
+ FsVolumeImpl fsv = (FsVolumeImpl) vol;
try (FsVolumeReference ref = fsv.obtainReference()) {
- fsv.checkDirs();
- } catch (DiskErrorException e) {
- FsDatasetImpl.LOG.warn("Removing failed volume " + fsv + ": ", e);
- if (failedVols == null) {
- failedVols = new HashSet<>(1);
- }
- failedVols.add(new File(fsv.getBasePath()).getAbsoluteFile());
addVolumeFailureInfo(fsv);
removeVolume(fsv);
} catch (ClosedChannelException e) {
@@ -260,14 +247,8 @@ class FsVolumeList {
FsDatasetImpl.LOG.error("Unexpected IOException", e);
}
}
-
- if (failedVols != null && failedVols.size() > 0) {
- FsDatasetImpl.LOG.warn("Completed checkDirs. Found " + failedVols.size()
- + " failure volumes.");
- }
waitVolumeRemoved(5000, checkDirsLockCondition);
- return failedVols;
}
}
@@ -405,7 +386,7 @@ class FsVolumeList {
void addBlockPool(final String bpid, final Configuration conf) throws IOException {
long totalStartTime = Time.monotonicNow();
-
+
final List<IOException> exceptions = Collections.synchronizedList(
new ArrayList<IOException>());
List<Thread> blockPoolAddingThreads = new ArrayList<Thread>();
@@ -442,12 +423,12 @@ class FsVolumeList {
if (!exceptions.isEmpty()) {
throw exceptions.get(0);
}
-
+
long totalTimeTaken = Time.monotonicNow() - totalStartTime;
FsDatasetImpl.LOG.info("Total time to scan all replicas for block pool " +
bpid + ": " + totalTimeTaken + "ms");
}
-
+
void removeBlockPool(String bpid, Map<DatanodeStorage, BlockListAsLongs>
blocksPerVolume) {
for (FsVolumeImpl v : volumes) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ec80de3c/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/TestBlockStatsMXBean.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/TestBlockStatsMXBean.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/TestBlockStatsMXBean.java
index 9e3112e..570776e 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/TestBlockStatsMXBean.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/TestBlockStatsMXBean.java
@@ -30,9 +30,11 @@ import java.net.URL;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.TimeUnit;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.StorageType;
+import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.DFSTestUtil;
import org.apache.hadoop.hdfs.HdfsConfiguration;
import org.apache.hadoop.hdfs.MiniDFSCluster;
@@ -52,6 +54,8 @@ public class TestBlockStatsMXBean {
@Before
public void setup() throws IOException {
HdfsConfiguration conf = new HdfsConfiguration();
+ conf.setTimeDuration(DFSConfigKeys.DFS_DATANODE_DISK_CHECK_MIN_GAP_KEY,
+ 0, TimeUnit.MILLISECONDS);
cluster = null;
StorageType[][] types = new StorageType[6][];
for (int i=0; i<3; i++) {
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[2/5] hadoop git commit: HDFS-11182. Update DataNode to use
DatasetVolumeChecker.
Posted by ar...@apache.org.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ec80de3c/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 dff8482..db5042b 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
@@ -68,18 +68,18 @@ import org.apache.hadoop.util.DataChecksum;
/**
* This class implements a simulated FSDataset.
- *
+ *
* Blocks that are created are recorded but their data (plus their CRCs) are
* discarded.
* Fixed data is returned when blocks are read; a null CRC meta file is
* created for such data.
- *
+ *
* This FSDataset does not remember any block information across its
* restarts; it does however offer an operation to inject blocks
* (See the TestInectionForSImulatedStorage()
* for a usage example of injection.
- *
- * Note the synchronization is coarse grained - it is at each method.
+ *
+ * Note the synchronization is coarse grained - it is at each method.
*/
public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
public final static int BYTE_MASK = 0xff;
@@ -95,7 +95,7 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
return true;
}
}
-
+
public static void setFactory(Configuration conf) {
conf.set(DFSConfigKeys.DFS_DATANODE_FSDATASET_FACTORY_KEY,
Factory.class.getName());
@@ -105,12 +105,12 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
byte firstByte = (byte) (b.getBlockId() & BYTE_MASK);
return (byte) ((firstByte + offsetInBlk) & BYTE_MASK);
}
-
+
public static final String CONFIG_PROPERTY_CAPACITY =
"dfs.datanode.simulateddatastorage.capacity";
-
+
public static final long DEFAULT_CAPACITY = 2L<<40; // 1 terabyte
-
+
public static final String CONFIG_PROPERTY_STATE =
"dfs.datanode.simulateddatastorage.state";
private static final DatanodeStorage.State DEFAULT_STATE =
@@ -145,7 +145,7 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
if (theBlock.getNumBytes() < 0) {
theBlock.setNumBytes(0);
}
- if (!storage.alloc(bpid, theBlock.getNumBytes())) {
+ if (!storage.alloc(bpid, theBlock.getNumBytes())) {
// expected length - actual length may
// be more - we find out at finalize
DataNode.LOG.warn("Lack of free storage on a block alloc");
@@ -160,7 +160,7 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
oStream = null;
}
}
-
+
@Override
public String getStorageUuid() {
return storage.getStorageUuid();
@@ -188,7 +188,7 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
theBlock.setNumBytes(length);
}
}
-
+
synchronized SimulatedInputStream getIStream() {
if (!finalized) {
// throw new IOException("Trying to read an unfinalized block");
@@ -197,12 +197,12 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
return new SimulatedInputStream(theBlock.getNumBytes(), theBlock);
}
}
-
+
synchronized void finalizeBlock(String bpid, long finalSize)
throws IOException {
if (finalized) {
throw new IOException(
- "Finalizing a block that has already been finalized" +
+ "Finalizing a block that has already been finalized" +
theBlock.getBlockId());
}
if (oStream == null) {
@@ -216,7 +216,7 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
throw new IOException(
"Size passed to finalize does not match the amount of data written");
}
- // We had allocated the expected length when block was created;
+ // We had allocated the expected length when block was created;
// adjust if necessary
long extraLen = finalSize - theBlock.getNumBytes();
if (extraLen > 0) {
@@ -227,7 +227,7 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
} else {
storage.free(bpid, -extraLen);
}
- theBlock.setNumBytes(finalSize);
+ theBlock.setNumBytes(finalSize);
finalized = true;
oStream = null;
@@ -248,7 +248,7 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
}
SimulatedInputStream getMetaIStream() {
- return new SimulatedInputStream(nullCrcFileData);
+ return new SimulatedInputStream(nullCrcFileData);
}
synchronized boolean isFinalized() {
@@ -257,8 +257,7 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
@Override
synchronized public ReplicaOutputStreams createStreams(boolean isCreate,
- DataChecksum requestedChecksum)
- throws IOException {
+ DataChecksum requestedChecksum) throws IOException {
if (finalized) {
throw new IOException("Trying to write to a finalized replica "
+ theBlock);
@@ -333,31 +332,31 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
return false;
}
}
-
+
/**
* Class is used for tracking block pool storage utilization similar
* to {@link BlockPoolSlice}
*/
private static class SimulatedBPStorage {
private long used; // in bytes
-
+
long getUsed() {
return used;
}
-
+
void alloc(long amount) {
used += amount;
}
-
+
void free(long amount) {
used -= amount;
}
-
+
SimulatedBPStorage() {
- used = 0;
+ used = 0;
}
}
-
+
/**
* Class used for tracking datanode level storage utilization similar
* to {@link FSVolumeSet}
@@ -368,15 +367,15 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
private final long capacity; // in bytes
private final DatanodeStorage dnStorage;
-
+
synchronized long getFree() {
return capacity - getUsed();
}
-
+
long getCapacity() {
return capacity;
}
-
+
synchronized long getUsed() {
long used = 0;
for (SimulatedBPStorage bpStorage : map.values()) {
@@ -384,11 +383,11 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
}
return used;
}
-
+
synchronized long getBlockPoolUsed(String bpid) throws IOException {
return getBPStorage(bpid).getUsed();
}
-
+
int getNumFailedVolumes() {
return 0;
}
@@ -398,20 +397,20 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
getBPStorage(bpid).alloc(amount);
return true;
}
- return false;
+ return false;
}
-
+
synchronized void free(String bpid, long amount) throws IOException {
getBPStorage(bpid).free(amount);
}
-
+
SimulatedStorage(long cap, DatanodeStorage.State state) {
capacity = cap;
dnStorage = new DatanodeStorage(
"SimulatedStorage-" + DatanodeStorage.generateUuid(),
state, StorageType.DEFAULT);
}
-
+
synchronized void addBlockPool(String bpid) {
SimulatedBPStorage bpStorage = map.get(bpid);
if (bpStorage != null) {
@@ -419,11 +418,11 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
}
map.put(bpid, new SimulatedBPStorage());
}
-
+
synchronized void removeBlockPool(String bpid) {
map.remove(bpid);
}
-
+
private SimulatedBPStorage getBPStorage(String bpid) throws IOException {
SimulatedBPStorage bpStorage = map.get(bpid);
if (bpStorage == null) {
@@ -435,7 +434,7 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
String getStorageUuid() {
return dnStorage.getStorageID();
}
-
+
DatanodeStorage getDnStorage() {
return dnStorage;
}
@@ -462,7 +461,17 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
@Override
public FsVolumeReference obtainReference() throws ClosedChannelException {
- return null;
+ return new FsVolumeReference() {
+ @Override
+ public void close() throws IOException {
+ // no-op.
+ }
+
+ @Override
+ public FsVolumeSpi getVolume() {
+ return SimulatedVolume.this;
+ }
+ };
}
@Override
@@ -562,7 +571,7 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
private final SimulatedVolume volume;
private final String datanodeUuid;
private final DataNode datanode;
-
+
public SimulatedFSDataset(DataStorage storage, Configuration conf) {
this(null, storage, conf);
@@ -611,14 +620,14 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
map = new HashMap<Block, BInfo>();
blockMap.put(bpid, map);
}
-
+
for (Block b: injectBlocks) {
BInfo binfo = new BInfo(bpid, b, false);
map.put(binfo.theBlock, binfo);
}
}
}
-
+
/** Get a map for a given block pool Id */
private Map<Block, BInfo> getMap(String bpid) throws IOException {
final Map<Block, BInfo> map = blockMap.get(bpid);
@@ -684,7 +693,7 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
public long getBlockPoolUsed(String bpid) throws IOException {
return storage.getBlockPoolUsed(bpid);
}
-
+
@Override // FSDatasetMBean
public long getRemaining() {
return storage.getFree();
@@ -775,7 +784,7 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
return null;
}
- @Override
+ @Override
public synchronized String getReplicaString(String bpid, long blockId) {
Replica r = null;
final Map<Block, BInfo> map = blockMap.get(bpid);
@@ -865,14 +874,14 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
*
* @throws ReplicaNotFoundException If the replica is not found
*
- * @throws UnexpectedReplicaStateException If the replica is not in the
+ * @throws UnexpectedReplicaStateException If the replica is not in the
* expected state.
*/
@Override // {@link FsDatasetSpi}
public void checkBlock(ExtendedBlock b, long minLength, ReplicaState state)
throws ReplicaNotFoundException, UnexpectedReplicaStateException {
final BInfo binfo = getBInfo(b);
-
+
if (binfo == null) {
throw new ReplicaNotFoundException(b);
}
@@ -956,7 +965,7 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
map.put(binfo.theBlock, binfo);
return binfo;
}
-
+
@Override // FsDatasetSpi
public synchronized ReplicaHandler recoverRbw(
ExtendedBlock b, long newGS, long minBytesRcvd, long maxBytesRcvd)
@@ -988,11 +997,11 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
public synchronized ReplicaHandler createTemporary(
StorageType storageType, ExtendedBlock b) throws IOException {
if (isValidBlock(b)) {
- throw new ReplicaAlreadyExistsException("Block " + b +
+ throw new ReplicaAlreadyExistsException("Block " + b +
" is valid, and cannot be written to.");
}
if (isValidRbw(b)) {
- throw new ReplicaAlreadyExistsException("Block " + b +
+ throw new ReplicaAlreadyExistsException("Block " + b +
" is being written, and cannot be written to.");
}
final Map<Block, BInfo> map = getMap(b.getBlockPoolId());
@@ -1006,12 +1015,12 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
final Map<Block, BInfo> map = getMap(b.getBlockPoolId());
BInfo binfo = map.get(b.getLocalBlock());
if (binfo == null) {
- throw new IOException("No such Block " + b );
+ throw new IOException("No such Block " + b );
}
-
+
return binfo.getIStream();
}
-
+
@Override // FsDatasetSpi
public synchronized InputStream getBlockInputStream(ExtendedBlock b,
long seekOffset) throws IOException {
@@ -1033,10 +1042,10 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
final Map<Block, BInfo> map = getMap(b.getBlockPoolId());
BInfo binfo = map.get(b.getLocalBlock());
if (binfo == null) {
- throw new IOException("No such Block " + b );
+ throw new IOException("No such Block " + b );
}
if (!binfo.finalized) {
- throw new IOException("Block " + b +
+ throw new IOException("Block " + b +
" is being written, its meta cannot be read");
}
final SimulatedInputStream sin = binfo.getMetaIStream();
@@ -1044,19 +1053,17 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
}
@Override
- public Set<File> checkDataDir() {
- // nothing to check for simulated data set
- return null;
+ public void handleVolumeFailures(Set<FsVolumeSpi> failedVolumes) {
}
@Override // FsDatasetSpi
public synchronized void adjustCrcChannelPosition(ExtendedBlock b,
- ReplicaOutputStreams stream,
+ ReplicaOutputStreams stream,
int checksumSize)
throws IOException {
}
- /**
+ /**
* Simulated input and output streams
*
*/
@@ -1065,7 +1072,7 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
int currentPos = 0;
byte[] data = null;
Block theBlock = null;
-
+
/**
* An input stream of size l with repeated bytes
* @param l size of the stream
@@ -1075,7 +1082,7 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
length = l;
theBlock = b;
}
-
+
/**
* An input stream of of the supplied data
* @param iData data to construct the stream
@@ -1084,7 +1091,7 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
data = iData;
length = data.length;
}
-
+
/**
* @return the lenght of the input stream
*/
@@ -1103,9 +1110,9 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
return simulatedByte(theBlock, currentPos++) & BYTE_MASK;
}
}
-
+
@Override
- public int read(byte[] b) throws IOException {
+ public int read(byte[] b) throws IOException {
if (b == null) {
throw new NullPointerException();
@@ -1128,7 +1135,7 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
return bytesRead;
}
}
-
+
/**
* This class implements an output stream that merely throws its data away, but records its
* length.
@@ -1136,15 +1143,15 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
*/
static private class SimulatedOutputStream extends OutputStream {
long length = 0;
-
+
/**
* constructor for Simulated Output Steram
*/
SimulatedOutputStream() {
}
-
+
/**
- *
+ *
* @return the length of the data created so far.
*/
long getLength() {
@@ -1156,17 +1163,17 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
void setLength(long length) {
this.length = length;
}
-
+
@Override
public void write(int arg0) throws IOException {
length++;
}
-
+
@Override
public void write(byte[] b) throws IOException {
length += b.length;
}
-
+
@Override
public void write(byte[] b,
int off,
@@ -1174,11 +1181,11 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
length += len;
}
}
-
+
private ObjectName mbeanName;
-
+
/**
* Register the FSDataset MBean using the name
* "hadoop:service=DataNode,name=FSDatasetState-<storageid>"
@@ -1187,7 +1194,7 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
*/
void registerMBean(final String storageId) {
// We wrap to bypass standard mbean naming convetion.
- // This wraping can be removed in java 6 as it is more flexible in
+ // This wraping can be removed in java 6 as it is more flexible in
// package naming for mbeans and their impl.
StandardMBean bean;
@@ -1198,7 +1205,7 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
} catch (NotCompliantMBeanException e) {
DataNode.LOG.warn("Error registering FSDatasetState MBean", e);
}
-
+
DataNode.LOG.info("Registered FSDatasetState MBean");
}
@@ -1211,7 +1218,7 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
public String getStorageInfo() {
return "Simulated FSDataset-" + datanodeUuid;
}
-
+
@Override
public boolean hasEnoughResource() {
return true;
@@ -1224,11 +1231,11 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
final Map<Block, BInfo> map = getMap(b.getBlockPoolId());
BInfo binfo = map.get(b.getLocalBlock());
if (binfo == null) {
- throw new IOException("No such Block " + b );
+ throw new IOException("No such Block " + b );
}
- return new ReplicaRecoveryInfo(binfo.getBlockId(), binfo.getBytesOnDisk(),
- binfo.getGenerationStamp(),
+ return new ReplicaRecoveryInfo(binfo.getBlockId(), binfo.getBytesOnDisk(),
+ binfo.getGenerationStamp(),
binfo.isFinalized()?ReplicaState.FINALIZED : ReplicaState.RBW);
}
@@ -1251,13 +1258,13 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
blockMap.put(bpid, map);
storage.addBlockPool(bpid);
}
-
+
@Override // FsDatasetSpi
public void shutdownBlockPool(String bpid) {
blockMap.remove(bpid);
storage.removeBlockPool(bpid);
}
-
+
@Override // FsDatasetSpi
public void deleteBlockPool(String bpid, boolean force) {
return;
@@ -1321,7 +1328,7 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
@Override
public FsVolumeReferences getFsVolumeReferences() {
- throw new UnsupportedOperationException();
+ return new FsVolumeReferences(Collections.singletonList(volume));
}
@Override
@@ -1386,17 +1393,17 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
// TODO Auto-generated method stub
return null;
}
-
+
@Override
public void setPinning(ExtendedBlock b) throws IOException {
blockMap.get(b.getBlockPoolId()).get(b.getLocalBlock()).pinned = true;
}
-
+
@Override
public boolean getPinning(ExtendedBlock b) throws IOException {
return blockMap.get(b.getBlockPoolId()).get(b.getLocalBlock()).pinned;
}
-
+
@Override
public boolean isDeletingBlock(String bpid, long blockId) {
throw new UnsupportedOperationException();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ec80de3c/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 5d4543d..25f4d5a 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
@@ -66,6 +66,7 @@ import java.util.Random;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -124,6 +125,8 @@ public class TestDataNodeHotSwapVolumes {
1000);
/* Allow 1 volume failure */
conf.setInt(DFSConfigKeys.DFS_DATANODE_FAILED_VOLUMES_TOLERATED_KEY, 1);
+ conf.setTimeDuration(DFSConfigKeys.DFS_DATANODE_DISK_CHECK_MIN_GAP_KEY,
+ 0, TimeUnit.MILLISECONDS);
MiniDFSNNTopology nnTopology =
MiniDFSNNTopology.simpleFederatedTopology(numNameNodes);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ec80de3c/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailure.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailure.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailure.java
index ed60560..e58f993 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailure.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailure.java
@@ -34,6 +34,7 @@ import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.apache.hadoop.conf.Configuration;
@@ -119,6 +120,8 @@ public class TestDataNodeVolumeFailure {
// Allow a single volume failure (there are two volumes)
conf.setInt(DFSConfigKeys.DFS_DATANODE_FAILED_VOLUMES_TOLERATED_KEY, 1);
conf.setInt(DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_KEY, 30);
+ conf.setTimeDuration(DFSConfigKeys.DFS_DATANODE_DISK_CHECK_MIN_GAP_KEY,
+ 0, TimeUnit.MILLISECONDS);
cluster = new MiniDFSCluster.Builder(conf).numDataNodes(dn_num).build();
cluster.waitActive();
fs = cluster.getFileSystem();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ec80de3c/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailureReporting.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailureReporting.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailureReporting.java
index e6f1366..23ec8a7 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailureReporting.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailureReporting.java
@@ -29,6 +29,7 @@ import static org.junit.Assume.assumeTrue;
import java.io.File;
import java.util.ArrayList;
+import java.util.concurrent.TimeUnit;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -621,6 +622,8 @@ public class TestDataNodeVolumeFailureReporting {
conf.setInt(DFSConfigKeys.DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_KEY, 1000);
conf.setInt(DFSConfigKeys.DFS_DATANODE_FAILED_VOLUMES_TOLERATED_KEY,
failedVolumesTolerated);
+ conf.setTimeDuration(DFSConfigKeys.DFS_DATANODE_DISK_CHECK_MIN_GAP_KEY,
+ 0, TimeUnit.MILLISECONDS);
cluster = new MiniDFSCluster.Builder(conf).numDataNodes(numDataNodes)
.storagesPerDatanode(storagesPerDatanode).build();
cluster.waitActive();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ec80de3c/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailureToleration.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailureToleration.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailureToleration.java
index 2f8239e..462fdae 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailureToleration.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailureToleration.java
@@ -24,6 +24,7 @@ import static org.junit.Assume.assumeTrue;
import java.io.File;
import java.io.IOException;
+import java.util.concurrent.TimeUnit;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
@@ -144,6 +145,8 @@ public class TestDataNodeVolumeFailureToleration {
// Bring up two additional datanodes that need both of their volumes
// functioning in order to stay up.
conf.setInt(DFSConfigKeys.DFS_DATANODE_FAILED_VOLUMES_TOLERATED_KEY, 0);
+ conf.setTimeDuration(DFSConfigKeys.DFS_DATANODE_DISK_CHECK_MIN_GAP_KEY,
+ 0, TimeUnit.MILLISECONDS);
cluster.startDataNodes(conf, 2, true, null, null);
cluster.waitActive();
final DatanodeManager dm = cluster.getNamesystem().getBlockManager(
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ec80de3c/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDiskError.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDiskError.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDiskError.java
index 06b6fcd..0553a91 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDiskError.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDiskError.java
@@ -26,7 +26,9 @@ import java.io.IOException;
import java.io.RandomAccessFile;
import java.net.InetSocketAddress;
import java.net.Socket;
+import java.util.concurrent.TimeUnit;
+import com.google.common.base.Supplier;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
@@ -49,8 +51,8 @@ import org.apache.hadoop.hdfs.server.blockmanagement.DatanodeDescriptor;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsDatasetSpi;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi;
import org.apache.hadoop.hdfs.server.namenode.NameNodeAdapter;
+import org.apache.hadoop.test.GenericTestUtils;
import org.apache.hadoop.util.DataChecksum;
-import org.apache.hadoop.util.Time;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@@ -69,6 +71,9 @@ public class TestDiskError {
public void setUp() throws Exception {
conf = new HdfsConfiguration();
conf.setLong(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, 512L);
+ conf.setTimeDuration(
+ DFSConfigKeys.DFS_DATANODE_DISK_CHECK_MIN_GAP_KEY,
+ 0, TimeUnit.MILLISECONDS);
cluster = new MiniDFSCluster.Builder(conf).numDataNodes(1).build();
cluster.waitActive();
fs = cluster.getFileSystem();
@@ -214,19 +219,22 @@ public class TestDiskError {
* Before refactoring the code the above function was not getting called
* @throws IOException, InterruptedException
*/
- @Test
- public void testcheckDiskError() throws IOException, InterruptedException {
+ @Test(timeout=60000)
+ public void testcheckDiskError() throws Exception {
if(cluster.getDataNodes().size() <= 0) {
cluster.startDataNodes(conf, 1, true, null, null);
cluster.waitActive();
}
- DataNode dataNode = cluster.getDataNodes().get(0);
- long slackTime = dataNode.checkDiskErrorInterval/2;
+ final DataNode dataNode = cluster.getDataNodes().get(0);
//checking for disk error
- dataNode.checkDiskErrorAsync();
- Thread.sleep(dataNode.checkDiskErrorInterval);
- long lastDiskErrorCheck = dataNode.getLastDiskErrorCheck();
- assertTrue("Disk Error check is not performed within " + dataNode.checkDiskErrorInterval + " ms", ((Time.monotonicNow()-lastDiskErrorCheck) < (dataNode.checkDiskErrorInterval + slackTime)));
+ final long lastCheckTimestamp = dataNode.getLastDiskErrorCheck();
+ dataNode.checkDiskError();
+ GenericTestUtils.waitFor(new Supplier<Boolean>() {
+ @Override
+ public Boolean get() {
+ return dataNode.getLastDiskErrorCheck() > lastCheckTimestamp;
+ }
+ }, 100, 60000);
}
@Test
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ec80de3c/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestDatasetVolumeChecker.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestDatasetVolumeChecker.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestDatasetVolumeChecker.java
index a9041ab..940e73b 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestDatasetVolumeChecker.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestDatasetVolumeChecker.java
@@ -21,7 +21,6 @@ package org.apache.hadoop.hdfs.server.datanode.checker;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import org.apache.hadoop.hdfs.HdfsConfiguration;
-import org.apache.hadoop.hdfs.server.datanode.StorageLocation;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.*;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi.VolumeCheckContext;
import org.apache.hadoop.util.DiskChecker.DiskErrorException;
@@ -35,7 +34,10 @@ import org.junit.runners.Parameterized.Parameters;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
@@ -229,6 +231,7 @@ public class TestDatasetVolumeChecker {
static FsDatasetSpi<FsVolumeSpi> makeDataset(List<FsVolumeSpi> volumes)
throws Exception {
// Create dataset and init volume health.
+ @SuppressWarnings("unchecked") // The cast is safe.
final FsDatasetSpi<FsVolumeSpi> dataset = mock(FsDatasetSpi.class);
final FsDatasetSpi.FsVolumeReferences references = new
FsDatasetSpi.FsVolumeReferences(volumes);
@@ -236,7 +239,7 @@ public class TestDatasetVolumeChecker {
return dataset;
}
- private static List<FsVolumeSpi> makeVolumes(
+ static List<FsVolumeSpi> makeVolumes(
int numVolumes, VolumeCheckResult health) throws Exception {
final List<FsVolumeSpi> volumes = new ArrayList<>(numVolumes);
for (int i = 0; i < numVolumes; ++i) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ec80de3c/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestDatasetVolumeCheckerFailures.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestDatasetVolumeCheckerFailures.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestDatasetVolumeCheckerFailures.java
index 56446de..6f3d748 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestDatasetVolumeCheckerFailures.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestDatasetVolumeCheckerFailures.java
@@ -21,10 +21,10 @@ package org.apache.hadoop.hdfs.server.datanode.checker;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.HdfsConfiguration;
-import org.apache.hadoop.hdfs.server.datanode.StorageLocation;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.*;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi.VolumeCheckContext;
import org.apache.hadoop.util.FakeTimer;
+import org.junit.Before;
import org.junit.Test;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
@@ -37,7 +37,6 @@ import java.util.*;
import static org.hamcrest.core.Is.is;
import static org.junit.Assert.assertThat;
-import static org.mockito.Matchers.anyObject;
import static org.mockito.Mockito.*;
@@ -48,6 +47,19 @@ public class TestDatasetVolumeCheckerFailures {
public static final Logger LOG =LoggerFactory.getLogger(
TestDatasetVolumeCheckerFailures.class);
+ private FakeTimer timer;
+ private Configuration conf;
+
+ private static final long MIN_DISK_CHECK_GAP_MS = 1000; // 1 second.
+
+ @Before
+ public void commonInit() {
+ timer = new FakeTimer();
+ conf = new HdfsConfiguration();
+ conf.setTimeDuration(DFSConfigKeys.DFS_DATANODE_DISK_CHECK_MIN_GAP_KEY,
+ MIN_DISK_CHECK_GAP_MS, TimeUnit.MILLISECONDS);
+ }
+
/**
* Test timeout in {@link DatasetVolumeChecker#checkAllVolumes}.
* @throws Exception
@@ -62,7 +74,6 @@ public class TestDatasetVolumeCheckerFailures {
TestDatasetVolumeChecker.makeDataset(volumes);
// Create a disk checker with a very low timeout.
- final HdfsConfiguration conf = new HdfsConfiguration();
conf.setTimeDuration(DFSConfigKeys.DFS_DATANODE_DISK_CHECK_TIMEOUT_KEY,
1, TimeUnit.SECONDS);
final DatasetVolumeChecker checker =
@@ -87,10 +98,10 @@ public class TestDatasetVolumeCheckerFailures {
final FsDatasetSpi<FsVolumeSpi> dataset =
TestDatasetVolumeChecker.makeDataset(volumes);
- DatasetVolumeChecker checker =
- new DatasetVolumeChecker(new HdfsConfiguration(), new FakeTimer());
+ DatasetVolumeChecker checker = new DatasetVolumeChecker(conf, timer);
Set<FsVolumeSpi> failedVolumes = checker.checkAllVolumes(dataset);
assertThat(failedVolumes.size(), is(0));
+ assertThat(checker.getNumSyncDatasetChecks(), is(0L));
// The closed volume should not have been checked as it cannot
// be referenced.
@@ -99,13 +110,10 @@ public class TestDatasetVolumeCheckerFailures {
@Test(timeout=60000)
public void testMinGapIsEnforcedForSyncChecks() throws Exception {
+ final List<FsVolumeSpi> volumes =
+ TestDatasetVolumeChecker.makeVolumes(1, VolumeCheckResult.HEALTHY);
final FsDatasetSpi<FsVolumeSpi> dataset =
- TestDatasetVolumeChecker.makeDataset(new ArrayList<FsVolumeSpi>());
- final FakeTimer timer = new FakeTimer();
- final Configuration conf = new HdfsConfiguration();
- final long minGapMs = 100;
- conf.setTimeDuration(DFSConfigKeys.DFS_DATANODE_DISK_CHECK_MIN_GAP_KEY,
- minGapMs, TimeUnit.MILLISECONDS);
+ TestDatasetVolumeChecker.makeDataset(volumes);
final DatasetVolumeChecker checker = new DatasetVolumeChecker(conf, timer);
checker.checkAllVolumes(dataset);
@@ -117,7 +125,7 @@ public class TestDatasetVolumeCheckerFailures {
assertThat(checker.getNumSkippedChecks(), is(1L));
// Re-check after advancing the timer. Ensure the check is performed.
- timer.advance(minGapMs);
+ timer.advance(MIN_DISK_CHECK_GAP_MS);
checker.checkAllVolumes(dataset);
assertThat(checker.getNumSyncDatasetChecks(), is(2L));
assertThat(checker.getNumSkippedChecks(), is(1L));
@@ -125,13 +133,10 @@ public class TestDatasetVolumeCheckerFailures {
@Test(timeout=60000)
public void testMinGapIsEnforcedForASyncChecks() throws Exception {
+ final List<FsVolumeSpi> volumes =
+ TestDatasetVolumeChecker.makeVolumes(1, VolumeCheckResult.HEALTHY);
final FsDatasetSpi<FsVolumeSpi> dataset =
- TestDatasetVolumeChecker.makeDataset(new ArrayList<FsVolumeSpi>());
- final FakeTimer timer = new FakeTimer();
- final Configuration conf = new HdfsConfiguration();
- final long minGapMs = 100;
- conf.setTimeDuration(DFSConfigKeys.DFS_DATANODE_DISK_CHECK_MIN_GAP_KEY,
- minGapMs, TimeUnit.MILLISECONDS);
+ TestDatasetVolumeChecker.makeDataset(volumes);
final DatasetVolumeChecker checker = new DatasetVolumeChecker(conf, timer);
checker.checkAllVolumesAsync(dataset, null);
@@ -143,7 +148,7 @@ public class TestDatasetVolumeCheckerFailures {
assertThat(checker.getNumSkippedChecks(), is(1L));
// Re-check after advancing the timer. Ensure the check is performed.
- timer.advance(minGapMs);
+ timer.advance(MIN_DISK_CHECK_GAP_MS);
checker.checkAllVolumesAsync(dataset, null);
assertThat(checker.getNumAsyncDatasetChecks(), is(2L));
assertThat(checker.getNumSkippedChecks(), is(1L));
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ec80de3c/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 7548e83..731bb0b 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
@@ -238,8 +238,7 @@ public class ExternalDatasetImpl implements FsDatasetSpi<ExternalVolumeImpl> {
}
@Override
- public Set<File> checkDataDir() {
- return null;
+ public void handleVolumeFailures(Set<FsVolumeSpi> failedVolumes) {
}
@Override
@@ -432,14 +431,14 @@ public class ExternalDatasetImpl implements FsDatasetSpi<ExternalVolumeImpl> {
}
@Override
- public void setPinning(ExtendedBlock block) throws IOException {
+ public void setPinning(ExtendedBlock block) throws IOException {
}
@Override
public boolean getPinning(ExtendedBlock block) throws IOException {
return false;
}
-
+
@Override
public boolean isDeletingBlock(String bpid, long blockId) {
return false;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ec80de3c/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 7484a40..10b2f56 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
@@ -51,12 +51,9 @@ import org.apache.hadoop.hdfs.server.datanode.ReplicaInfo;
import org.apache.hadoop.hdfs.server.datanode.ShortCircuitRegistry;
import org.apache.hadoop.hdfs.server.datanode.StorageLocation;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsDatasetSpi;
-import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeReference;
-import org.apache.hadoop.hdfs.server.datanode.fsdataset.RoundRobinVolumeChoosingPolicy;
import org.apache.hadoop.hdfs.server.protocol.NamespaceInfo;
import org.apache.hadoop.io.MultipleIOException;
import org.apache.hadoop.test.GenericTestUtils;
-import org.apache.hadoop.util.DiskChecker;
import org.apache.hadoop.util.FakeTimer;
import org.apache.hadoop.util.StringUtils;
import org.junit.Assert;
@@ -64,8 +61,6 @@ import org.junit.Before;
import org.junit.Test;
import org.mockito.Matchers;
import org.mockito.Mockito;
-import org.mockito.invocation.InvocationOnMock;
-import org.mockito.stubbing.Answer;
import java.io.File;
import java.io.FileOutputStream;
@@ -74,16 +69,18 @@ import java.io.OutputStreamWriter;
import java.io.Writer;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.concurrent.CountDownLatch;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
+import java.util.concurrent.TimeUnit;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DN_CACHED_DFSUSED_CHECK_INTERVAL_MS;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_SCAN_PERIOD_HOURS_KEY;
+import static org.hamcrest.core.Is.is;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
@@ -92,13 +89,10 @@ import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyListOf;
import static org.mockito.Matchers.anyString;
import static org.mockito.Matchers.eq;
-import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
-import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import org.slf4j.Logger;
@@ -120,7 +114,7 @@ public class TestFsDatasetImpl {
private DataNode datanode;
private DataStorage storage;
private FsDatasetImpl dataset;
-
+
private final static String BLOCKPOOL = "BP-TEST";
private static Storage.StorageDirectory createStorageDirectory(File root) {
@@ -324,64 +318,6 @@ public class TestFsDatasetImpl {
assertEquals(numExistingVolumes, getNumVolumes());
}
- @Test(timeout = 5000)
- public void testChangeVolumeWithRunningCheckDirs() throws IOException {
- RoundRobinVolumeChoosingPolicy<FsVolumeImpl> blockChooser =
- new RoundRobinVolumeChoosingPolicy<>();
- conf.setLong(DFSConfigKeys.DFS_DATANODE_SCAN_PERIOD_HOURS_KEY, -1);
- final BlockScanner blockScanner = new BlockScanner(datanode);
- final FsVolumeList volumeList = new FsVolumeList(
- Collections.<VolumeFailureInfo>emptyList(), blockScanner, blockChooser);
- final List<FsVolumeImpl> oldVolumes = new ArrayList<>();
-
- // Initialize FsVolumeList with 5 mock volumes.
- final int NUM_VOLUMES = 5;
- for (int i = 0; i < NUM_VOLUMES; i++) {
- FsVolumeImpl volume = mock(FsVolumeImpl.class);
- oldVolumes.add(volume);
- when(volume.getBasePath()).thenReturn("data" + i);
- when(volume.checkClosed()).thenReturn(true);
- FsVolumeReference ref = mock(FsVolumeReference.class);
- when(ref.getVolume()).thenReturn(volume);
- volumeList.addVolume(ref);
- }
-
- // When call checkDirs() on the 2nd volume, anther "thread" removes the 5th
- // volume and add another volume. It does not affect checkDirs() running.
- final FsVolumeImpl newVolume = mock(FsVolumeImpl.class);
- final FsVolumeReference newRef = mock(FsVolumeReference.class);
- when(newRef.getVolume()).thenReturn(newVolume);
- when(newVolume.getBasePath()).thenReturn("data4");
- FsVolumeImpl blockedVolume = volumeList.getVolumes().get(1);
- doAnswer(new Answer() {
- @Override
- public Object answer(InvocationOnMock invocationOnMock)
- throws Throwable {
- volumeList.removeVolume(new File("data4"), false);
- volumeList.addVolume(newRef);
- return null;
- }
- }).when(blockedVolume).checkDirs();
-
- FsVolumeImpl brokenVolume = volumeList.getVolumes().get(2);
- doThrow(new DiskChecker.DiskErrorException("broken"))
- .when(brokenVolume).checkDirs();
-
- volumeList.checkDirs();
-
- // Since FsVolumeImpl#checkDirs() get a snapshot of the list of volumes
- // before running removeVolume(), it is supposed to run checkDirs() on all
- // the old volumes.
- for (FsVolumeImpl volume : oldVolumes) {
- verify(volume).checkDirs();
- }
- // New volume is not visible to checkDirs() process.
- verify(newVolume, never()).checkDirs();
- assertTrue(volumeList.getVolumes().contains(newVolume));
- assertFalse(volumeList.getVolumes().contains(brokenVolume));
- assertEquals(NUM_VOLUMES - 1, volumeList.getVolumes().size());
- }
-
@Test
public void testAddVolumeFailureReleasesInUseLock() throws IOException {
FsDatasetImpl spyDataset = spy(dataset);
@@ -417,7 +353,7 @@ public class TestFsDatasetImpl {
FsDatasetTestUtil.assertFileLockReleased(badDir.toString());
}
-
+
@Test
public void testDeletingBlocks() throws IOException {
HdfsConfiguration conf = new HdfsConfiguration();
@@ -425,7 +361,7 @@ public class TestFsDatasetImpl {
try {
cluster.waitActive();
DataNode dn = cluster.getDataNodes().get(0);
-
+
FsDatasetSpi<?> ds = DataNodeTestUtils.getFSDataset(dn);
ds.addBlockPool(BLOCKPOOL, conf);
FsVolumeImpl vol;
@@ -697,6 +633,9 @@ public class TestFsDatasetImpl {
Configuration config = new HdfsConfiguration();
config.setLong(
DFSConfigKeys.DFS_DATANODE_XCEIVER_STOP_TIMEOUT_MILLIS_KEY, 1000);
+ config.setTimeDuration(
+ DFSConfigKeys.DFS_DATANODE_DISK_CHECK_MIN_GAP_KEY, 0,
+ TimeUnit.MILLISECONDS);
config.setInt(DFSConfigKeys.DFS_DATANODE_FAILED_VOLUMES_TOLERATED_KEY, 1);
cluster = new MiniDFSCluster.Builder(config).numDataNodes(1).build();
@@ -714,6 +653,8 @@ public class TestFsDatasetImpl {
getVolume(block);
File finalizedDir = volume.getFinalizedDir(cluster.getNamesystem()
.getBlockPoolId());
+ LocatedBlock lb = DFSTestUtil.getAllBlocks(fs, filePath).get(0);
+ DatanodeInfo info = lb.getLocations()[0];
if (finalizedDir.exists()) {
// Remove write and execute access so that checkDiskErrorThread detects
@@ -724,15 +665,14 @@ public class TestFsDatasetImpl {
Assert.assertTrue("Reference count for the volume should be greater "
+ "than 0", volume.getReferenceCount() > 0);
// Invoke the synchronous checkDiskError method
- dataNode.getFSDataset().checkDataDir();
+ dataNode.checkDiskError();
// Sleep for 1 second so that datanode can interrupt and cluster clean up
GenericTestUtils.waitFor(new Supplier<Boolean>() {
@Override public Boolean get() {
return volume.getReferenceCount() == 0;
}
}, 100, 10);
- LocatedBlock lb = DFSTestUtil.getAllBlocks(fs, filePath).get(0);
- DatanodeInfo info = lb.getLocations()[0];
+ assertThat(dataNode.getFSDataset().getNumFailedVolumes(), is(1));
try {
out.close();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ec80de3c/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/TestFsVolumeList.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/TestFsVolumeList.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/TestFsVolumeList.java
index 3d4c38c..f511dc6 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/TestFsVolumeList.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/TestFsVolumeList.java
@@ -102,38 +102,6 @@ public class TestFsVolumeList {
}
@Test(timeout=30000)
- public void testCheckDirsWithClosedVolume() throws IOException {
- FsVolumeList volumeList = new FsVolumeList(
- Collections.<VolumeFailureInfo>emptyList(), blockScanner, blockChooser);
- final List<FsVolumeImpl> volumes = new ArrayList<>();
- for (int i = 0; i < 3; i++) {
- File curDir = new File(baseDir, "volume-" + i);
- curDir.mkdirs();
- FsVolumeImpl volume = new FsVolumeImpl(dataset, "storage-id", curDir,
- conf, StorageType.DEFAULT);
- volumes.add(volume);
- volumeList.addVolume(volume.obtainReference());
- }
-
- // Close the 2nd volume.
- volumes.get(1).setClosed();
- try {
- GenericTestUtils.waitFor(new Supplier<Boolean>() {
- @Override
- public Boolean get() {
- return volumes.get(1).checkClosed();
- }
- }, 100, 3000);
- } catch (TimeoutException e) {
- fail("timed out while waiting for volume to be removed.");
- } catch (InterruptedException ie) {
- Thread.currentThread().interrupt();
- }
- // checkDirs() should ignore the 2nd volume since it is closed.
- volumeList.checkDirs();
- }
-
- @Test(timeout=30000)
public void testReleaseVolumeRefIfNoBlockScanner() throws IOException {
FsVolumeList volumeList = new FsVolumeList(
Collections.<VolumeFailureInfo>emptyList(), null, blockChooser);
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[4/5] hadoop git commit: HDFS-11274. Datanode should only check the
failed volume upon IO errors. Contributed by Xiaoyu Yao.
Posted by ar...@apache.org.
HDFS-11274. Datanode should only check the failed volume upon IO errors. Contributed by Xiaoyu Yao.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/eafaddca
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/eafaddca
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/eafaddca
Branch: refs/heads/branch-2
Commit: eafaddca1a69dd02c6e72f0b21b66f58b0c18dab
Parents: ec80de3
Author: Xiaoyu Yao <xy...@apache.org>
Authored: Wed Dec 28 22:08:13 2016 -0800
Committer: Arpit Agarwal <ar...@apache.org>
Committed: Mon Jan 16 22:14:04 2017 -0800
----------------------------------------------------------------------
.../hdfs/server/datanode/BlockReceiver.java | 12 +-
.../server/datanode/CountingFileIoEvents.java | 3 +-
.../hadoop/hdfs/server/datanode/DataNode.java | 95 +++++++++-----
.../server/datanode/DefaultFileIoEvents.java | 2 +-
.../hdfs/server/datanode/DirectoryScanner.java | 2 +-
.../hdfs/server/datanode/FileIoEvents.java | 36 ++++--
.../hdfs/server/datanode/FileIoProvider.java | 89 +++++++------
.../server/datanode/ProfilingFileIoEvents.java | 2 +-
.../hdfs/server/datanode/ReplicaInfo.java | 2 +-
.../server/datanode/checker/AsyncChecker.java | 5 +-
.../datanode/checker/DatasetVolumeChecker.java | 72 +++++++----
.../checker/StorageLocationChecker.java | 8 +-
.../datanode/checker/ThrottledAsyncChecker.java | 19 ++-
.../datanode/fsdataset/impl/BlockPoolSlice.java | 5 +-
.../datanode/fsdataset/impl/FsDatasetImpl.java | 13 +-
.../datanode/fsdataset/impl/FsVolumeImpl.java | 39 +++---
.../server/datanode/SimulatedFSDataset.java | 2 +-
.../datanode/TestDataNodeHotSwapVolumes.java | 2 +-
.../TestDataNodeVolumeFailureReporting.java | 26 ++--
.../checker/TestDatasetVolumeChecker.java | 52 +++++---
.../TestDatasetVolumeCheckerFailures.java | 23 ----
.../checker/TestThrottledAsyncChecker.java | 129 +++++++++++--------
.../fsdataset/impl/TestFsDatasetImpl.java | 2 +-
23 files changed, 376 insertions(+), 264 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eafaddca/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 7fdffb4..517a709 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
@@ -276,10 +276,9 @@ class BlockReceiver implements Closeable {
IOException cause = DatanodeUtil.getCauseIfDiskError(ioe);
DataNode.LOG.warn("IOException in BlockReceiver constructor"
+ (cause == null ? "" : ". Cause is "), cause);
-
- if (cause != null) { // possible disk error
+ if (cause != null) {
ioe = cause;
- datanode.checkDiskErrorAsync();
+ // Volume error check moved to FileIoProvider
}
throw ioe;
@@ -361,9 +360,8 @@ class BlockReceiver implements Closeable {
if (measuredFlushTime) {
datanode.metrics.addFlushNanos(flushTotalNanos);
}
- // disk check
if(ioe != null) {
- datanode.checkDiskErrorAsync();
+ // Volume error check moved to FileIoProvider
throw ioe;
}
}
@@ -786,7 +784,7 @@ class BlockReceiver implements Closeable {
manageWriterOsCache(offsetInBlock);
}
} catch (IOException iex) {
- datanode.checkDiskErrorAsync();
+ // Volume error check moved to FileIoProvider
throw iex;
}
}
@@ -1395,7 +1393,7 @@ class BlockReceiver implements Closeable {
} catch (IOException e) {
LOG.warn("IOException in BlockReceiver.run(): ", e);
if (running) {
- datanode.checkDiskErrorAsync();
+ // Volume error check moved to FileIoProvider
LOG.info(myString, e);
running = false;
if (!Thread.interrupted()) { // failure not caused by interruption
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eafaddca/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/CountingFileIoEvents.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/CountingFileIoEvents.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/CountingFileIoEvents.java
index a70c151..7c6bfd6 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/CountingFileIoEvents.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/CountingFileIoEvents.java
@@ -37,7 +37,7 @@ import java.util.concurrent.atomic.AtomicLong;
*/
@InterfaceAudience.Private
@InterfaceStability.Unstable
-public class CountingFileIoEvents implements FileIoEvents {
+public class CountingFileIoEvents extends FileIoEvents {
private final Map<OPERATION, Counts> counts;
private static class Counts {
@@ -90,7 +90,6 @@ public class CountingFileIoEvents implements FileIoEvents {
public void onFailure(
@Nullable FsVolumeSpi volume, OPERATION op, Exception e, long begin) {
counts.get(op).failures.incrementAndGet();
-
}
@Override
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eafaddca/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
index bddde1f..dc7d267 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
@@ -371,6 +371,7 @@ public class DataNode extends ReconfigurableBase
SaslDataTransferServer saslServer;
private final boolean getHdfsBlockLocationsEnabled;
private ObjectName dataNodeInfoBeanName;
+ // Test verification only
private volatile long lastDiskErrorCheck;
private String supergroup;
private boolean isPermissionEnabled;
@@ -408,7 +409,7 @@ public class DataNode extends ReconfigurableBase
this.tracer = createTracer(conf);
this.tracerConfigurationManager =
new TracerConfigurationManager(DATANODE_HTRACE_PREFIX, conf);
- this.fileIoProvider = new FileIoProvider(conf);
+ this.fileIoProvider = new FileIoProvider(conf, this);
this.fileDescriptorPassingDisabledReason = null;
this.maxNumberOfBlocksToLog = 0;
this.confVersion = null;
@@ -433,7 +434,7 @@ public class DataNode extends ReconfigurableBase
this.tracer = createTracer(conf);
this.tracerConfigurationManager =
new TracerConfigurationManager(DATANODE_HTRACE_PREFIX, conf);
- this.fileIoProvider = new FileIoProvider(conf);
+ this.fileIoProvider = new FileIoProvider(conf, this);
this.blockScanner = new BlockScanner(this);
this.lastDiskErrorCheck = 0;
this.maxNumberOfBlocksToLog = conf.getLong(DFS_MAX_NUM_BLOCKS_TO_LOG_KEY,
@@ -783,7 +784,7 @@ public class DataNode extends ReconfigurableBase
/**
* Remove volumes from DataNode.
- * See {@link #removeVolumes(Set, boolean)} for details.
+ * See {@link #removeVolumes(Collection, boolean)} for details.
*
* @param locations the StorageLocations of the volumes to be removed.
* @throws IOException
@@ -810,7 +811,7 @@ public class DataNode extends ReconfigurableBase
* <ul>Reset configuration DATA_DIR and {@link #dataDirs} to represent
* active volumes.</ul>
* </li>
- * @param absoluteVolumePaths the absolute path of volumes.
+ * @param storageLocations the absolute path of volumes.
* @param clearFailure if true, clears the failure information related to the
* volumes.
* @throws IOException
@@ -1258,7 +1259,7 @@ public class DataNode extends ReconfigurableBase
* If conf's CONFIG_PROPERTY_SIMULATED property is set
* then a simulated storage based data node is created.
*
- * @param dataDirs - only for a non-simulated storage data node
+ * @param dataDirectories - only for a non-simulated storage data node
* @throws IOException
*/
void startDataNode(List<StorageLocation> dataDirectories,
@@ -2020,16 +2021,39 @@ public class DataNode extends ReconfigurableBase
tracer.close();
}
-
/**
- * Check if there is a disk failure asynchronously and if so, handle the error
+ * Check if there is a disk failure asynchronously
+ * and if so, handle the error.
*/
+ @VisibleForTesting
public void checkDiskErrorAsync() {
volumeChecker.checkAllVolumesAsync(
data, new DatasetVolumeChecker.Callback() {
@Override
public void call(Set<FsVolumeSpi> healthyVolumes,
Set<FsVolumeSpi> failedVolumes) {
+ if (failedVolumes.size() > 0) {
+ LOG.warn("checkDiskErrorAsync callback got {} failed volumes: {}",
+ failedVolumes.size(), failedVolumes);
+ } else {
+ LOG.debug("checkDiskErrorAsync: no volume failures detected");
+ }
+ lastDiskErrorCheck = Time.monotonicNow();
+ handleVolumeFailures(failedVolumes);
+ }
+ });
+ }
+
+ /**
+ * Check if there is a disk failure asynchronously
+ * and if so, handle the error.
+ */
+ public void checkDiskErrorAsync(FsVolumeSpi volume) {
+ volumeChecker.checkVolume(
+ volume, new DatasetVolumeChecker.Callback() {
+ @Override
+ public void call(Set<FsVolumeSpi> healthyVolumes,
+ Set<FsVolumeSpi> failedVolumes) {
if (failedVolumes.size() > 0) {
LOG.warn("checkDiskErrorAsync callback got {} failed volumes: {}",
failedVolumes.size(), failedVolumes);
@@ -2037,14 +2061,15 @@ public class DataNode extends ReconfigurableBase
LOG.debug("checkDiskErrorAsync: no volume failures detected");
}
lastDiskErrorCheck = Time.monotonicNow();
- DataNode.this.handleVolumeFailures(failedVolumes);
+ handleVolumeFailures(failedVolumes);
}
});
}
- private void handleDiskError(String errMsgr) {
+ private void handleDiskError(String failedVolumes) {
final boolean hasEnoughResources = data.hasEnoughResource();
- LOG.warn("DataNode.handleDiskError: Keep Running: " + hasEnoughResources);
+ LOG.warn("DataNode.handleDiskError on : [" + failedVolumes +
+ "] Keep Running: " + hasEnoughResources);
// If we have enough active valid volumes then we do not want to
// shutdown the DN completely.
@@ -2054,7 +2079,7 @@ public class DataNode extends ReconfigurableBase
//inform NameNodes
for(BPOfferService bpos: blockPoolManager.getAllNamenodeThreads()) {
- bpos.trySendErrorReport(dpError, errMsgr);
+ bpos.trySendErrorReport(dpError, failedVolumes);
}
if(hasEnoughResources) {
@@ -2062,7 +2087,8 @@ public class DataNode extends ReconfigurableBase
return; // do not shutdown
}
- LOG.warn("DataNode is shutting down: " + errMsgr);
+ LOG.warn("DataNode is shutting down due to failed volumes: ["
+ + failedVolumes + "]");
shouldRun = false;
}
@@ -2412,8 +2438,11 @@ public class DataNode extends ReconfigurableBase
}
LOG.warn(bpReg + ":Failed to transfer " + b + " to " +
targets[0] + " got ", ie);
- // check if there are any disk problem
- checkDiskErrorAsync();
+ // disk check moved to FileIoProvider
+ IOException cause = DatanodeUtil.getCauseIfDiskError(ie);
+ if (cause != null) { // possible disk error
+ LOG.warn("IOException in DataTransfer#run(). Cause is ", cause);
+ }
} finally {
xmitsInProgress.getAndDecrement();
IOUtils.closeStream(blockSender);
@@ -3167,28 +3196,36 @@ public class DataNode extends ReconfigurableBase
}
private void handleVolumeFailures(Set<FsVolumeSpi> unhealthyVolumes) {
+ if (unhealthyVolumes.isEmpty()) {
+ LOG.debug("handleVolumeFailures done with empty " +
+ "unhealthyVolumes");
+ return;
+ }
+
data.handleVolumeFailures(unhealthyVolumes);
final Set<File> unhealthyDirs = new HashSet<>(unhealthyVolumes.size());
- if (!unhealthyVolumes.isEmpty()) {
- StringBuilder sb = new StringBuilder("DataNode failed volumes:");
- for (FsVolumeSpi vol : unhealthyVolumes) {
- unhealthyDirs.add(new File(vol.getBasePath()).getAbsoluteFile());
- sb.append(vol).append(";");
- }
+ StringBuilder sb = new StringBuilder("DataNode failed volumes:");
+ for (FsVolumeSpi vol : unhealthyVolumes) {
+ unhealthyDirs.add(new File(vol.getBasePath()).getAbsoluteFile());
+ sb.append(vol).append(";");
+ }
- try {
- // Remove all unhealthy volumes from DataNode.
- removeVolumes(unhealthyDirs, false);
- } catch (IOException e) {
- LOG.warn("Error occurred when removing unhealthy storage dirs: "
- + e.getMessage(), e);
- }
- LOG.info(sb.toString());
- handleDiskError(sb.toString());
+ try {
+ // Remove all unhealthy volumes from DataNode.
+ removeVolumes(unhealthyDirs, false);
+ } catch (IOException e) {
+ LOG.warn("Error occurred when removing unhealthy storage dirs: "
+ + e.getMessage(), e);
}
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(sb.toString());
+ }
+ // send blockreport regarding volume failure
+ handleDiskError(sb.toString());
}
+ @VisibleForTesting
public long getLastDiskErrorCheck() {
return lastDiskErrorCheck;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eafaddca/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DefaultFileIoEvents.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DefaultFileIoEvents.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DefaultFileIoEvents.java
index bd4932b..6a12aae 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DefaultFileIoEvents.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DefaultFileIoEvents.java
@@ -31,7 +31,7 @@ import javax.annotation.Nullable;
*/
@InterfaceAudience.Private
@InterfaceStability.Unstable
-public final class DefaultFileIoEvents implements FileIoEvents {
+public final class DefaultFileIoEvents extends FileIoEvents {
@Override
public long beforeMetadataOp(
@Nullable FsVolumeSpi volume, OPERATION op) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eafaddca/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DirectoryScanner.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DirectoryScanner.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DirectoryScanner.java
index d9a6749..15208c4 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DirectoryScanner.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DirectoryScanner.java
@@ -855,7 +855,7 @@ public class DirectoryScanner implements Runnable {
} catch (IOException ioe) {
LOG.warn("Exception occured while compiling report: ", ioe);
// Initiate a check on disk failure.
- datanode.checkDiskErrorAsync();
+ datanode.checkDiskErrorAsync(volume);
// Ignore this directory and proceed.
return report;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eafaddca/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/FileIoEvents.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/FileIoEvents.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/FileIoEvents.java
index 48e703f..10f2a0c 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/FileIoEvents.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/FileIoEvents.java
@@ -32,7 +32,7 @@ import javax.annotation.Nullable;
*/
@InterfaceAudience.Private
@InterfaceStability.Unstable
-public interface FileIoEvents {
+public abstract class FileIoEvents {
/**
* Invoked before a filesystem metadata operation.
@@ -42,7 +42,7 @@ public interface FileIoEvents {
* @return timestamp at which the operation was started. 0 if
* unavailable.
*/
- long beforeMetadataOp(@Nullable FsVolumeSpi volume, OPERATION op);
+ abstract long beforeMetadataOp(@Nullable FsVolumeSpi volume, OPERATION op);
/**
* Invoked after a filesystem metadata operation has completed.
@@ -52,7 +52,8 @@ public interface FileIoEvents {
* @param begin timestamp at which the operation was started. 0
* if unavailable.
*/
- void afterMetadataOp(@Nullable FsVolumeSpi volume, OPERATION op, long begin);
+ abstract void afterMetadataOp(@Nullable FsVolumeSpi volume, OPERATION op,
+ long begin);
/**
* Invoked before a read/write/flush/channel transfer operation.
@@ -63,7 +64,8 @@ public interface FileIoEvents {
* @return timestamp at which the operation was started. 0 if
* unavailable.
*/
- long beforeFileIo(@Nullable FsVolumeSpi volume, OPERATION op, long len);
+ abstract long beforeFileIo(@Nullable FsVolumeSpi volume, OPERATION op,
+ long len);
/**
@@ -76,22 +78,38 @@ public interface FileIoEvents {
* @return timestamp at which the operation was started. 0 if
* unavailable.
*/
- void afterFileIo(@Nullable FsVolumeSpi volume, OPERATION op,
- long begin, long len);
+ abstract void afterFileIo(@Nullable FsVolumeSpi volume, OPERATION op,
+ long begin, long len);
/**
* Invoked if an operation fails with an exception.
- * @param volume target volume for the operation. Null if unavailable.
+ * @param volume target volume for the operation. Null if unavailable.
* @param op type of operation.
* @param e Exception encountered during the operation.
* @param begin time at which the operation was started.
*/
- void onFailure(
+ abstract void onFailure(
@Nullable FsVolumeSpi volume, OPERATION op, Exception e, long begin);
/**
+ * Invoked by FileIoProvider if an operation fails with an exception.
+ * @param datanode datanode that runs volume check upon volume io failure
+ * @param volume target volume for the operation. Null if unavailable.
+ * @param op type of operation.
+ * @param e Exception encountered during the operation.
+ * @param begin time at which the operation was started.
+ */
+ void onFailure(DataNode datanode,
+ @Nullable FsVolumeSpi volume, OPERATION op, Exception e, long begin) {
+ onFailure(volume, op, e, begin);
+ if (datanode != null && volume != null) {
+ datanode.checkDiskErrorAsync(volume);
+ }
+ }
+
+ /**
* Return statistics as a JSON string.
* @return
*/
- @Nullable String getStatistics();
+ @Nullable abstract String getStatistics();
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eafaddca/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/FileIoProvider.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/FileIoProvider.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/FileIoProvider.java
index df7718e..4a71179 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/FileIoProvider.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/FileIoProvider.java
@@ -79,12 +79,16 @@ public class FileIoProvider {
FileIoProvider.class);
private final FileIoEvents eventHooks;
+ private final DataNode datanode;
/**
* @param conf Configuration object. May be null. When null,
* the event handlers are no-ops.
+ * @param datanode datanode that owns this FileIoProvider. Used for
+ * IO error based volume checker callback
*/
- public FileIoProvider(@Nullable Configuration conf) {
+ public FileIoProvider(@Nullable Configuration conf,
+ final DataNode datanode) {
if (conf != null) {
final Class<? extends FileIoEvents> clazz = conf.getClass(
DFSConfigKeys.DFS_DATANODE_FILE_IO_EVENTS_CLASS_KEY,
@@ -94,6 +98,7 @@ public class FileIoProvider {
} else {
eventHooks = new DefaultFileIoEvents();
}
+ this.datanode = datanode;
}
/**
@@ -139,7 +144,7 @@ public class FileIoProvider {
f.flush();
eventHooks.afterFileIo(volume, FLUSH, begin, 0);
} catch (Exception e) {
- eventHooks.onFailure(volume, FLUSH, e, begin);
+ eventHooks.onFailure(datanode, volume, FLUSH, e, begin);
throw e;
}
}
@@ -157,7 +162,7 @@ public class FileIoProvider {
fos.getChannel().force(true);
eventHooks.afterFileIo(volume, SYNC, begin, 0);
} catch (Exception e) {
- eventHooks.onFailure(volume, SYNC, e, begin);
+ eventHooks.onFailure(datanode, volume, SYNC, e, begin);
throw e;
}
}
@@ -176,7 +181,7 @@ public class FileIoProvider {
NativeIO.POSIX.syncFileRangeIfPossible(outFd, offset, numBytes, flags);
eventHooks.afterFileIo(volume, SYNC, begin, 0);
} catch (Exception e) {
- eventHooks.onFailure(volume, SYNC, e, begin);
+ eventHooks.onFailure(datanode, volume, SYNC, e, begin);
throw e;
}
}
@@ -196,7 +201,7 @@ public class FileIoProvider {
identifier, outFd, offset, length, flags);
eventHooks.afterMetadataOp(volume, FADVISE, begin);
} catch (Exception e) {
- eventHooks.onFailure(volume, FADVISE, e, begin);
+ eventHooks.onFailure(datanode, volume, FADVISE, e, begin);
throw e;
}
}
@@ -214,7 +219,7 @@ public class FileIoProvider {
eventHooks.afterMetadataOp(volume, DELETE, begin);
return deleted;
} catch (Exception e) {
- eventHooks.onFailure(volume, DELETE, e, begin);
+ eventHooks.onFailure(datanode, volume, DELETE, e, begin);
throw e;
}
}
@@ -236,7 +241,7 @@ public class FileIoProvider {
}
return deleted;
} catch (Exception e) {
- eventHooks.onFailure(volume, DELETE, e, begin);
+ eventHooks.onFailure(datanode, volume, DELETE, e, begin);
throw e;
}
}
@@ -264,7 +269,7 @@ public class FileIoProvider {
waitTime, transferTime);
eventHooks.afterFileIo(volume, TRANSFER, begin, count);
} catch (Exception e) {
- eventHooks.onFailure(volume, TRANSFER, e, begin);
+ eventHooks.onFailure(datanode, volume, TRANSFER, e, begin);
throw e;
}
}
@@ -285,7 +290,7 @@ public class FileIoProvider {
eventHooks.afterMetadataOp(volume, OPEN, begin);
return created;
} catch (Exception e) {
- eventHooks.onFailure(volume, OPEN, e, begin);
+ eventHooks.onFailure(datanode, volume, OPEN, e, begin);
throw e;
}
}
@@ -312,7 +317,7 @@ public class FileIoProvider {
return fis;
} catch(Exception e) {
org.apache.commons.io.IOUtils.closeQuietly(fis);
- eventHooks.onFailure(volume, OPEN, e, begin);
+ eventHooks.onFailure(datanode, volume, OPEN, e, begin);
throw e;
}
}
@@ -328,7 +333,7 @@ public class FileIoProvider {
* @param f File object.
* @param append if true, then bytes will be written to the end of the
* file rather than the beginning.
- * @param FileOutputStream to the given file object.
+ * @return FileOutputStream to the given file object.
* @throws FileNotFoundException
*/
public FileOutputStream getFileOutputStream(
@@ -342,7 +347,7 @@ public class FileIoProvider {
return fos;
} catch(Exception e) {
org.apache.commons.io.IOUtils.closeQuietly(fos);
- eventHooks.onFailure(volume, OPEN, e, begin);
+ eventHooks.onFailure(datanode, volume, OPEN, e, begin);
throw e;
}
}
@@ -372,7 +377,7 @@ public class FileIoProvider {
* before delegating to the wrapped stream.
*
* @param volume target volume. null if unavailable.
- * @param f File object.
+ * @param fd File descriptor object.
* @return FileOutputStream to the given file object.
* @throws FileNotFoundException
*/
@@ -407,7 +412,7 @@ public class FileIoProvider {
return fis;
} catch(Exception e) {
org.apache.commons.io.IOUtils.closeQuietly(fis);
- eventHooks.onFailure(volume, OPEN, e, begin);
+ eventHooks.onFailure(datanode, volume, OPEN, e, begin);
throw e;
}
}
@@ -438,7 +443,7 @@ public class FileIoProvider {
return fis;
} catch(Exception e) {
org.apache.commons.io.IOUtils.closeQuietly(fis);
- eventHooks.onFailure(volume, OPEN, e, begin);
+ eventHooks.onFailure(datanode, volume, OPEN, e, begin);
throw e;
}
}
@@ -468,7 +473,7 @@ public class FileIoProvider {
return raf;
} catch(Exception e) {
org.apache.commons.io.IOUtils.closeQuietly(raf);
- eventHooks.onFailure(volume, OPEN, e, begin);
+ eventHooks.onFailure(datanode, volume, OPEN, e, begin);
throw e;
}
}
@@ -487,7 +492,7 @@ public class FileIoProvider {
eventHooks.afterMetadataOp(volume, DELETE, begin);
return deleted;
} catch(Exception e) {
- eventHooks.onFailure(volume, DELETE, e, begin);
+ eventHooks.onFailure(datanode, volume, DELETE, e, begin);
throw e;
}
}
@@ -508,7 +513,7 @@ public class FileIoProvider {
FileUtil.replaceFile(src, target);
eventHooks.afterMetadataOp(volume, MOVE, begin);
} catch(Exception e) {
- eventHooks.onFailure(volume, MOVE, e, begin);
+ eventHooks.onFailure(datanode, volume, MOVE, e, begin);
throw e;
}
}
@@ -530,7 +535,7 @@ public class FileIoProvider {
Storage.rename(src, target);
eventHooks.afterMetadataOp(volume, MOVE, begin);
} catch(Exception e) {
- eventHooks.onFailure(volume, MOVE, e, begin);
+ eventHooks.onFailure(datanode, volume, MOVE, e, begin);
throw e;
}
}
@@ -552,7 +557,7 @@ public class FileIoProvider {
FileUtils.moveFile(src, target);
eventHooks.afterMetadataOp(volume, MOVE, begin);
} catch(Exception e) {
- eventHooks.onFailure(volume, MOVE, e, begin);
+ eventHooks.onFailure(datanode, volume, MOVE, e, begin);
throw e;
}
}
@@ -576,7 +581,7 @@ public class FileIoProvider {
Files.move(src, target, options);
eventHooks.afterMetadataOp(volume, MOVE, begin);
} catch(Exception e) {
- eventHooks.onFailure(volume, MOVE, e, begin);
+ eventHooks.onFailure(datanode, volume, MOVE, e, begin);
throw e;
}
}
@@ -625,7 +630,7 @@ public class FileIoProvider {
Storage.nativeCopyFileUnbuffered(src, target, preserveFileDate);
eventHooks.afterFileIo(volume, NATIVE_COPY, begin, length);
} catch(Exception e) {
- eventHooks.onFailure(volume, NATIVE_COPY, e, begin);
+ eventHooks.onFailure(datanode, volume, NATIVE_COPY, e, begin);
throw e;
}
}
@@ -650,7 +655,7 @@ public class FileIoProvider {
isDirectory = !created && dir.isDirectory();
eventHooks.afterMetadataOp(volume, MKDIRS, begin);
} catch(Exception e) {
- eventHooks.onFailure(volume, MKDIRS, e, begin);
+ eventHooks.onFailure(datanode, volume, MKDIRS, e, begin);
throw e;
}
@@ -676,7 +681,7 @@ public class FileIoProvider {
succeeded = dir.isDirectory() || dir.mkdirs();
eventHooks.afterMetadataOp(volume, MKDIRS, begin);
} catch(Exception e) {
- eventHooks.onFailure(volume, MKDIRS, e, begin);
+ eventHooks.onFailure(datanode, volume, MKDIRS, e, begin);
throw e;
}
@@ -702,7 +707,7 @@ public class FileIoProvider {
eventHooks.afterMetadataOp(volume, LIST, begin);
return children;
} catch(Exception e) {
- eventHooks.onFailure(volume, LIST, e, begin);
+ eventHooks.onFailure(datanode, volume, LIST, e, begin);
throw e;
}
}
@@ -712,7 +717,7 @@ public class FileIoProvider {
* {@link FileUtil#listFiles(File)}.
*
* @param volume target volume. null if unavailable.
- * @param Driectory to be listed.
+ * @param dir directory to be listed.
* @return array of strings representing the directory entries.
* @throws IOException
*/
@@ -724,7 +729,7 @@ public class FileIoProvider {
eventHooks.afterMetadataOp(volume, LIST, begin);
return children;
} catch(Exception e) {
- eventHooks.onFailure(volume, LIST, e, begin);
+ eventHooks.onFailure(datanode, volume, LIST, e, begin);
throw e;
}
}
@@ -747,7 +752,7 @@ public class FileIoProvider {
eventHooks.afterMetadataOp(volume, LIST, begin);
return children;
} catch(Exception e) {
- eventHooks.onFailure(volume, LIST, e, begin);
+ eventHooks.onFailure(datanode, volume, LIST, e, begin);
throw e;
}
}
@@ -769,7 +774,7 @@ public class FileIoProvider {
eventHooks.afterMetadataOp(volume, LIST, begin);
return count;
} catch(Exception e) {
- eventHooks.onFailure(volume, LIST, e, begin);
+ eventHooks.onFailure(datanode, volume, LIST, e, begin);
throw e;
}
}
@@ -788,7 +793,7 @@ public class FileIoProvider {
eventHooks.afterMetadataOp(volume, EXISTS, begin);
return exists;
} catch(Exception e) {
- eventHooks.onFailure(volume, EXISTS, e, begin);
+ eventHooks.onFailure(datanode, volume, EXISTS, e, begin);
throw e;
}
}
@@ -829,7 +834,7 @@ public class FileIoProvider {
eventHooks.afterFileIo(volume, READ, begin, 1);
return b;
} catch(Exception e) {
- eventHooks.onFailure(volume, READ, e, begin);
+ eventHooks.onFailure(datanode, volume, READ, e, begin);
throw e;
}
}
@@ -845,7 +850,7 @@ public class FileIoProvider {
eventHooks.afterFileIo(volume, READ, begin, numBytesRead);
return numBytesRead;
} catch(Exception e) {
- eventHooks.onFailure(volume, READ, e, begin);
+ eventHooks.onFailure(datanode, volume, READ, e, begin);
throw e;
}
}
@@ -861,7 +866,7 @@ public class FileIoProvider {
eventHooks.afterFileIo(volume, READ, begin, numBytesRead);
return numBytesRead;
} catch(Exception e) {
- eventHooks.onFailure(volume, READ, e, begin);
+ eventHooks.onFailure(datanode, volume, READ, e, begin);
throw e;
}
}
@@ -903,7 +908,7 @@ public class FileIoProvider {
super.write(b);
eventHooks.afterFileIo(volume, WRITE, begin, 1);
} catch(Exception e) {
- eventHooks.onFailure(volume, WRITE, e, begin);
+ eventHooks.onFailure(datanode, volume, WRITE, e, begin);
throw e;
}
}
@@ -918,7 +923,7 @@ public class FileIoProvider {
super.write(b);
eventHooks.afterFileIo(volume, WRITE, begin, b.length);
} catch(Exception e) {
- eventHooks.onFailure(volume, WRITE, e, begin);
+ eventHooks.onFailure(datanode, volume, WRITE, e, begin);
throw e;
}
}
@@ -933,7 +938,7 @@ public class FileIoProvider {
super.write(b, off, len);
eventHooks.afterFileIo(volume, WRITE, begin, len);
} catch(Exception e) {
- eventHooks.onFailure(volume, WRITE, e, begin);
+ eventHooks.onFailure(datanode, volume, WRITE, e, begin);
throw e;
}
}
@@ -961,7 +966,7 @@ public class FileIoProvider {
eventHooks.afterFileIo(volume, READ, begin, 1);
return b;
} catch(Exception e) {
- eventHooks.onFailure(volume, READ, e, begin);
+ eventHooks.onFailure(datanode, volume, READ, e, begin);
throw e;
}
}
@@ -974,7 +979,7 @@ public class FileIoProvider {
eventHooks.afterFileIo(volume, READ, begin, numBytesRead);
return numBytesRead;
} catch(Exception e) {
- eventHooks.onFailure(volume, READ, e, begin);
+ eventHooks.onFailure(datanode, volume, READ, e, begin);
throw e;
}
}
@@ -987,7 +992,7 @@ public class FileIoProvider {
eventHooks.afterFileIo(volume, READ, begin, numBytesRead);
return numBytesRead;
} catch(Exception e) {
- eventHooks.onFailure(volume, READ, e, begin);
+ eventHooks.onFailure(datanode, volume, READ, e, begin);
throw e;
}
}
@@ -999,7 +1004,7 @@ public class FileIoProvider {
super.write(b);
eventHooks.afterFileIo(volume, WRITE, begin, 1);
} catch(Exception e) {
- eventHooks.onFailure(volume, WRITE, e, begin);
+ eventHooks.onFailure(datanode, volume, WRITE, e, begin);
throw e;
}
}
@@ -1011,7 +1016,7 @@ public class FileIoProvider {
super.write(b);
eventHooks.afterFileIo(volume, WRITE, begin, b.length);
} catch(Exception e) {
- eventHooks.onFailure(volume, WRITE, e, begin);
+ eventHooks.onFailure(datanode, volume, WRITE, e, begin);
throw e;
}
}
@@ -1023,7 +1028,7 @@ public class FileIoProvider {
super.write(b, off, len);
eventHooks.afterFileIo(volume, WRITE, begin, len);
} catch(Exception e) {
- eventHooks.onFailure(volume, WRITE, e, begin);
+ eventHooks.onFailure(datanode, volume, WRITE, e, begin);
throw e;
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eafaddca/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/ProfilingFileIoEvents.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/ProfilingFileIoEvents.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/ProfilingFileIoEvents.java
index 5835fe8..affd093 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/ProfilingFileIoEvents.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/ProfilingFileIoEvents.java
@@ -30,7 +30,7 @@ import javax.annotation.Nullable;
* related operations on datanode volumes.
*/
@InterfaceAudience.Private
-class ProfilingFileIoEvents implements FileIoEvents {
+class ProfilingFileIoEvents extends FileIoEvents {
@Override
public long beforeMetadataOp(@Nullable FsVolumeSpi volume,
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eafaddca/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/ReplicaInfo.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/ReplicaInfo.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/ReplicaInfo.java
index 47ca1f1..9817f97 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/ReplicaInfo.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/ReplicaInfo.java
@@ -64,7 +64,7 @@ abstract public class ReplicaInfo extends Block
/** This is used by some tests and FsDatasetUtil#computeChecksum. */
private static final FileIoProvider DEFAULT_FILE_IO_PROVIDER =
- new FileIoProvider(null);
+ new FileIoProvider(null, null);
/**
* Constructor
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eafaddca/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/AsyncChecker.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/AsyncChecker.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/AsyncChecker.java
index 1d534a3..997c0cb 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/AsyncChecker.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/AsyncChecker.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.hdfs.server.datanode.checker;
+import com.google.common.base.Optional;
import com.google.common.util.concurrent.ListenableFuture;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
@@ -43,10 +44,10 @@ public interface AsyncChecker<K, V> {
* @param context the interpretation of the context depends on the
* target.
*
- * @return returns a {@link ListenableFuture} that can be used to
+ * @return returns a {@link Optional of ListenableFuture} that can be used to
* retrieve the result of the asynchronous check.
*/
- ListenableFuture<V> schedule(Checkable<K, V> target, K context);
+ Optional<ListenableFuture<V>> schedule(Checkable<K, V> target, K context);
/**
* Cancel all executing checks and wait for them to complete.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eafaddca/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/DatasetVolumeChecker.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/DatasetVolumeChecker.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/DatasetVolumeChecker.java
index 8f346dc..5ef3eec 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/DatasetVolumeChecker.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/DatasetVolumeChecker.java
@@ -19,6 +19,7 @@
package org.apache.hadoop.hdfs.server.datanode.checker;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.FutureCallback;
@@ -191,18 +192,26 @@ public class DatasetVolumeChecker {
for (int i = 0; i < references.size(); ++i) {
final FsVolumeReference reference = references.getReference(i);
- allVolumes.add(reference.getVolume());
- ListenableFuture<VolumeCheckResult> future =
+ Optional<ListenableFuture<VolumeCheckResult>> olf =
delegateChecker.schedule(reference.getVolume(), IGNORED_CONTEXT);
LOG.info("Scheduled health check for volume {}", reference.getVolume());
- Futures.addCallback(future, new ResultHandler(
- reference, healthyVolumes, failedVolumes, numVolumes, new Callback() {
- @Override
- public void call(Set<FsVolumeSpi> ignored1,
- Set<FsVolumeSpi> ignored2) {
+ if (olf.isPresent()) {
+ allVolumes.add(reference.getVolume());
+ Futures.addCallback(olf.get(),
+ new ResultHandler(reference, healthyVolumes, failedVolumes,
+ numVolumes, new Callback() {
+ @Override
+ public void call(Set<FsVolumeSpi> ignored1,
+ Set<FsVolumeSpi> ignored2) {
+ latch.countDown();
+ }
+ }));
+ } else {
+ IOUtils.cleanup(null, reference);
+ if (numVolumes.decrementAndGet() == 0) {
latch.countDown();
}
- }));
+ }
}
// Wait until our timeout elapses, after which we give up on
@@ -263,18 +272,26 @@ public class DatasetVolumeChecker {
final Set<FsVolumeSpi> healthyVolumes = new HashSet<>();
final Set<FsVolumeSpi> failedVolumes = new HashSet<>();
final AtomicLong numVolumes = new AtomicLong(references.size());
+ boolean added = false;
LOG.info("Checking {} volumes", references.size());
for (int i = 0; i < references.size(); ++i) {
final FsVolumeReference reference = references.getReference(i);
// The context parameter is currently ignored.
- ListenableFuture<VolumeCheckResult> future =
+ Optional<ListenableFuture<VolumeCheckResult>> olf =
delegateChecker.schedule(reference.getVolume(), IGNORED_CONTEXT);
- Futures.addCallback(future, new ResultHandler(
- reference, healthyVolumes, failedVolumes, numVolumes, callback));
+ if (olf.isPresent()) {
+ added = true;
+ Futures.addCallback(olf.get(),
+ new ResultHandler(reference, healthyVolumes, failedVolumes,
+ numVolumes, callback));
+ } else {
+ IOUtils.cleanup(null, reference);
+ numVolumes.decrementAndGet();
+ }
}
numAsyncDatasetChecks.incrementAndGet();
- return true;
+ return added;
}
/**
@@ -291,7 +308,7 @@ public class DatasetVolumeChecker {
}
/**
- * Check a single volume, returning a {@link ListenableFuture}
+ * Check a single volume asynchronously, returning a {@link ListenableFuture}
* that can be used to retrieve the final result.
*
* If the volume cannot be referenced then it is already closed and
@@ -305,21 +322,32 @@ public class DatasetVolumeChecker {
public boolean checkVolume(
final FsVolumeSpi volume,
Callback callback) {
+ if (volume == null) {
+ LOG.debug("Cannot schedule check on null volume");
+ return false;
+ }
+
FsVolumeReference volumeReference;
try {
volumeReference = volume.obtainReference();
} catch (ClosedChannelException e) {
// The volume has already been closed.
- callback.call(new HashSet<FsVolumeSpi>(), new HashSet<FsVolumeSpi>());
return false;
}
- ListenableFuture<VolumeCheckResult> future =
+
+ Optional<ListenableFuture<VolumeCheckResult>> olf =
delegateChecker.schedule(volume, IGNORED_CONTEXT);
- numVolumeChecks.incrementAndGet();
- Futures.addCallback(future, new ResultHandler(
- volumeReference, new HashSet<FsVolumeSpi>(), new HashSet<FsVolumeSpi>(),
- new AtomicLong(1), callback));
- return true;
+ if (olf.isPresent()) {
+ numVolumeChecks.incrementAndGet();
+ Futures.addCallback(olf.get(),
+ new ResultHandler(volumeReference, new HashSet<FsVolumeSpi>(),
+ new HashSet<FsVolumeSpi>(),
+ new AtomicLong(1), callback));
+ return true;
+ } else {
+ IOUtils.cleanup(null, volumeReference);
+ }
+ return false;
}
/**
@@ -343,8 +371,8 @@ public class DatasetVolumeChecker {
* successful, add the volume here.
* @param failedVolumes set of failed volumes. If the disk check fails,
* add the volume here.
- * @param semaphore semaphore used to trigger callback invocation.
- * @param callback invoked when the semaphore can be successfully acquired.
+ * @param volumeCounter volumeCounter used to trigger callback invocation.
+ * @param callback invoked when the volumeCounter reaches 0.
*/
ResultHandler(FsVolumeReference reference,
Set<FsVolumeSpi> healthyVolumes,
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eafaddca/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/StorageLocationChecker.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/StorageLocationChecker.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/StorageLocationChecker.java
index 7337ad0..6e323e0 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/StorageLocationChecker.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/StorageLocationChecker.java
@@ -20,6 +20,7 @@ package org.apache.hadoop.hdfs.server.datanode.checker;
import static org.apache.hadoop.hdfs.DFSConfigKeys.*;
+import com.google.common.base.Optional;
import com.google.common.collect.Maps;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
@@ -158,8 +159,11 @@ public class StorageLocationChecker {
// Start parallel disk check operations on all StorageLocations.
for (StorageLocation location : dataDirs) {
goodLocations.put(location, true);
- futures.put(location,
- delegateChecker.schedule(location, context));
+ Optional<ListenableFuture<VolumeCheckResult>> olf =
+ delegateChecker.schedule(location, context);
+ if (olf.isPresent()) {
+ futures.put(location, olf.get());
+ }
}
if (maxVolumeFailuresTolerated >= dataDirs.size()) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eafaddca/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/ThrottledAsyncChecker.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/ThrottledAsyncChecker.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/ThrottledAsyncChecker.java
index 89db88e..ad56c61 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/ThrottledAsyncChecker.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/ThrottledAsyncChecker.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.hdfs.server.datanode.checker;
+import com.google.common.base.Optional;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
@@ -101,13 +102,11 @@ public class ThrottledAsyncChecker<K, V> implements AsyncChecker<K, V> {
* will receive the same Future.
*/
@Override
- public synchronized ListenableFuture<V> schedule(
- final Checkable<K, V> target,
- final K context) {
- LOG.debug("Scheduling a check of {}", target);
-
+ public Optional<ListenableFuture<V>> schedule(
+ final Checkable<K, V> target, final K context) {
+ LOG.info("Scheduling a check for {}", target);
if (checksInProgress.containsKey(target)) {
- return checksInProgress.get(target);
+ return Optional.absent();
}
if (completedChecks.containsKey(target)) {
@@ -115,11 +114,9 @@ public class ThrottledAsyncChecker<K, V> implements AsyncChecker<K, V> {
final long msSinceLastCheck = timer.monotonicNow() - result.completedAt;
if (msSinceLastCheck < minMsBetweenChecks) {
LOG.debug("Skipped checking {}. Time since last check {}ms " +
- "is less than the min gap {}ms.",
+ "is less than the min gap {}ms.",
target, msSinceLastCheck, minMsBetweenChecks);
- return result.result != null ?
- Futures.immediateFuture(result.result) :
- Futures.<V>immediateFailedFuture(result.exception);
+ return Optional.absent();
}
}
@@ -132,7 +129,7 @@ public class ThrottledAsyncChecker<K, V> implements AsyncChecker<K, V> {
});
checksInProgress.put(target, lf);
addResultCachingCallback(target, lf);
- return lf;
+ return Optional.of(lf);
}
/**
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eafaddca/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/BlockPoolSlice.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/BlockPoolSlice.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/BlockPoolSlice.java
index cd44557..31cf39f 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/BlockPoolSlice.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/BlockPoolSlice.java
@@ -277,7 +277,10 @@ class BlockPoolSlice {
fileIoProvider.getFileOutputStream(volume, outFile), "UTF-8")) {
// mtime is written last, so that truncated writes won't be valid.
out.write(Long.toString(used) + " " + Long.toString(timer.now()));
- fileIoProvider.flush(volume, out);
+ // This is only called as part of the volume shutdown.
+ // We explicitly avoid calling flush with fileIoProvider which triggers
+ // volume check upon io exception to avoid cyclic volume checks.
+ out.flush();
}
} catch (IOException ioe) {
// If write failed, the volume might be bad. Since the cache file is
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eafaddca/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 e37934e..cc1d0c1 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
@@ -1974,17 +1974,22 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
*/
File validateBlockFile(String bpid, long blockId) {
//Should we check for metadata file too?
- final File f;
+ File f = null;
+ ReplicaInfo info;
try(AutoCloseableLock lock = datasetLock.acquire()) {
- f = getFile(bpid, blockId, false);
+ info = volumeMap.get(bpid, blockId);
+ if (info != null) {
+ f = info.getBlockFile();
+ }
}
if(f != null ) {
- if(f.exists())
+ if(f.exists()) {
return f;
+ }
// if file is not null, but doesn't exist - possibly disk failed
- datanode.checkDiskErrorAsync();
+ datanode.checkDiskErrorAsync(info.getVolume());
}
if (LOG.isDebugEnabled()) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eafaddca/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeImpl.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeImpl.java
index b1ce931..71d93ae 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeImpl.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeImpl.java
@@ -79,7 +79,7 @@ import com.google.common.util.concurrent.ThreadFactoryBuilder;
/**
* The underlying volume used to store replica.
- *
+ *
* It uses the {@link FsDatasetImpl} object for synchronization.
*/
@InterfaceAudience.Private
@@ -98,7 +98,7 @@ public class FsVolumeImpl implements FsVolumeSpi {
private final Map<String, BlockPoolSlice> bpSlices
= new ConcurrentHashMap<String, BlockPoolSlice>();
private final File currentDir; // <StorageDirectory>/current
- private final DF usage;
+ private final DF usage;
private final long reserved;
private CloseableReferenceCount reference = new CloseableReferenceCount();
@@ -120,7 +120,7 @@ public class FsVolumeImpl implements FsVolumeSpi {
* contention.
*/
protected ThreadPoolExecutor cacheExecutor;
-
+
FsVolumeImpl(FsDatasetImpl dataset, String storageID, File currentDir,
Configuration conf, StorageType storageType) throws IOException {
this.dataset = dataset;
@@ -137,7 +137,8 @@ public class FsVolumeImpl implements FsVolumeSpi {
this.configuredCapacity = -1;
// dataset.datanode may be null in some tests.
this.fileIoProvider = dataset.datanode != null ?
- dataset.datanode.getFileIoProvider() : new FileIoProvider(conf);
+ dataset.datanode.getFileIoProvider() :
+ new FileIoProvider(conf, dataset.datanode);
cacheExecutor = initializeCacheExecutor(parent);
this.metrics = DataNodeVolumeMetrics.create(conf, parent.getAbsolutePath());
}
@@ -288,7 +289,7 @@ public class FsVolumeImpl implements FsVolumeSpi {
File getCurrentDir() {
return currentDir;
}
-
+
File getRbwDir(String bpid) throws IOException {
return getBlockPoolSlice(bpid).getRbwDir();
}
@@ -358,11 +359,11 @@ public class FsVolumeImpl implements FsVolumeSpi {
long getBlockPoolUsed(String bpid) throws IOException {
return getBlockPoolSlice(bpid).getDfsUsed();
}
-
+
/**
* Return either the configured capacity of the file system if configured; or
* the capacity of the file system excluding space reserved for non-HDFS.
- *
+ *
* @return the unreserved number of bytes left in this filesystem. May be
* zero.
*/
@@ -389,7 +390,7 @@ public class FsVolumeImpl implements FsVolumeSpi {
/*
* Calculate the available space of the filesystem, excluding space reserved
* for non-HDFS and space reserved for RBW
- *
+ *
* @return the available number of bytes left in this filesystem. May be zero.
*/
@Override
@@ -460,7 +461,7 @@ public class FsVolumeImpl implements FsVolumeSpi {
public String getBasePath() {
return currentDir.getParent();
}
-
+
@Override
public boolean isTransientStorage() {
return storageType.isTransient();
@@ -481,9 +482,9 @@ public class FsVolumeImpl implements FsVolumeSpi {
*/
@Override
public String[] getBlockPoolList() {
- return bpSlices.keySet().toArray(new String[bpSlices.keySet().size()]);
+ return bpSlices.keySet().toArray(new String[bpSlices.keySet().size()]);
}
-
+
/**
* Temporary files. They get moved to the finalized block directory when
* the block is finalized.
@@ -692,7 +693,7 @@ public class FsVolumeImpl implements FsVolumeSpi {
LOG.trace("getSubdirEntries({}, {}): no entries found in {}",
storageID, bpid, dir.getAbsolutePath());
} else {
- LOG.trace("getSubdirEntries({}, {}): listed {} entries in {}",
+ LOG.trace("getSubdirEntries({}, {}): listed {} entries in {}",
storageID, bpid, entries.size(), dir.getAbsolutePath());
}
cache = entries;
@@ -910,7 +911,7 @@ public class FsVolumeImpl implements FsVolumeSpi {
}
return VolumeCheckResult.HEALTHY;
}
-
+
void getVolumeMap(ReplicaMap volumeMap,
final RamDiskReplicaTracker ramDiskReplicaMap)
throws IOException {
@@ -918,7 +919,7 @@ public class FsVolumeImpl implements FsVolumeSpi {
s.getVolumeMap(volumeMap, ramDiskReplicaMap);
}
}
-
+
void getVolumeMap(String bpid, ReplicaMap volumeMap,
final RamDiskReplicaTracker ramDiskReplicaMap)
throws IOException {
@@ -966,7 +967,7 @@ public class FsVolumeImpl implements FsVolumeSpi {
}
bpSlices.put(bpid, bp);
}
-
+
void shutdownBlockPool(String bpid, BlockListAsLongs blocksListsAsLongs) {
BlockPoolSlice bp = bpSlices.get(bpid);
if (bp != null) {
@@ -992,7 +993,7 @@ public class FsVolumeImpl implements FsVolumeSpi {
}
return true;
}
-
+
void deleteBPDirectories(String bpid, boolean force) throws IOException {
File volumeCurrentDir = this.getCurrentDir();
File bpDir = new File(volumeCurrentDir, bpid);
@@ -1000,7 +1001,7 @@ public class FsVolumeImpl implements FsVolumeSpi {
// nothing to be deleted
return;
}
- File tmpDir = new File(bpDir, DataStorage.STORAGE_DIR_TMP);
+ File tmpDir = new File(bpDir, DataStorage.STORAGE_DIR_TMP);
File bpCurrentDir = new File(bpDir, DataStorage.STORAGE_DIR_CURRENT);
File finalizedDir = new File(bpCurrentDir,
DataStorage.STORAGE_DIR_FINALIZED);
@@ -1049,12 +1050,12 @@ public class FsVolumeImpl implements FsVolumeSpi {
public String getStorageID() {
return storageID;
}
-
+
@Override
public StorageType getStorageType() {
return storageType;
}
-
+
DatanodeStorage toDatanodeStorage() {
return new DatanodeStorage(storageID, DatanodeStorage.State.NORMAL, storageType);
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eafaddca/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 db5042b..dd2ca99 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
@@ -589,7 +589,7 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
}
registerMBean(datanodeUuid);
- this.fileIoProvider = new FileIoProvider(conf);
+ this.fileIoProvider = new FileIoProvider(conf, datanode);
this.storage = new SimulatedStorage(
conf.getLong(CONFIG_PROPERTY_CAPACITY, DEFAULT_CAPACITY),
conf.getEnum(CONFIG_PROPERTY_STATE, DEFAULT_STATE));
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eafaddca/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 25f4d5a..600769b 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
@@ -942,7 +942,7 @@ public class TestDataNodeHotSwapVolumes {
DataNodeTestUtils.injectDataDirFailure(dirToFail);
// Call and wait DataNode to detect disk failure.
long lastDiskErrorCheck = dn.getLastDiskErrorCheck();
- dn.checkDiskErrorAsync();
+ dn.checkDiskErrorAsync(failedVolume);
while (dn.getLastDiskErrorCheck() == lastDiskErrorCheck) {
Thread.sleep(100);
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eafaddca/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailureReporting.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailureReporting.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailureReporting.java
index 23ec8a7..79a52bb 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailureReporting.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailureReporting.java
@@ -17,7 +17,7 @@
*/
package org.apache.hadoop.hdfs.server.datanode;
-import static org.apache.hadoop.test.MetricsAsserts.assertCounter;
+import static org.apache.hadoop.test.MetricsAsserts.getLongCounter;
import static org.apache.hadoop.test.MetricsAsserts.getMetrics;
import static org.hamcrest.core.Is.is;
import static org.junit.Assert.assertArrayEquals;
@@ -76,7 +76,7 @@ public class TestDataNodeVolumeFailureReporting {
final int WAIT_FOR_HEARTBEATS = 3000;
// Wait at least (2 * re-check + 10 * heartbeat) seconds for
- // a datanode to be considered dead by the namenode.
+ // a datanode to be considered dead by the namenode.
final int WAIT_FOR_DEATH = 15000;
@Before
@@ -158,7 +158,7 @@ public class TestDataNodeVolumeFailureReporting {
assert (WAIT_FOR_HEARTBEATS * 10) > WAIT_FOR_DEATH;
// Eventually the NN should report two volume failures
- DFSTestUtil.waitForDatanodeStatus(dm, 3, 0, 2,
+ DFSTestUtil.waitForDatanodeStatus(dm, 3, 0, 2,
origCapacity - (1*dnCapacity), WAIT_FOR_HEARTBEATS);
checkAggregateFailuresAtNameNode(true, 2);
checkFailuresAtNameNode(dm, dns.get(0), true, dn1Vol1.getAbsolutePath());
@@ -185,7 +185,7 @@ public class TestDataNodeVolumeFailureReporting {
* did not grow or shrink the data volume while the test was running).
*/
dnCapacity = DFSTestUtil.getDatanodeCapacity(dm, 0);
- DFSTestUtil.waitForDatanodeStatus(dm, 3, 0, 3,
+ DFSTestUtil.waitForDatanodeStatus(dm, 3, 0, 3,
origCapacity - (3*dnCapacity), WAIT_FOR_HEARTBEATS);
checkAggregateFailuresAtNameNode(true, 3);
checkFailuresAtNameNode(dm, dns.get(0), true, dn1Vol1.getAbsolutePath());
@@ -211,7 +211,7 @@ public class TestDataNodeVolumeFailureReporting {
dn3Vol2.getAbsolutePath());
// The NN considers the DN dead
- DFSTestUtil.waitForDatanodeStatus(dm, 2, 1, 2,
+ DFSTestUtil.waitForDatanodeStatus(dm, 2, 1, 2,
origCapacity - (4*dnCapacity), WAIT_FOR_HEARTBEATS);
checkAggregateFailuresAtNameNode(true, 2);
checkFailuresAtNameNode(dm, dns.get(0), true, dn1Vol1.getAbsolutePath());
@@ -236,7 +236,7 @@ public class TestDataNodeVolumeFailureReporting {
* and that the volume failure count should be reported as zero by
* both the metrics and the NN.
*/
- DFSTestUtil.waitForDatanodeStatus(dm, 3, 0, 0, origCapacity,
+ DFSTestUtil.waitForDatanodeStatus(dm, 3, 0, 0, origCapacity,
WAIT_FOR_HEARTBEATS);
checkAggregateFailuresAtNameNode(true, 0);
dns = cluster.getDataNodes();
@@ -259,8 +259,8 @@ public class TestDataNodeVolumeFailureReporting {
long origCapacity = DFSTestUtil.getLiveDatanodeCapacity(dm);
long dnCapacity = DFSTestUtil.getDatanodeCapacity(dm, 0);
- // Fail the first volume on both datanodes (we have to keep the
- // third healthy so one node in the pipeline will not fail).
+ // Fail the first volume on both datanodes (we have to keep the
+ // third healthy so one node in the pipeline will not fail).
File dn1Vol1 = new File(dataDir, "data"+(2*0+1));
File dn2Vol1 = new File(dataDir, "data"+(2*1+1));
DataNodeTestUtils.injectDataDirFailure(dn1Vol1, dn2Vol1);
@@ -271,7 +271,7 @@ public class TestDataNodeVolumeFailureReporting {
ArrayList<DataNode> dns = cluster.getDataNodes();
// The NN reports two volumes failures
- DFSTestUtil.waitForDatanodeStatus(dm, 3, 0, 2,
+ DFSTestUtil.waitForDatanodeStatus(dm, 3, 0, 2,
origCapacity - (1*dnCapacity), WAIT_FOR_HEARTBEATS);
checkAggregateFailuresAtNameNode(true, 2);
checkFailuresAtNameNode(dm, dns.get(0), true, dn1Vol1.getAbsolutePath());
@@ -318,6 +318,12 @@ public class TestDataNodeVolumeFailureReporting {
DFSTestUtil.createFile(fs, file1, 1024, (short)3, 1L);
DFSTestUtil.waitReplication(fs, file1, (short)3);
+ // Create additional file to trigger failure based volume check on dn1Vol2
+ // and dn2Vol2.
+ Path file2 = new Path("/test2");
+ DFSTestUtil.createFile(fs, file2, 1024, (short)3, 1L);
+ DFSTestUtil.waitReplication(fs, file2, (short)3);
+
ArrayList<DataNode> dns = cluster.getDataNodes();
assertTrue("DN1 should be up", dns.get(0).isDatanodeUp());
assertTrue("DN2 should be up", dns.get(1).isDatanodeUp());
@@ -536,8 +542,6 @@ public class TestDataNodeVolumeFailureReporting {
private void checkFailuresAtDataNode(DataNode dn,
long expectedVolumeFailuresCounter, boolean expectCapacityKnown,
String... expectedFailedVolumes) throws Exception {
- assertCounter("VolumeFailures", expectedVolumeFailuresCounter,
- getMetrics(dn.getMetrics().name()));
FsDatasetSpi<?> fsd = dn.getFSDataset();
assertEquals(expectedFailedVolumes.length, fsd.getNumFailedVolumes());
assertArrayEquals(expectedFailedVolumes, fsd.getFailedStorageLocations());
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eafaddca/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestDatasetVolumeChecker.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestDatasetVolumeChecker.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestDatasetVolumeChecker.java
index 940e73b..d6b4af9 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestDatasetVolumeChecker.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestDatasetVolumeChecker.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.hdfs.server.datanode.checker;
+import com.google.common.base.Optional;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import org.apache.hadoop.hdfs.HdfsConfiguration;
@@ -103,24 +104,28 @@ public class TestDatasetVolumeChecker {
/**
* Request a check and ensure it triggered {@link FsVolumeSpi#check}.
*/
- checker.checkVolume(volume, new DatasetVolumeChecker.Callback() {
- @Override
- public void call(Set<FsVolumeSpi> healthyVolumes,
- Set<FsVolumeSpi> failedVolumes) {
- numCallbackInvocations.incrementAndGet();
- if (expectedVolumeHealth != null && expectedVolumeHealth != FAILED) {
- assertThat(healthyVolumes.size(), is(1));
- assertThat(failedVolumes.size(), is(0));
- } else {
- assertThat(healthyVolumes.size(), is(0));
- assertThat(failedVolumes.size(), is(1));
- }
- }
- });
+ boolean result =
+ checker.checkVolume(volume, new DatasetVolumeChecker.Callback() {
+ @Override
+ public void call(Set<FsVolumeSpi> healthyVolumes,
+ Set<FsVolumeSpi> failedVolumes) {
+ numCallbackInvocations.incrementAndGet();
+ if (expectedVolumeHealth != null &&
+ expectedVolumeHealth != FAILED) {
+ assertThat(healthyVolumes.size(), is(1));
+ assertThat(failedVolumes.size(), is(0));
+ } else {
+ assertThat(healthyVolumes.size(), is(0));
+ assertThat(failedVolumes.size(), is(1));
+ }
+ }
+ });
// Ensure that the check was invoked at least once.
verify(volume, times(1)).check(any(VolumeCheckContext.class));
- assertThat(numCallbackInvocations.get(), is(1L));
+ if (result) {
+ assertThat(numCallbackInvocations.get(), is(1L));
+ }
}
/**
@@ -172,7 +177,7 @@ public class TestDatasetVolumeChecker {
checker.setDelegateChecker(new DummyChecker());
final AtomicLong numCallbackInvocations = new AtomicLong(0);
- checker.checkAllVolumesAsync(
+ boolean result = checker.checkAllVolumesAsync(
dataset, new DatasetVolumeChecker.Callback() {
@Override
public void call(
@@ -192,7 +197,9 @@ public class TestDatasetVolumeChecker {
});
// The callback should be invoked exactly once.
- assertThat(numCallbackInvocations.get(), is(1L));
+ if (result) {
+ assertThat(numCallbackInvocations.get(), is(1L));
+ }
// Ensure each volume's check() method was called exactly once.
for (FsVolumeSpi volume : volumes) {
@@ -206,15 +213,18 @@ public class TestDatasetVolumeChecker {
*/
static class DummyChecker
implements AsyncChecker<VolumeCheckContext, VolumeCheckResult> {
+
@Override
- public ListenableFuture<VolumeCheckResult> schedule(
+ public Optional<ListenableFuture<VolumeCheckResult>> schedule(
Checkable<VolumeCheckContext, VolumeCheckResult> target,
VolumeCheckContext context) {
try {
- return Futures.immediateFuture(target.check(context));
+ return Optional.of(
+ Futures.immediateFuture(target.check(context)));
} catch (Exception e) {
LOG.info("check routine threw exception " + e);
- return Futures.immediateFailedFuture(e);
+ return Optional.of(
+ Futures.<VolumeCheckResult>immediateFailedFuture(e));
}
}
@@ -259,4 +269,4 @@ public class TestDatasetVolumeChecker {
}
return volumes;
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eafaddca/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestDatasetVolumeCheckerFailures.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestDatasetVolumeCheckerFailures.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestDatasetVolumeCheckerFailures.java
index 6f3d748..04d8bcc 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestDatasetVolumeCheckerFailures.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestDatasetVolumeCheckerFailures.java
@@ -131,29 +131,6 @@ public class TestDatasetVolumeCheckerFailures {
assertThat(checker.getNumSkippedChecks(), is(1L));
}
- @Test(timeout=60000)
- public void testMinGapIsEnforcedForASyncChecks() throws Exception {
- final List<FsVolumeSpi> volumes =
- TestDatasetVolumeChecker.makeVolumes(1, VolumeCheckResult.HEALTHY);
- final FsDatasetSpi<FsVolumeSpi> dataset =
- TestDatasetVolumeChecker.makeDataset(volumes);
- final DatasetVolumeChecker checker = new DatasetVolumeChecker(conf, timer);
-
- checker.checkAllVolumesAsync(dataset, null);
- assertThat(checker.getNumAsyncDatasetChecks(), is(1L));
-
- // Re-check without advancing the timer. Ensure the check is skipped.
- checker.checkAllVolumesAsync(dataset, null);
- assertThat(checker.getNumAsyncDatasetChecks(), is(1L));
- assertThat(checker.getNumSkippedChecks(), is(1L));
-
- // Re-check after advancing the timer. Ensure the check is performed.
- timer.advance(MIN_DISK_CHECK_GAP_MS);
- checker.checkAllVolumesAsync(dataset, null);
- assertThat(checker.getNumAsyncDatasetChecks(), is(2L));
- assertThat(checker.getNumSkippedChecks(), is(1L));
- }
-
/**
* Create a mock FsVolumeSpi whose {@link FsVolumeSpi#check} routine
* hangs forever.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eafaddca/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestThrottledAsyncChecker.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestThrottledAsyncChecker.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestThrottledAsyncChecker.java
index 30bdc08..1109772 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestThrottledAsyncChecker.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestThrottledAsyncChecker.java
@@ -18,15 +18,14 @@
package org.apache.hadoop.hdfs.server.datanode.checker;
+import com.google.common.base.Optional;
import com.google.common.base.Supplier;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import org.apache.hadoop.test.GenericTestUtils;
import org.apache.hadoop.util.FakeTimer;
-import org.junit.Rule;
import org.junit.Test;
-import org.junit.rules.ExpectedException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -38,10 +37,7 @@ import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
-import static org.hamcrest.CoreMatchers.is;
-import static org.hamcrest.core.Is.isA;
import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -53,9 +49,6 @@ public class TestThrottledAsyncChecker {
LoggerFactory.getLogger(TestThrottledAsyncChecker.class);
private static final long MIN_ERROR_CHECK_GAP = 1000;
- @Rule
- public ExpectedException thrown = ExpectedException.none();
-
/**
* Test various scheduling combinations to ensure scheduling and
* throttling behave as expected.
@@ -70,34 +63,34 @@ public class TestThrottledAsyncChecker {
getExecutorService());
// check target1 and ensure we get back the expected result.
- assertTrue(checker.schedule(target1, true).get());
- assertThat(target1.numChecks.get(), is(1L));
+ assertTrue(checker.schedule(target1, true).isPresent());
+ waitTestCheckableCheckCount(target1, 1L);
// Check target1 again without advancing the timer. target1 should not
- // be checked again and the cached result should be returned.
- assertTrue(checker.schedule(target1, true).get());
- assertThat(target1.numChecks.get(), is(1L));
+ // be checked again.
+ assertFalse(checker.schedule(target1, true).isPresent());
+ waitTestCheckableCheckCount(target1, 1L);
// Schedule target2 scheduled without advancing the timer.
// target2 should be checked as it has never been checked before.
- assertTrue(checker.schedule(target2, true).get());
- assertThat(target2.numChecks.get(), is(1L));
+ assertTrue(checker.schedule(target2, true).isPresent());
+ waitTestCheckableCheckCount(target2, 1L);
// Advance the timer but just short of the min gap.
// Neither target1 nor target2 should be checked again.
timer.advance(MIN_ERROR_CHECK_GAP - 1);
- assertTrue(checker.schedule(target1, true).get());
- assertThat(target1.numChecks.get(), is(1L));
- assertTrue(checker.schedule(target2, true).get());
- assertThat(target2.numChecks.get(), is(1L));
+ assertFalse(checker.schedule(target1, true).isPresent());
+ waitTestCheckableCheckCount(target1, 1L);
+ assertFalse(checker.schedule(target2, true).isPresent());
+ waitTestCheckableCheckCount(target2, 1L);
// Advance the timer again.
// Both targets should be checked now.
timer.advance(MIN_ERROR_CHECK_GAP);
- assertTrue(checker.schedule(target1, true).get());
- assertThat(target1.numChecks.get(), is(2L));
- assertTrue(checker.schedule(target2, true).get());
- assertThat(target1.numChecks.get(), is(2L));
+ assertTrue(checker.schedule(target1, true).isPresent());
+ waitTestCheckableCheckCount(target1, 2L);
+ assertTrue(checker.schedule(target2, true).isPresent());
+ waitTestCheckableCheckCount(target2, 2L);
}
@Test (timeout=60000)
@@ -109,13 +102,16 @@ public class TestThrottledAsyncChecker {
new ThrottledAsyncChecker<>(timer, MIN_ERROR_CHECK_GAP,
getExecutorService());
- ListenableFuture<Boolean> lf = checker.schedule(target, true);
- Futures.addCallback(lf, callback);
+ Optional<ListenableFuture<Boolean>> olf =
+ checker.schedule(target, true);
+ if (olf.isPresent()) {
+ Futures.addCallback(olf.get(), callback);
+ }
// Request immediate cancellation.
checker.shutdownAndWait(0, TimeUnit.MILLISECONDS);
try {
- assertFalse(lf.get());
+ assertFalse(olf.get().get());
fail("Failed to get expected InterruptedException");
} catch (ExecutionException ee) {
assertTrue(ee.getCause() instanceof InterruptedException);
@@ -130,27 +126,33 @@ public class TestThrottledAsyncChecker {
final ThrottledAsyncChecker<Boolean, Boolean> checker =
new ThrottledAsyncChecker<>(timer, MIN_ERROR_CHECK_GAP,
getExecutorService());
- final ListenableFuture<Boolean> lf1 = checker.schedule(target, true);
- final ListenableFuture<Boolean> lf2 = checker.schedule(target, true);
+ final Optional<ListenableFuture<Boolean>> olf1 =
+ checker.schedule(target, true);
- // Ensure that concurrent requests return the same future object.
- assertTrue(lf1 == lf2);
+ final Optional<ListenableFuture<Boolean>> olf2 =
+ checker.schedule(target, true);
+
+ // Ensure that concurrent requests return the future object
+ // for the first caller.
+ assertTrue(olf1.isPresent());
+ assertFalse(olf2.isPresent());
// Unblock the latch and wait for it to finish execution.
target.latch.countDown();
- lf1.get();
+ olf1.get().get();
GenericTestUtils.waitFor(new Supplier<Boolean>() {
@Override
public Boolean get() {
- // We should not get back the same future as before.
+ // We should get an absent Optional.
// This can take a short while until the internal callback in
// ThrottledAsyncChecker is scheduled for execution.
// Also this should not trigger a new check operation as the timer
// was not advanced. If it does trigger a new check then the test
// will fail with a timeout.
- final ListenableFuture<Boolean> lf3 = checker.schedule(target, true);
- return lf3 != lf2;
+ final Optional<ListenableFuture<Boolean>> olf3 =
+ checker.schedule(target, true);
+ return !olf3.isPresent();
}
}, 100, 10000);
}
@@ -168,15 +170,30 @@ public class TestThrottledAsyncChecker {
new ThrottledAsyncChecker<>(timer, MIN_ERROR_CHECK_GAP,
getExecutorService());
- assertTrue(checker.schedule(target1, true).get());
- assertThat(target1.numChecks.get(), is(1L));
+ assertTrue(checker.schedule(target1, true).isPresent());
+ waitTestCheckableCheckCount(target1, 1L);
timer.advance(MIN_ERROR_CHECK_GAP + 1);
- assertFalse(checker.schedule(target1, false).get());
- assertThat(target1.numChecks.get(), is(2L));
+ assertTrue(checker.schedule(target1, false).isPresent());
+ waitTestCheckableCheckCount(target1, 2L);
+
}
+ private void waitTestCheckableCheckCount(
+ final TestCheckableBase target,
+ final long expectedChecks) throws Exception {
+ GenericTestUtils.waitFor(new Supplier<Boolean>() {
+ @Override
+ public Boolean get() {
+ // This can take a short while until the internal callback in
+ // ThrottledAsyncChecker is scheduled for execution.
+ // If it does trigger a new check then the test
+ // will fail with a timeout.
+ return target.getTotalChecks() == expectedChecks;
+ }
+ }, 100, 10000);
+ }
/**
- * Ensure that the exeption from a failed check is cached
+ * Ensure that the exception from a failed check is cached
* and returned without re-running the check when the minimum
* gap has not elapsed.
*
@@ -190,13 +207,11 @@ public class TestThrottledAsyncChecker {
new ThrottledAsyncChecker<>(timer, MIN_ERROR_CHECK_GAP,
getExecutorService());
- thrown.expectCause(isA(DummyException.class));
- checker.schedule(target1, true).get();
- assertThat(target1.numChecks.get(), is(1L));
+ assertTrue(checker.schedule(target1, true).isPresent());
+ waitTestCheckableCheckCount(target1, 1L);
- thrown.expectCause(isA(DummyException.class));
- checker.schedule(target1, true).get();
- assertThat(target1.numChecks.get(), is(2L));
+ assertFalse(checker.schedule(target1, true).isPresent());
+ waitTestCheckableCheckCount(target1, 1L);
}
/**
@@ -206,28 +221,38 @@ public class TestThrottledAsyncChecker {
return new ScheduledThreadPoolExecutor(1);
}
+ private abstract static class TestCheckableBase
+ implements Checkable<Boolean, Boolean> {
+ protected final AtomicLong numChecks = new AtomicLong(0);
+
+ public long getTotalChecks() {
+ return numChecks.get();
+ }
+
+ public void incrTotalChecks() {
+ numChecks.incrementAndGet();
+ }
+ }
+
/**
* A Checkable that just returns its input.
*/
private static class NoOpCheckable
- implements Checkable<Boolean, Boolean> {
- private final AtomicLong numChecks = new AtomicLong(0);
+ extends TestCheckableBase {
@Override
public Boolean check(Boolean context) {
- numChecks.incrementAndGet();
+ incrTotalChecks();
return context;
}
}
private static class ThrowingCheckable
- implements Checkable<Boolean, Boolean> {
- private final AtomicLong numChecks = new AtomicLong(0);
+ extends TestCheckableBase {
@Override
public Boolean check(Boolean context) throws DummyException {
- numChecks.incrementAndGet();
+ incrTotalChecks();
throw new DummyException();
}
-
}
private static class DummyException extends Exception {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eafaddca/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 10b2f56..77b5258 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
@@ -159,7 +159,7 @@ public class TestFsDatasetImpl {
this.conf = new Configuration();
this.conf.setLong(DFS_DATANODE_SCAN_PERIOD_HOURS_KEY, 0);
- final FileIoProvider fileIoProvider = new FileIoProvider(conf);
+ final FileIoProvider fileIoProvider = new FileIoProvider(conf, null);
when(datanode.getFileIoProvider()).thenReturn(fileIoProvider);
when(datanode.getConf()).thenReturn(conf);
final DNConf dnConf = new DNConf(datanode);
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[5/5] hadoop git commit: HDFS-11279. Cleanup unused
DataNode#checkDiskErrorAsync(). Contributed by Hanisha Koneru.
Posted by ar...@apache.org.
HDFS-11279. Cleanup unused DataNode#checkDiskErrorAsync(). Contributed by Hanisha Koneru.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/8ec9dca2
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/8ec9dca2
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/8ec9dca2
Branch: refs/heads/branch-2
Commit: 8ec9dca2e112e08b480043da5766bf28a9231f49
Parents: eafaddc
Author: Arpit Agarwal <ar...@apache.org>
Authored: Mon Jan 16 16:37:38 2017 -0800
Committer: Arpit Agarwal <ar...@apache.org>
Committed: Mon Jan 16 22:14:06 2017 -0800
----------------------------------------------------------------------
.../hadoop/hdfs/server/datanode/DataNode.java | 23 -------
.../datanode/checker/DatasetVolumeChecker.java | 69 --------------------
.../hdfs/server/datanode/DataNodeTestUtils.java | 14 ++++
.../datanode/TestDataNodeHotSwapVolumes.java | 18 +----
.../datanode/TestDataNodeVolumeFailure.java | 17 ++---
.../TestDataNodeVolumeFailureReporting.java | 3 -
.../server/datanode/TestDirectoryScanner.java | 2 -
.../checker/TestDatasetVolumeChecker.java | 47 -------------
8 files changed, 25 insertions(+), 168 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8ec9dca2/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
index dc7d267..dabadeb 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
@@ -2025,29 +2025,6 @@ public class DataNode extends ReconfigurableBase
* Check if there is a disk failure asynchronously
* and if so, handle the error.
*/
- @VisibleForTesting
- public void checkDiskErrorAsync() {
- volumeChecker.checkAllVolumesAsync(
- data, new DatasetVolumeChecker.Callback() {
- @Override
- public void call(Set<FsVolumeSpi> healthyVolumes,
- Set<FsVolumeSpi> failedVolumes) {
- if (failedVolumes.size() > 0) {
- LOG.warn("checkDiskErrorAsync callback got {} failed volumes: {}",
- failedVolumes.size(), failedVolumes);
- } else {
- LOG.debug("checkDiskErrorAsync: no volume failures detected");
- }
- lastDiskErrorCheck = Time.monotonicNow();
- handleVolumeFailures(failedVolumes);
- }
- });
- }
-
- /**
- * Check if there is a disk failure asynchronously
- * and if so, handle the error.
- */
public void checkDiskErrorAsync(FsVolumeSpi volume) {
volumeChecker.checkVolume(
volume, new DatasetVolumeChecker.Callback() {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8ec9dca2/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/DatasetVolumeChecker.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/DatasetVolumeChecker.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/DatasetVolumeChecker.java
index 5ef3eec..0d96e72 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/DatasetVolumeChecker.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/DatasetVolumeChecker.java
@@ -233,68 +233,6 @@ public class DatasetVolumeChecker {
}
/**
- * Start checks against all volumes of a dataset, invoking the
- * given callback when the operation has completed. The function
- * does not wait for the checks to complete.
- *
- * If a volume cannot be referenced then it is already closed and
- * cannot be checked. No error is propagated to the callback for that
- * volume.
- *
- * @param dataset - FsDatasetSpi to be checked.
- * @param callback - Callback to be invoked when the checks are complete.
- * @return true if the check was scheduled and the callback will be invoked.
- * false if the check was not scheduled and the callback will not be
- * invoked.
- */
- public boolean checkAllVolumesAsync(
- final FsDatasetSpi<? extends FsVolumeSpi> dataset,
- Callback callback) {
- final long gap = timer.monotonicNow() - lastAllVolumesCheck;
- if (gap < minDiskCheckGapMs) {
- numSkippedChecks.incrementAndGet();
- LOG.trace(
- "Skipped checking all volumes, time since last check {} is less " +
- "than the minimum gap between checks ({} ms).",
- gap, minDiskCheckGapMs);
- return false;
- }
-
- final FsDatasetSpi.FsVolumeReferences references =
- dataset.getFsVolumeReferences();
-
- if (references.size() == 0) {
- LOG.warn("checkAllVolumesAsync - no volumes can be referenced");
- return false;
- }
-
- lastAllVolumesCheck = timer.monotonicNow();
- final Set<FsVolumeSpi> healthyVolumes = new HashSet<>();
- final Set<FsVolumeSpi> failedVolumes = new HashSet<>();
- final AtomicLong numVolumes = new AtomicLong(references.size());
- boolean added = false;
-
- LOG.info("Checking {} volumes", references.size());
- for (int i = 0; i < references.size(); ++i) {
- final FsVolumeReference reference = references.getReference(i);
- // The context parameter is currently ignored.
- Optional<ListenableFuture<VolumeCheckResult>> olf =
- delegateChecker.schedule(reference.getVolume(), IGNORED_CONTEXT);
- if (olf.isPresent()) {
- added = true;
- Futures.addCallback(olf.get(),
- new ResultHandler(reference, healthyVolumes, failedVolumes,
- numVolumes, callback));
- } else {
- IOUtils.cleanup(null, reference);
- numVolumes.decrementAndGet();
- }
- }
- numAsyncDatasetChecks.incrementAndGet();
- return added;
- }
-
- /**
* A callback interface that is supplied the result of running an
* async disk check on multiple volumes.
*/
@@ -490,13 +428,6 @@ public class DatasetVolumeChecker {
}
/**
- * Return the number of {@link #checkAllVolumesAsync} invocations.
- */
- public long getNumAsyncDatasetChecks() {
- return numAsyncDatasetChecks.get();
- }
-
- /**
* Return the number of checks skipped because the minimum gap since the
* last check had not elapsed.
*/
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8ec9dca2/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/DataNodeTestUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/DataNodeTestUtils.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/DataNodeTestUtils.java
index 42bcbf3..d35bf05 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/DataNodeTestUtils.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/DataNodeTestUtils.java
@@ -30,7 +30,9 @@ import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.MiniDFSCluster;
import org.apache.hadoop.hdfs.protocol.DatanodeID;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsDatasetSpi;
+import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.impl.FsDatasetTestUtil;
+import org.apache.hadoop.hdfs.server.datanode.fsdataset.impl.FsVolumeImpl;
import org.apache.hadoop.hdfs.server.protocol.DatanodeRegistration;
import org.apache.hadoop.hdfs.server.protocol.InterDatanodeProtocol;
@@ -214,4 +216,16 @@ public class DataNodeTestUtils {
LOG.warn("Could not reconfigure DataNode.", e);
}
}
+
+ /** Get the FsVolume on the given basePath. */
+ public static FsVolumeImpl getVolume(DataNode dn, File basePath) throws
+ IOException {
+ try (FsDatasetSpi.FsVolumeReferences volumes = dn.getFSDataset()
+ .getFsVolumeReferences()) {
+ for (FsVolumeSpi vol : volumes) {
+ return (FsVolumeImpl) vol;
+ }
+ }
+ return null;
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8ec9dca2/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 600769b..7edf5ca 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
@@ -900,20 +900,6 @@ public class TestDataNodeHotSwapVolumes {
is(dn.getConf().get(DFS_DATANODE_DATA_DIR_KEY)));
}
- /** Get the FsVolume on the given basePath */
- private FsVolumeImpl getVolume(DataNode dn, File basePath)
- throws IOException {
- try (FsDatasetSpi.FsVolumeReferences volumes =
- dn.getFSDataset().getFsVolumeReferences()) {
- for (FsVolumeSpi vol : volumes) {
- if (vol.getBasePath().equals(basePath.getPath())) {
- return (FsVolumeImpl) vol;
- }
- }
- }
- return null;
- }
-
/**
* Verify that {@link DataNode#checkDiskError()} removes all metadata in
* DataNode upon a volume failure. Thus we can run reconfig on the same
@@ -934,7 +920,7 @@ public class TestDataNodeHotSwapVolumes {
final String oldDataDir = dn.getConf().get(DFS_DATANODE_DATA_DIR_KEY);
File dirToFail = new File(cluster.getDataDirectory(), "data1");
- FsVolumeImpl failedVolume = getVolume(dn, dirToFail);
+ FsVolumeImpl failedVolume = DataNodeTestUtils.getVolume(dn, dirToFail);
assertTrue("No FsVolume was found for " + dirToFail,
failedVolume != null);
long used = failedVolume.getDfsUsed();
@@ -957,7 +943,7 @@ public class TestDataNodeHotSwapVolumes {
is(dn.getConf().get(DFS_DATANODE_DATA_DIR_KEY)));
createFile(new Path("/test2"), 32, (short)2);
- FsVolumeImpl restoredVolume = getVolume(dn, dirToFail);
+ FsVolumeImpl restoredVolume = DataNodeTestUtils.getVolume(dn, dirToFail);
assertTrue(restoredVolume != null);
assertTrue(restoredVolume != failedVolume);
// More data has been written to this volume.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8ec9dca2/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailure.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailure.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailure.java
index e58f993..e134bcc 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailure.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailure.java
@@ -239,7 +239,7 @@ public class TestDataNodeVolumeFailure {
File dn0Vol1 = new File(dataDir, "data" + (2 * 0 + 1));
DataNodeTestUtils.injectDataDirFailure(dn0Vol1);
DataNode dn0 = cluster.getDataNodes().get(0);
- checkDiskErrorSync(dn0);
+ checkDiskErrorSync(dn0, DataNodeTestUtils.getVolume(dn0, dn0Vol1));
// Verify dn0Vol1 has been completely removed from DN0.
// 1. dn0Vol1 is removed from DataStorage.
@@ -285,10 +285,10 @@ public class TestDataNodeVolumeFailure {
assertFalse(dataDirStrs[0].contains(dn0Vol1.getAbsolutePath()));
}
- private static void checkDiskErrorSync(DataNode dn)
+ private static void checkDiskErrorSync(DataNode dn, FsVolumeSpi volume)
throws InterruptedException {
final long lastDiskErrorCheck = dn.getLastDiskErrorCheck();
- dn.checkDiskErrorAsync();
+ dn.checkDiskErrorAsync(volume);
// Wait 10 seconds for checkDiskError thread to finish and discover volume
// failures.
int count = 100;
@@ -312,7 +312,8 @@ public class TestDataNodeVolumeFailure {
final File dn0Vol2 = new File(dataDir, "data" + (2 * 0 + 2));
DataNodeTestUtils.injectDataDirFailure(dn0Vol1, dn0Vol2);
DataNode dn0 = cluster.getDataNodes().get(0);
- checkDiskErrorSync(dn0);
+ checkDiskErrorSync(dn0, DataNodeTestUtils.getVolume(dn0, dn0Vol1));
+ checkDiskErrorSync(dn0, DataNodeTestUtils.getVolume(dn0, dn0Vol2));
// DN0 should stop after the number of failure disks exceed tolerated
// value (1).
@@ -333,7 +334,7 @@ public class TestDataNodeVolumeFailure {
// Fail dn0Vol1 first.
DataNodeTestUtils.injectDataDirFailure(dn0Vol1);
- checkDiskErrorSync(dn0);
+ checkDiskErrorSync(dn0, DataNodeTestUtils.getVolume(dn0, dn0Vol1));
// Hot swap out the failure volume.
String dataDirs = dn0Vol2.getPath();
@@ -352,7 +353,7 @@ public class TestDataNodeVolumeFailure {
// Fail dn0Vol2. Now since dn0Vol1 has been fixed, DN0 has sufficient
// resources, thus it should keep running.
DataNodeTestUtils.injectDataDirFailure(dn0Vol2);
- checkDiskErrorSync(dn0);
+ checkDiskErrorSync(dn0, DataNodeTestUtils.getVolume(dn0, dn0Vol2));
assertTrue(dn0.shouldRun());
}
@@ -379,12 +380,12 @@ public class TestDataNodeVolumeFailure {
// Fail dn0Vol1 first and hot swap it.
DataNodeTestUtils.injectDataDirFailure(dn0Vol1);
- checkDiskErrorSync(dn0);
+ checkDiskErrorSync(dn0, DataNodeTestUtils.getVolume(dn0, dn0Vol1));
assertTrue(dn0.shouldRun());
// Fail dn0Vol2, now dn0 should stop, because we only tolerate 1 disk failure.
DataNodeTestUtils.injectDataDirFailure(dn0Vol2);
- checkDiskErrorSync(dn0);
+ checkDiskErrorSync(dn0, DataNodeTestUtils.getVolume(dn0, dn0Vol2));
assertFalse(dn0.shouldRun());
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8ec9dca2/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailureReporting.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailureReporting.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailureReporting.java
index 79a52bb..5191083 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailureReporting.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailureReporting.java
@@ -17,9 +17,6 @@
*/
package org.apache.hadoop.hdfs.server.datanode;
-import static org.apache.hadoop.test.MetricsAsserts.getLongCounter;
-import static org.apache.hadoop.test.MetricsAsserts.getMetrics;
-import static org.hamcrest.core.Is.is;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8ec9dca2/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDirectoryScanner.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDirectoryScanner.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDirectoryScanner.java
index 6f7a6fa..ea52d15 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDirectoryScanner.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDirectoryScanner.java
@@ -54,8 +54,6 @@ import org.apache.hadoop.hdfs.HdfsConfiguration;
import org.apache.hadoop.hdfs.MiniDFSCluster;
import org.apache.hadoop.hdfs.server.datanode.checker.VolumeCheckResult;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.DataNodeVolumeMetrics;
-import org.apache.hadoop.util.AutoCloseableLock;
-import org.apache.hadoop.util.AutoCloseableLock;
import org.apache.hadoop.hdfs.protocol.Block;
import org.apache.hadoop.hdfs.protocol.HdfsConstants;
import org.apache.hadoop.hdfs.protocol.LocatedBlock;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8ec9dca2/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestDatasetVolumeChecker.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestDatasetVolumeChecker.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestDatasetVolumeChecker.java
index d6b4af9..2a1c824 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestDatasetVolumeChecker.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestDatasetVolumeChecker.java
@@ -161,53 +161,6 @@ public class TestDatasetVolumeChecker {
}
/**
- * Unit test for {@link DatasetVolumeChecker#checkAllVolumesAsync}.
- *
- * @throws Exception
- */
- @Test(timeout=10000)
- public void testCheckAllVolumesAsync() throws Exception {
- LOG.info("Executing {}", testName.getMethodName());
-
- final List<FsVolumeSpi> volumes = makeVolumes(
- NUM_VOLUMES, expectedVolumeHealth);
- final FsDatasetSpi<FsVolumeSpi> dataset = makeDataset(volumes);
- final DatasetVolumeChecker checker =
- new DatasetVolumeChecker(new HdfsConfiguration(), new FakeTimer());
- checker.setDelegateChecker(new DummyChecker());
- final AtomicLong numCallbackInvocations = new AtomicLong(0);
-
- boolean result = checker.checkAllVolumesAsync(
- dataset, new DatasetVolumeChecker.Callback() {
- @Override
- public void call(
- Set<FsVolumeSpi> healthyVolumes,
- Set<FsVolumeSpi> failedVolumes) {
- LOG.info("Got back {} failed volumes", failedVolumes.size());
- if (expectedVolumeHealth == null ||
- expectedVolumeHealth == FAILED) {
- assertThat(healthyVolumes.size(), is(0));
- assertThat(failedVolumes.size(), is(NUM_VOLUMES));
- } else {
- assertThat(healthyVolumes.size(), is(NUM_VOLUMES));
- assertThat(failedVolumes.size(), is(0));
- }
- numCallbackInvocations.incrementAndGet();
- }
- });
-
- // The callback should be invoked exactly once.
- if (result) {
- assertThat(numCallbackInvocations.get(), is(1L));
- }
-
- // Ensure each volume's check() method was called exactly once.
- for (FsVolumeSpi volume : volumes) {
- verify(volume, times(1)).check(any(VolumeCheckContext.class));
- }
- }
-
- /**
* A checker to wraps the result of {@link FsVolumeSpi#check} in
* an ImmediateFuture.
*/
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org