HDFS-11264: [SPS]: Double checks to ensure that SPS/Mover are not running together. Contributed by Rakesh R.

This commit is contained in:
Uma Maheswara Rao G 2017-07-12 17:56:56 -07:00 committed by Uma Maheswara Rao Gangumalla
parent 0b360b16ab
commit 5eb24ef7e7
3 changed files with 34 additions and 27 deletions

View File

@ -128,6 +128,14 @@ public StoragePolicySatisfier(final Namesystem namesystem,
*/
public synchronized void start(boolean reconfigStart) {
isRunning = true;
if (checkIfMoverRunning()) {
isRunning = false;
LOG.error(
"Stopping StoragePolicySatisfier thread " + "as Mover ID file "
+ HdfsServerConstants.MOVER_ID_PATH.toString()
+ " been opened. Maybe a Mover instance is running!");
return;
}
if (reconfigStart) {
LOG.info("Starting StoragePolicySatisfier, as admin requested to "
+ "activate it.");
@ -211,20 +219,6 @@ private void addDropSPSWorkCommandsToAllDNs() {
@Override
public void run() {
boolean isMoverRunning = !checkIfMoverRunning();
synchronized (this) {
isRunning = isMoverRunning;
if (!isRunning) {
// Stopping monitor thread and clearing queues as well
this.clearQueues();
this.storageMovementsMonitor.stopGracefully();
LOG.error(
"Stopping StoragePolicySatisfier thread " + "as Mover ID file "
+ HdfsServerConstants.MOVER_ID_PATH.toString()
+ " been opened. Maybe a Mover instance is running!");
return;
}
}
while (namesystem.isRunning() && isRunning) {
try {
if (!namesystem.isInSafeMode()) {
@ -274,25 +268,34 @@ public void run() {
// we want to check block movements.
Thread.sleep(3000);
} catch (Throwable t) {
synchronized (this) {
handleException(t);
}
}
}
private void handleException(Throwable t) {
// double check to avoid entering into synchronized block.
if (isRunning) {
synchronized (this) {
if (isRunning) {
isRunning = false;
// Stopping monitor thread and clearing queues as well
this.clearQueues();
this.storageMovementsMonitor.stopGracefully();
}
if (!namesystem.isRunning()) {
LOG.info("Stopping StoragePolicySatisfier.");
if (!(t instanceof InterruptedException)) {
LOG.info("StoragePolicySatisfier received an exception"
+ " while shutting down.", t);
if (!namesystem.isRunning()) {
LOG.info("Stopping StoragePolicySatisfier.");
if (!(t instanceof InterruptedException)) {
LOG.info("StoragePolicySatisfier received an exception"
+ " while shutting down.", t);
}
return;
}
break;
}
LOG.error("StoragePolicySatisfier thread received runtime exception. "
+ "Stopping Storage policy satisfier work", t);
break;
}
}
LOG.error("StoragePolicySatisfier thread received runtime exception. "
+ "Stopping Storage policy satisfier work", t);
return;
}
private BlocksMovingAnalysisStatus analyseBlocksStorageMovementsAndAssignToDN(

View File

@ -927,7 +927,8 @@ public void testSPSShouldNotLeakXattrIfSatisfyStoragePolicyCallOnECFiles()
String fooDir = "/foo";
client.mkdirs(fooDir, new FsPermission((short) 777), true);
// set an EC policy on "/foo" directory
client.setErasureCodingPolicy(fooDir, null);
client.setErasureCodingPolicy(fooDir,
StripedFileTestUtil.getDefaultECPolicy().getName());
// write file to fooDir
final String testFile = "/foo/bar";

View File

@ -323,6 +323,8 @@ public void testSPSWhenFileHasLowRedundancyBlocks() throws Exception {
conf.set(DFSConfigKeys
.DFS_STORAGE_POLICY_SATISFIER_RECHECK_TIMEOUT_MILLIS_KEY,
"3000");
conf.set(DFSConfigKeys.DFS_NAMENODE_EC_POLICIES_ENABLED_KEY,
StripedFileTestUtil.getDefaultECPolicy().getName());
initConfWithStripe(conf, defaultStripeBlockSize);
final MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf)
.numDataNodes(numOfDatanodes)
@ -346,7 +348,8 @@ public void testSPSWhenFileHasLowRedundancyBlocks() throws Exception {
Path barDir = new Path("/bar");
fs.mkdirs(barDir);
// set an EC policy on "/bar" directory
fs.setErasureCodingPolicy(barDir, null);
fs.setErasureCodingPolicy(barDir,
StripedFileTestUtil.getDefaultECPolicy().getName());
// write file to barDir
final Path fooFile = new Path("/bar/foo");