HDFS-1477. Support reconfiguring dfs.heartbeat.interval and dfs.namenode.heartbeat.recheck-interval without NN restart. (Contributed by Xiaobing Zhou)

This commit is contained in:
Arpit Agarwal 2016-03-10 19:03:55 -08:00
parent adf1cdf3d5
commit e01c6ea688
9 changed files with 357 additions and 86 deletions

View File

@ -332,7 +332,8 @@ public class BlockManager implements BlockStatsMXBean {
DFSConfigKeys.DFS_NAMENODE_STARTUP_DELAY_BLOCK_DELETION_SEC_KEY, DFSConfigKeys.DFS_NAMENODE_STARTUP_DELAY_BLOCK_DELETION_SEC_KEY,
DFSConfigKeys.DFS_NAMENODE_STARTUP_DELAY_BLOCK_DELETION_SEC_DEFAULT) * 1000L; DFSConfigKeys.DFS_NAMENODE_STARTUP_DELAY_BLOCK_DELETION_SEC_DEFAULT) * 1000L;
invalidateBlocks = new InvalidateBlocks( invalidateBlocks = new InvalidateBlocks(
datanodeManager.blockInvalidateLimit, startupDelayBlockDeletionInMs); datanodeManager.getBlockInvalidateLimit(),
startupDelayBlockDeletionInMs);
// Compute the map capacity by allocating 2% of total memory // Compute the map capacity by allocating 2% of total memory
blocksMap = new BlocksMap( blocksMap = new BlocksMap(

View File

@ -23,6 +23,7 @@ import static org.apache.hadoop.util.Time.monotonicNow;
import com.google.common.annotations.VisibleForTesting; import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions; import com.google.common.base.Preconditions;
import com.google.common.net.InetAddresses; import com.google.common.net.InetAddresses;
import org.apache.commons.logging.Log; import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory; import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.HadoopIllegalArgumentException; import org.apache.hadoop.HadoopIllegalArgumentException;
@ -70,6 +71,8 @@ public class DatanodeManager {
private final HeartbeatManager heartbeatManager; private final HeartbeatManager heartbeatManager;
private final FSClusterStats fsClusterStats; private final FSClusterStats fsClusterStats;
private volatile long heartbeatIntervalSeconds;
private volatile int heartbeatRecheckInterval;
/** /**
* Stores the datanode -> block map. * Stores the datanode -> block map.
* <p> * <p>
@ -113,7 +116,7 @@ public class DatanodeManager {
/** The period to wait for datanode heartbeat.*/ /** The period to wait for datanode heartbeat.*/
private long heartbeatExpireInterval; private long heartbeatExpireInterval;
/** Ask Datanode only up to this many blocks to delete. */ /** Ask Datanode only up to this many blocks to delete. */
final int blockInvalidateLimit; private volatile int blockInvalidateLimit;
/** The interval for judging stale DataNodes for read/write */ /** The interval for judging stale DataNodes for read/write */
private final long staleInterval; private final long staleInterval;
@ -227,10 +230,10 @@ public class DatanodeManager {
dnsToSwitchMapping.resolve(locations); dnsToSwitchMapping.resolve(locations);
} }
final long heartbeatIntervalSeconds = conf.getLong( heartbeatIntervalSeconds = conf.getLong(
DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_KEY, DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_KEY,
DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_DEFAULT); DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_DEFAULT);
final int heartbeatRecheckInterval = conf.getInt( heartbeatRecheckInterval = conf.getInt(
DFSConfigKeys.DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_KEY, DFSConfigKeys.DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_KEY,
DFSConfigKeys.DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_DEFAULT); // 5 minutes DFSConfigKeys.DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_DEFAULT); // 5 minutes
this.heartbeatExpireInterval = 2 * heartbeatRecheckInterval this.heartbeatExpireInterval = 2 * heartbeatRecheckInterval
@ -348,6 +351,10 @@ public class DatanodeManager {
return fsClusterStats; return fsClusterStats;
} }
int getBlockInvalidateLimit() {
return blockInvalidateLimit;
}
/** @return the datanode statistics. */ /** @return the datanode statistics. */
public DatanodeStatistics getDatanodeStatistics() { public DatanodeStatistics getDatanodeStatistics() {
return heartbeatManager; return heartbeatManager;
@ -1103,6 +1110,14 @@ public class DatanodeManager {
return staleInterval; return staleInterval;
} }
public long getHeartbeatInterval() {
return this.heartbeatIntervalSeconds;
}
public long getHeartbeatRecheckInterval() {
return this.heartbeatRecheckInterval;
}
/** /**
* Set the number of current stale DataNodes. The HeartbeatManager got this * Set the number of current stale DataNodes. The HeartbeatManager got this
* number based on DataNodes' heartbeats. * number based on DataNodes' heartbeats.
@ -1667,5 +1682,28 @@ public class DatanodeManager {
} }
}; };
} }
public void setHeartbeatInterval(long intervalSeconds) {
setHeartbeatInterval(intervalSeconds,
this.heartbeatRecheckInterval);
}
public void setHeartbeatRecheckInterval(int recheckInterval) {
setHeartbeatInterval(this.heartbeatIntervalSeconds,
recheckInterval);
}
/**
* Set parameters derived from heartbeat interval.
*/
private void setHeartbeatInterval(long intervalSeconds,
int recheckInterval) {
this.heartbeatIntervalSeconds = intervalSeconds;
this.heartbeatRecheckInterval = recheckInterval;
this.heartbeatExpireInterval = 2L * recheckInterval + 10 * 1000
* intervalSeconds;
this.blockInvalidateLimit = Math.max(20 * (int) (intervalSeconds),
DFSConfigKeys.DFS_BLOCK_INVALIDATE_LIMIT_DEFAULT);
}
} }

View File

@ -2995,6 +2995,7 @@ public class DataNode extends ReconfigurableBase
@Override // ClientDatanodeProtocol & ReconfigurationProtocol @Override // ClientDatanodeProtocol & ReconfigurationProtocol
public List<String> listReconfigurableProperties() public List<String> listReconfigurableProperties()
throws IOException { throws IOException {
checkSuperuserPrivilege();
return RECONFIGURABLE_PROPERTIES; return RECONFIGURABLE_PROPERTIES;
} }

View File

@ -21,11 +21,14 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Joiner; import com.google.common.base.Joiner;
import com.google.common.base.Preconditions; import com.google.common.base.Preconditions;
import com.google.common.collect.Lists; import com.google.common.collect.Lists;
import org.apache.commons.logging.Log; import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory; import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.HadoopIllegalArgumentException; import org.apache.hadoop.HadoopIllegalArgumentException;
import org.apache.hadoop.classification.InterfaceAudience; import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.ReconfigurableBase;
import org.apache.hadoop.conf.ReconfigurationException;
import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Trash; import org.apache.hadoop.fs.Trash;
import org.apache.hadoop.ha.HAServiceProtocol.HAServiceState; import org.apache.hadoop.ha.HAServiceProtocol.HAServiceState;
@ -41,6 +44,7 @@ import org.apache.hadoop.hdfs.HdfsConfiguration;
import org.apache.hadoop.hdfs.client.HdfsClientConfigKeys; import org.apache.hadoop.hdfs.client.HdfsClientConfigKeys;
import org.apache.hadoop.hdfs.protocol.ClientProtocol; import org.apache.hadoop.hdfs.protocol.ClientProtocol;
import org.apache.hadoop.hdfs.protocol.HdfsConstants; import org.apache.hadoop.hdfs.protocol.HdfsConstants;
import org.apache.hadoop.hdfs.server.blockmanagement.DatanodeManager;
import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.NamenodeRole; import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.NamenodeRole;
import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.RollingUpgradeStartupOption; import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.RollingUpgradeStartupOption;
import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.StartupOption; import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.StartupOption;
@ -93,6 +97,7 @@ import java.security.PrivilegedExceptionAction;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Arrays; import java.util.Arrays;
import java.util.Collection; import java.util.Collection;
import java.util.Collections;
import java.util.List; import java.util.List;
import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
@ -141,6 +146,10 @@ import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_SUPPORT_ALLOW_FO
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMESERVICE_ID; import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMESERVICE_ID;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_SECONDARY_NAMENODE_KEYTAB_FILE_KEY; import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_SECONDARY_NAMENODE_KEYTAB_FILE_KEY;
import static org.apache.hadoop.hdfs.DFSConfigKeys.HADOOP_USER_GROUP_METRICS_PERCENTILES_INTERVALS; import static org.apache.hadoop.hdfs.DFSConfigKeys.HADOOP_USER_GROUP_METRICS_PERCENTILES_INTERVALS;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_KEY;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_DEFAULT;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_KEY;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_DEFAULT;
import static org.apache.hadoop.util.ExitUtil.terminate; import static org.apache.hadoop.util.ExitUtil.terminate;
import static org.apache.hadoop.util.ToolRunner.confirmPrompt; import static org.apache.hadoop.util.ToolRunner.confirmPrompt;
@ -182,7 +191,8 @@ import static org.apache.hadoop.util.ToolRunner.confirmPrompt;
* NameNode state, for example partial blocksMap etc. * NameNode state, for example partial blocksMap etc.
**********************************************************/ **********************************************************/
@InterfaceAudience.Private @InterfaceAudience.Private
public class NameNode implements NameNodeStatusMXBean { public class NameNode extends ReconfigurableBase implements
NameNodeStatusMXBean {
static{ static{
HdfsConfiguration.init(); HdfsConfiguration.init();
} }
@ -260,7 +270,12 @@ public class NameNode implements NameNodeStatusMXBean {
public static final String[] NAMESERVICE_SPECIFIC_KEYS = { public static final String[] NAMESERVICE_SPECIFIC_KEYS = {
DFS_HA_AUTO_FAILOVER_ENABLED_KEY DFS_HA_AUTO_FAILOVER_ENABLED_KEY
}; };
/** A list of property that are reconfigurable at runtime. */
static final List<String> RECONFIGURABLE_PROPERTIES = Collections
.unmodifiableList(Arrays.asList(DFS_HEARTBEAT_INTERVAL_KEY,
DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_KEY));
private static final String USAGE = "Usage: hdfs namenode [" private static final String USAGE = "Usage: hdfs namenode ["
+ StartupOption.BACKUP.getName() + "] | \n\t[" + StartupOption.BACKUP.getName() + "] | \n\t["
+ StartupOption.CHECKPOINT.getName() + "] | \n\t[" + StartupOption.CHECKPOINT.getName() + "] | \n\t["
@ -329,7 +344,6 @@ public class NameNode implements NameNodeStatusMXBean {
LogFactory.getLog("NameNodeMetricsLog"); LogFactory.getLog("NameNodeMetricsLog");
protected FSNamesystem namesystem; protected FSNamesystem namesystem;
protected final Configuration conf;
protected final NamenodeRole role; protected final NamenodeRole role;
private volatile HAState state; private volatile HAState state;
private final boolean haEnabled; private final boolean haEnabled;
@ -864,12 +878,12 @@ public class NameNode implements NameNodeStatusMXBean {
protected NameNode(Configuration conf, NamenodeRole role) protected NameNode(Configuration conf, NamenodeRole role)
throws IOException { throws IOException {
super(conf);
this.tracer = new Tracer.Builder("NameNode"). this.tracer = new Tracer.Builder("NameNode").
conf(TraceUtils.wrapHadoopConf(NAMENODE_HTRACE_PREFIX, conf)). conf(TraceUtils.wrapHadoopConf(NAMENODE_HTRACE_PREFIX, conf)).
build(); build();
this.tracerConfigurationManager = this.tracerConfigurationManager =
new TracerConfigurationManager(NAMENODE_HTRACE_PREFIX, conf); new TracerConfigurationManager(NAMENODE_HTRACE_PREFIX, conf);
this.conf = conf;
this.role = role; this.role = role;
setClientNamenodeAddress(conf); setClientNamenodeAddress(conf);
String nsId = getNameServiceId(conf); String nsId = getNameServiceId(conf);
@ -880,7 +894,7 @@ public class NameNode implements NameNodeStatusMXBean {
this.haContext = createHAContext(); this.haContext = createHAContext();
try { try {
initializeGenericKeys(conf, nsId, namenodeId); initializeGenericKeys(conf, nsId, namenodeId);
initialize(conf); initialize(getConf());
try { try {
haContext.writeLock(); haContext.writeLock();
state.prepareToEnterState(haContext); state.prepareToEnterState(haContext);
@ -1804,7 +1818,7 @@ public class NameNode implements NameNodeStatusMXBean {
public void startActiveServices() throws IOException { public void startActiveServices() throws IOException {
try { try {
namesystem.startActiveServices(); namesystem.startActiveServices();
startTrashEmptier(conf); startTrashEmptier(getConf());
} catch (Throwable t) { } catch (Throwable t) {
doImmediateShutdown(t); doImmediateShutdown(t);
} }
@ -1825,7 +1839,7 @@ public class NameNode implements NameNodeStatusMXBean {
@Override @Override
public void startStandbyServices() throws IOException { public void startStandbyServices() throws IOException {
try { try {
namesystem.startStandbyServices(conf); namesystem.startStandbyServices(getConf());
} catch (Throwable t) { } catch (Throwable t) {
doImmediateShutdown(t); doImmediateShutdown(t);
} }
@ -1902,8 +1916,8 @@ public class NameNode implements NameNodeStatusMXBean {
*/ */
void checkHaStateChange(StateChangeRequestInfo req) void checkHaStateChange(StateChangeRequestInfo req)
throws AccessControlException { throws AccessControlException {
boolean autoHaEnabled = conf.getBoolean(DFS_HA_AUTO_FAILOVER_ENABLED_KEY, boolean autoHaEnabled = getConf().getBoolean(
DFS_HA_AUTO_FAILOVER_ENABLED_DEFAULT); DFS_HA_AUTO_FAILOVER_ENABLED_KEY, DFS_HA_AUTO_FAILOVER_ENABLED_DEFAULT);
switch (req.getSource()) { switch (req.getSource()) {
case REQUEST_BY_USER: case REQUEST_BY_USER:
if (autoHaEnabled) { if (autoHaEnabled) {
@ -1930,4 +1944,75 @@ public class NameNode implements NameNodeStatusMXBean {
break; break;
} }
} }
/*
* {@inheritDoc}
* */
@Override // ReconfigurableBase
public Collection<String> getReconfigurableProperties() {
return RECONFIGURABLE_PROPERTIES;
}
/*
* {@inheritDoc}
* */
@Override // ReconfigurableBase
protected String reconfigurePropertyImpl(String property, String newVal)
throws ReconfigurationException {
final DatanodeManager datanodeManager = namesystem.getBlockManager()
.getDatanodeManager();
switch (property) {
case DFS_HEARTBEAT_INTERVAL_KEY:
namesystem.writeLock();
try {
if (newVal == null) {
// set to default
datanodeManager.setHeartbeatInterval(DFS_HEARTBEAT_INTERVAL_DEFAULT);
return String.valueOf(DFS_HEARTBEAT_INTERVAL_DEFAULT);
} else {
datanodeManager.setHeartbeatInterval(Long.parseLong(newVal));
return String.valueOf(datanodeManager.getHeartbeatInterval());
}
} catch (NumberFormatException nfe) {
throw new ReconfigurationException(property, newVal, getConf().get(
property), nfe);
} finally {
namesystem.writeUnlock();
LOG.info("RECONFIGURE* changed heartbeatInterval to "
+ datanodeManager.getHeartbeatInterval());
}
case DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_KEY:
namesystem.writeLock();
try {
if (newVal == null) {
// set to default
datanodeManager
.setHeartbeatRecheckInterval(
DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_DEFAULT);
return String
.valueOf(DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_DEFAULT);
} else {
datanodeManager.setHeartbeatRecheckInterval(Integer.parseInt(newVal));
return String.valueOf(datanodeManager.getHeartbeatRecheckInterval());
}
} catch (NumberFormatException nfe) {
throw new ReconfigurationException(property, newVal, getConf().get(
property), nfe);
} finally {
namesystem.writeUnlock();
LOG.info("RECONFIGURE* changed heartbeatRecheckInterval to "
+ datanodeManager.getHeartbeatRecheckInterval());
}
default:
break;
}
throw new ReconfigurationException(property, newVal, getConf()
.get(property));
}
@Override // ReconfigurableBase
protected Configuration getNewConf() {
return new HdfsConfiguration();
}
} }

View File

@ -43,7 +43,6 @@ import com.google.common.collect.Lists;
import org.apache.hadoop.HadoopIllegalArgumentException; import org.apache.hadoop.HadoopIllegalArgumentException;
import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.ReconfigurationException;
import org.apache.hadoop.conf.ReconfigurationTaskStatus; import org.apache.hadoop.conf.ReconfigurationTaskStatus;
import org.apache.hadoop.crypto.CryptoProtocolVersion; import org.apache.hadoop.crypto.CryptoProtocolVersion;
import org.apache.hadoop.fs.BatchedRemoteIterator.BatchedEntries; import org.apache.hadoop.fs.BatchedRemoteIterator.BatchedEntries;
@ -2109,7 +2108,7 @@ class NameNodeRpcServer implements NamenodeProtocols {
checkNNStartup(); checkNNStartup();
namesystem.checkOperation(OperationCategory.READ); // only active namesystem.checkOperation(OperationCategory.READ); // only active
namesystem.checkSuperuserPrivilege(); namesystem.checkSuperuserPrivilege();
int maxEventsPerRPC = nn.conf.getInt( int maxEventsPerRPC = nn.getConf().getInt(
DFSConfigKeys.DFS_NAMENODE_INOTIFY_MAX_EVENTS_PER_RPC_KEY, DFSConfigKeys.DFS_NAMENODE_INOTIFY_MAX_EVENTS_PER_RPC_KEY,
DFSConfigKeys.DFS_NAMENODE_INOTIFY_MAX_EVENTS_PER_RPC_DEFAULT); DFSConfigKeys.DFS_NAMENODE_INOTIFY_MAX_EVENTS_PER_RPC_DEFAULT);
FSEditLog log = namesystem.getFSImage().getEditLog(); FSEditLog log = namesystem.getFSImage().getEditLog();
@ -2224,23 +2223,24 @@ class NameNodeRpcServer implements NamenodeProtocols {
} }
@Override // ReconfigurationProtocol @Override // ReconfigurationProtocol
public void startReconfiguration() { public void startReconfiguration() throws IOException {
throw new UnsupportedOperationException( checkNNStartup();
"Namenode startReconfiguration is not implemented.", namesystem.checkSuperuserPrivilege();
new ReconfigurationException()); nn.startReconfigurationTask();
} }
@Override // ReconfigurationProtocol @Override // ReconfigurationProtocol
public ReconfigurationTaskStatus getReconfigurationStatus() { public ReconfigurationTaskStatus getReconfigurationStatus()
throw new UnsupportedOperationException( throws IOException {
" Namenode getReconfigurationStatus is not implemented.", checkNNStartup();
new ReconfigurationException()); namesystem.checkSuperuserPrivilege();
return nn.getReconfigurationTaskStatus();
} }
@Override // ReconfigurationProtocol @Override // ReconfigurationProtocol
public List<String> listReconfigurableProperties() { public List<String> listReconfigurableProperties() throws IOException {
throw new UnsupportedOperationException( checkNNStartup();
" Namenode listReconfigurableProperties is not implemented.", namesystem.checkSuperuserPrivilege();
new ReconfigurationException()); return NameNode.RECONFIGURABLE_PROPERTIES;
} }
} }

View File

@ -938,7 +938,7 @@ public class NamenodeFsck implements DataEncryptionKeyFactory {
setInetSocketAddress(targetAddr). setInetSocketAddress(targetAddr).
setCachingStrategy(CachingStrategy.newDropBehind()). setCachingStrategy(CachingStrategy.newDropBehind()).
setClientCacheContext(dfs.getClientContext()). setClientCacheContext(dfs.getClientContext()).
setConfiguration(namenode.conf). setConfiguration(namenode.getConf()).
setTracer(tracer). setTracer(tracer).
setRemotePeerFactory(new RemotePeerFactory() { setRemotePeerFactory(new RemotePeerFactory() {
@Override @Override

View File

@ -81,7 +81,7 @@ public class TestComputeInvalidateWork {
@Test(timeout=120000) @Test(timeout=120000)
public void testCompInvalidate() throws Exception { public void testCompInvalidate() throws Exception {
final int blockInvalidateLimit = bm.getDatanodeManager() final int blockInvalidateLimit = bm.getDatanodeManager()
.blockInvalidateLimit; .getBlockInvalidateLimit();
namesystem.writeLock(); namesystem.writeLock();
try { try {
for (int i=0; i<nodes.length; i++) { for (int i=0; i<nodes.length; i++) {

View File

@ -0,0 +1,126 @@
/**
* 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.namenode;
import java.io.IOException;
import org.junit.Test;
import org.junit.Before;
import org.junit.After;
import static org.junit.Assert.*;
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.MiniDFSCluster;
import org.apache.hadoop.hdfs.HdfsConfiguration;
import org.apache.hadoop.hdfs.server.blockmanagement.DatanodeManager;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_KEY;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_DEFAULT;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_KEY;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_DEFAULT;
public class TestNameNodeReconfigure {
public static final Log LOG = LogFactory
.getLog(TestNameNodeReconfigure.class);
private MiniDFSCluster cluster;
@Before
public void setUp() throws IOException {
Configuration conf = new HdfsConfiguration();
cluster = new MiniDFSCluster.Builder(conf).build();
}
/**
* Test that we can modify configuration properties.
*/
@Test
public void testReconfigure() throws ReconfigurationException {
final NameNode nameNode = cluster.getNameNode();
final DatanodeManager datanodeManager = nameNode.namesystem
.getBlockManager().getDatanodeManager();
// change properties
nameNode.reconfigureProperty(DFS_HEARTBEAT_INTERVAL_KEY, "" + 6);
nameNode.reconfigureProperty(DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_KEY,
"" + (10 * 60 * 1000));
// try invalid values
try {
nameNode.reconfigureProperty(DFS_HEARTBEAT_INTERVAL_KEY, "text");
fail("ReconfigurationException expected");
} catch (ReconfigurationException expected) {
assertTrue(expected.getCause() instanceof NumberFormatException);
}
try {
nameNode.reconfigureProperty(DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_KEY,
"text");
fail("ReconfigurationException expected");
} catch (ReconfigurationException expected) {
assertTrue(expected.getCause() instanceof NumberFormatException);
}
// verify change
assertEquals(
DFS_HEARTBEAT_INTERVAL_KEY + " has wrong value",
6,
nameNode.getConf().getLong(DFS_HEARTBEAT_INTERVAL_KEY,
DFS_HEARTBEAT_INTERVAL_DEFAULT));
assertEquals(DFS_HEARTBEAT_INTERVAL_KEY + " has wrong value", 6,
datanodeManager.getHeartbeatInterval());
assertEquals(
DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_KEY + " has wrong value",
10 * 60 * 1000,
nameNode.getConf().getInt(DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_KEY,
DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_DEFAULT));
assertEquals(DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_KEY
+ " has wrong value", 10 * 60 * 1000,
datanodeManager.getHeartbeatRecheckInterval());
// revert to defaults
nameNode.reconfigureProperty(DFS_HEARTBEAT_INTERVAL_KEY, null);
nameNode.reconfigureProperty(DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_KEY,
null);
// verify defaults
assertEquals(DFS_HEARTBEAT_INTERVAL_KEY + " has wrong value", null,
nameNode.getConf().get(DFS_HEARTBEAT_INTERVAL_KEY));
assertEquals(DFS_HEARTBEAT_INTERVAL_KEY + " has wrong value",
DFS_HEARTBEAT_INTERVAL_DEFAULT, datanodeManager.getHeartbeatInterval());
assertEquals(DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_KEY
+ " has wrong value", null,
nameNode.getConf().get(DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_KEY));
assertEquals(DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_KEY
+ " has wrong value", DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_DEFAULT,
datanodeManager.getHeartbeatRecheckInterval());
}
@After
public void shutDown() throws IOException {
if (cluster != null) {
cluster.shutdown();
}
}
}

View File

@ -18,7 +18,11 @@
package org.apache.hadoop.hdfs.tools; package org.apache.hadoop.hdfs.tools;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_DATA_DIR_KEY; import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_DATA_DIR_KEY;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_DEFAULT;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_KEY;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_KEY;
import com.google.common.base.Supplier;
import com.google.common.collect.Lists; import com.google.common.collect.Lists;
import org.apache.commons.logging.Log; import org.apache.commons.logging.Log;
@ -31,6 +35,7 @@ import org.apache.hadoop.hdfs.server.common.Storage;
import org.apache.hadoop.hdfs.server.datanode.DataNode; import org.apache.hadoop.hdfs.server.datanode.DataNode;
import org.apache.hadoop.hdfs.server.datanode.StorageLocation; import org.apache.hadoop.hdfs.server.datanode.StorageLocation;
import org.apache.hadoop.hdfs.server.namenode.NameNode; import org.apache.hadoop.hdfs.server.namenode.NameNode;
import org.apache.hadoop.test.GenericTestUtils;
import org.junit.After; import org.junit.After;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;
@ -42,6 +47,7 @@ import java.io.PrintStream;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.List; import java.util.List;
import java.util.Scanner; import java.util.Scanner;
import java.util.concurrent.TimeoutException;
import static org.hamcrest.CoreMatchers.allOf; import static org.hamcrest.CoreMatchers.allOf;
import static org.hamcrest.CoreMatchers.anyOf; import static org.hamcrest.CoreMatchers.anyOf;
@ -89,12 +95,6 @@ public class TestDFSAdmin {
namenode = cluster.getNameNode(); namenode = cluster.getNameNode();
} }
private void startReconfiguration(String nodeType, String address,
final List<String> outs, final List<String> errs) throws IOException {
reconfigurationOutErrFormatter("startReconfiguration", nodeType,
address, outs, errs);
}
private void getReconfigurableProperties(String nodeType, String address, private void getReconfigurableProperties(String nodeType, String address,
final List<String> outs, final List<String> errs) throws IOException { final List<String> outs, final List<String> errs) throws IOException {
reconfigurationOutErrFormatter("getReconfigurableProperties", nodeType, reconfigurationOutErrFormatter("getReconfigurableProperties", nodeType,
@ -151,9 +151,10 @@ public class TestDFSAdmin {
* @param expectedSuccuss set true if the reconfiguration task should success. * @param expectedSuccuss set true if the reconfiguration task should success.
* @throws IOException * @throws IOException
* @throws InterruptedException * @throws InterruptedException
* @throws TimeoutException
*/ */
private void testDataNodeGetReconfigurationStatus(boolean expectedSuccuss) private void testDataNodeGetReconfigurationStatus(boolean expectedSuccuss)
throws IOException, InterruptedException { throws IOException, InterruptedException, TimeoutException {
ReconfigurationUtil ru = mock(ReconfigurationUtil.class); ReconfigurationUtil ru = mock(ReconfigurationUtil.class);
datanode.setReconfigurationUtil(ru); datanode.setReconfigurationUtil(ru);
@ -179,21 +180,10 @@ public class TestDFSAdmin {
assertThat(admin.startReconfiguration("datanode", address), is(0)); assertThat(admin.startReconfiguration("datanode", address), is(0));
int count = 100;
final List<String> outs = Lists.newArrayList(); final List<String> outs = Lists.newArrayList();
final List<String> errs = Lists.newArrayList(); final List<String> errs = Lists.newArrayList();
while (count > 0) { awaitReconfigurationFinished("datanode", address, outs, errs);
outs.clear();
errs.clear();
getReconfigurationStatus("datanode", address, outs, errs);
if (!outs.isEmpty() && outs.get(0).contains("finished")) {
break;
}
count--;
Thread.sleep(100);
}
LOG.info(String.format("count=%d", count));
assertTrue(count > 0);
if (expectedSuccuss) { if (expectedSuccuss) {
assertThat(outs.size(), is(4)); assertThat(outs.size(), is(4));
} else { } else {
@ -232,59 +222,89 @@ public class TestDFSAdmin {
@Test(timeout = 30000) @Test(timeout = 30000)
public void testDataNodeGetReconfigurationStatus() throws IOException, public void testDataNodeGetReconfigurationStatus() throws IOException,
InterruptedException { InterruptedException, TimeoutException {
testDataNodeGetReconfigurationStatus(true); testDataNodeGetReconfigurationStatus(true);
restartCluster(); restartCluster();
testDataNodeGetReconfigurationStatus(false); testDataNodeGetReconfigurationStatus(false);
} }
@Test(timeout = 30000)
public void testNameNodeStartReconfiguration() throws IOException {
final String address = namenode.getHostAndPort();
final List<String> outs = Lists.newArrayList();
final List<String> errs = Lists.newArrayList();
startReconfiguration("namenode", address, outs, errs);
assertEquals(0, outs.size());
assertTrue(errs.size() > 1);
assertThat(
errs.get(0),
is(allOf(containsString("Namenode"), containsString("reconfiguring:"),
containsString("startReconfiguration"),
containsString("is not implemented"),
containsString("UnsupportedOperationException"))));
}
@Test(timeout = 30000) @Test(timeout = 30000)
public void testNameNodeGetReconfigurableProperties() throws IOException { public void testNameNodeGetReconfigurableProperties() throws IOException {
final String address = namenode.getHostAndPort(); final String address = namenode.getHostAndPort();
final List<String> outs = Lists.newArrayList(); final List<String> outs = Lists.newArrayList();
final List<String> errs = Lists.newArrayList(); final List<String> errs = Lists.newArrayList();
getReconfigurableProperties("namenode", address, outs, errs); getReconfigurableProperties("namenode", address, outs, errs);
assertEquals(0, outs.size()); assertEquals(3, outs.size());
assertTrue(errs.size() > 1); assertEquals(DFS_HEARTBEAT_INTERVAL_KEY, outs.get(1));
assertThat( assertEquals(DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_KEY, outs.get(2));
errs.get(0), assertEquals(errs.size(), 0);
is(allOf(containsString("Namenode"), }
containsString("reconfiguration:"),
containsString("listReconfigurableProperties"), void awaitReconfigurationFinished(final String nodeType,
containsString("is not implemented"), final String address, final List<String> outs, final List<String> errs)
containsString("UnsupportedOperationException")))); throws TimeoutException, IOException, InterruptedException {
GenericTestUtils.waitFor(new Supplier<Boolean>() {
@Override
public Boolean get() {
outs.clear();
errs.clear();
try {
getReconfigurationStatus(nodeType, address, outs, errs);
} catch (IOException e) {
LOG.error(String.format(
"call getReconfigurationStatus on %s[%s] failed.", nodeType,
address), e);
}
return !outs.isEmpty() && outs.get(0).contains("finished");
}
}, 100, 100 * 100);
} }
@Test(timeout = 30000) @Test(timeout = 30000)
public void testNameNodeGetReconfigurationStatus() throws IOException { public void testNameNodeGetReconfigurationStatus() throws IOException,
InterruptedException, TimeoutException {
ReconfigurationUtil ru = mock(ReconfigurationUtil.class);
namenode.setReconfigurationUtil(ru);
final String address = namenode.getHostAndPort(); final String address = namenode.getHostAndPort();
List<ReconfigurationUtil.PropertyChange> changes =
new ArrayList<>();
changes.add(new ReconfigurationUtil.PropertyChange(
DFS_HEARTBEAT_INTERVAL_KEY, String.valueOf(6),
namenode.getConf().get(DFS_HEARTBEAT_INTERVAL_KEY)));
changes.add(new ReconfigurationUtil.PropertyChange(
"randomKey", "new123", "old456"));
when(ru.parseChangedProperties(any(Configuration.class),
any(Configuration.class))).thenReturn(changes);
assertThat(admin.startReconfiguration("namenode", address), is(0));
final List<String> outs = Lists.newArrayList(); final List<String> outs = Lists.newArrayList();
final List<String> errs = Lists.newArrayList(); final List<String> errs = Lists.newArrayList();
getReconfigurationStatus("namenode", address, outs, errs); awaitReconfigurationFinished("namenode", address, outs, errs);
assertEquals(0, outs.size());
assertTrue(errs.size() > 1); // verify change
assertThat( assertEquals(
errs.get(0), DFS_HEARTBEAT_INTERVAL_KEY + " has wrong value",
is(allOf(containsString("Namenode"), 6,
containsString("reloading configuration:"), namenode
containsString("getReconfigurationStatus"), .getConf()
containsString("is not implemented"), .getLong(DFS_HEARTBEAT_INTERVAL_KEY,
containsString("UnsupportedOperationException")))); DFS_HEARTBEAT_INTERVAL_DEFAULT));
assertEquals(DFS_HEARTBEAT_INTERVAL_KEY + " has wrong value",
6,
namenode
.getNamesystem()
.getBlockManager()
.getDatanodeManager()
.getHeartbeatInterval());
int offset = 1;
assertThat(outs.get(offset), containsString("SUCCESS: Changed property "
+ DFS_HEARTBEAT_INTERVAL_KEY));
assertThat(outs.get(offset + 1),
is(allOf(containsString("From:"), containsString("3"))));
assertThat(outs.get(offset + 2),
is(allOf(containsString("To:"), containsString("6"))));
} }
} }