Revert HDFS-6940.

This commit is contained in:
Konstantin V Shvachko 2014-09-09 17:30:10 -07:00
parent 28d99db992
commit 05af0ff4be
6 changed files with 26 additions and 57 deletions

View File

@ -449,8 +449,6 @@ Release 2.6.0 - UNRELEASED
HDFS-6376. Distcp data between two HA clusters requires another configuration. HDFS-6376. Distcp data between two HA clusters requires another configuration.
(Dave Marion and Haohui Mai via jing9) (Dave Marion and Haohui Mai via jing9)
HDFS-6940. Refactoring to allow ConsensusNode implementation. (shv)
HDFS-6943. Improve NN allocateBlock log to include replicas' datanode IPs. HDFS-6943. Improve NN allocateBlock log to include replicas' datanode IPs.
(Ming Ma via wheat9) (Ming Ma via wheat9)

View File

@ -164,7 +164,7 @@ public int getPendingDataNodeMessageCount() {
final BlocksMap blocksMap; final BlocksMap blocksMap;
/** Replication thread. */ /** Replication thread. */
Daemon replicationThread; final Daemon replicationThread = new Daemon(new ReplicationMonitor());
/** Store blocks -> datanodedescriptor(s) map of corrupt replicas */ /** Store blocks -> datanodedescriptor(s) map of corrupt replicas */
final CorruptReplicasMap corruptReplicas = new CorruptReplicasMap(); final CorruptReplicasMap corruptReplicas = new CorruptReplicasMap();
@ -263,7 +263,6 @@ public BlockManager(final Namesystem namesystem, final FSClusterStats stats,
this.namesystem = namesystem; this.namesystem = namesystem;
datanodeManager = new DatanodeManager(this, namesystem, conf); datanodeManager = new DatanodeManager(this, namesystem, conf);
heartbeatManager = datanodeManager.getHeartbeatManager(); heartbeatManager = datanodeManager.getHeartbeatManager();
setReplicationMonitor(new ReplicationMonitor());
final long pendingPeriod = conf.getLong( final long pendingPeriod = conf.getLong(
DFSConfigKeys.DFS_NAMENODE_STARTUP_DELAY_BLOCK_DELETION_SEC_KEY, DFSConfigKeys.DFS_NAMENODE_STARTUP_DELAY_BLOCK_DELETION_SEC_KEY,
@ -396,22 +395,6 @@ private static BlockTokenSecretManager createBlockTokenSecretManager(
} }
} }
public long getReplicationRecheckInterval() {
return replicationRecheckInterval;
}
public AtomicLong excessBlocksCount() {
return excessBlocksCount;
}
public void clearInvalidateBlocks() {
invalidateBlocks.clear();
}
void setReplicationMonitor(Runnable replicationMonitor) {
replicationThread = new Daemon(replicationMonitor);
}
public void setBlockPoolId(String blockPoolId) { public void setBlockPoolId(String blockPoolId) {
if (isBlockTokenEnabled()) { if (isBlockTokenEnabled()) {
blockTokenSecretManager.setBlockPoolId(blockPoolId); blockTokenSecretManager.setBlockPoolId(blockPoolId);
@ -1633,7 +1616,7 @@ else if (excessBlocks != null && excessBlocks.contains(block)) {
* If there were any replication requests that timed out, reap them * If there were any replication requests that timed out, reap them
* and put them back into the neededReplication queue * and put them back into the neededReplication queue
*/ */
void processPendingReplications() { private void processPendingReplications() {
Block[] timedOutItems = pendingReplications.getTimedOutBlocks(); Block[] timedOutItems = pendingReplications.getTimedOutBlocks();
if (timedOutItems != null) { if (timedOutItems != null) {
namesystem.writeLock(); namesystem.writeLock();

View File

@ -1053,7 +1053,7 @@ private void refreshHostsReader(Configuration conf) throws IOException {
* 3. Added to exclude --> start decommission. * 3. Added to exclude --> start decommission.
* 4. Removed from exclude --> stop decommission. * 4. Removed from exclude --> stop decommission.
*/ */
void refreshDatanodes() { private void refreshDatanodes() {
for(DatanodeDescriptor node : datanodeMap.values()) { for(DatanodeDescriptor node : datanodeMap.values()) {
// Check if not include. // Check if not include.
if (!hostFileManager.isIncluded(node)) { if (!hostFileManager.isIncluded(node)) {
@ -1586,9 +1586,5 @@ public void clearPendingCachingCommands() {
public void setShouldSendCachingCommands(boolean shouldSendCachingCommands) { public void setShouldSendCachingCommands(boolean shouldSendCachingCommands) {
this.shouldSendCachingCommands = shouldSendCachingCommands; this.shouldSendCachingCommands = shouldSendCachingCommands;
} }
public HostFileManager getHostFileManager() {
return this.hostFileManager;
}
} }

View File

@ -129,10 +129,6 @@ synchronized boolean hasIncludes() {
void refresh(String includeFile, String excludeFile) throws IOException { void refresh(String includeFile, String excludeFile) throws IOException {
HostSet newIncludes = readFile("included", includeFile); HostSet newIncludes = readFile("included", includeFile);
HostSet newExcludes = readFile("excluded", excludeFile); HostSet newExcludes = readFile("excluded", excludeFile);
setHosts(newIncludes, newExcludes);
}
void setHosts(HostSet newIncludes, HostSet newExcludes) {
synchronized (this) { synchronized (this) {
includes = newIncludes; includes = newIncludes;
excludes = newExcludes; excludes = newExcludes;

View File

@ -978,7 +978,7 @@ private List<AuditLogger> initAuditLoggers(Configuration conf) {
return Collections.unmodifiableList(auditLoggers); return Collections.unmodifiableList(auditLoggers);
} }
protected void loadFSImage(StartupOption startOpt) throws IOException { private void loadFSImage(StartupOption startOpt) throws IOException {
final FSImage fsImage = getFSImage(); final FSImage fsImage = getFSImage();
// format before starting up if requested // format before starting up if requested
@ -1026,7 +1026,7 @@ protected void loadFSImage(StartupOption startOpt) throws IOException {
imageLoadComplete(); imageLoadComplete();
} }
protected void startSecretManager() { private void startSecretManager() {
if (dtSecretManager != null) { if (dtSecretManager != null) {
try { try {
dtSecretManager.startThreads(); dtSecretManager.startThreads();
@ -1038,7 +1038,7 @@ protected void startSecretManager() {
} }
} }
protected void startSecretManagerIfNecessary() { private void startSecretManagerIfNecessary() {
boolean shouldRun = shouldUseDelegationTokens() && boolean shouldRun = shouldUseDelegationTokens() &&
!isInSafeMode() && getEditLog().isOpenForWrite(); !isInSafeMode() && getEditLog().isOpenForWrite();
boolean running = dtSecretManager.isRunning(); boolean running = dtSecretManager.isRunning();
@ -1188,7 +1188,7 @@ public boolean inTransitionToActive() {
return haEnabled && inActiveState() && startingActiveService; return haEnabled && inActiveState() && startingActiveService;
} }
protected boolean shouldUseDelegationTokens() { private boolean shouldUseDelegationTokens() {
return UserGroupInformation.isSecurityEnabled() || return UserGroupInformation.isSecurityEnabled() ||
alwaysUseDelegationTokensForTests; alwaysUseDelegationTokensForTests;
} }
@ -2729,7 +2729,6 @@ private LocatedBlock appendFileInternal(FSPermissionChecker pc, String src,
* @throws UnresolvedLinkException * @throws UnresolvedLinkException
* @throws IOException * @throws IOException
*/ */
protected
LocatedBlock prepareFileForWrite(String src, INodeFile file, LocatedBlock prepareFileForWrite(String src, INodeFile file,
String leaseHolder, String clientMachine, String leaseHolder, String clientMachine,
boolean writeToEditLog, boolean writeToEditLog,
@ -3186,7 +3185,6 @@ FileState analyzeFileState(String src,
return new FileState(pendingFile, src); return new FileState(pendingFile, src);
} }
protected
LocatedBlock makeLocatedBlock(Block blk, DatanodeStorageInfo[] locs, LocatedBlock makeLocatedBlock(Block blk, DatanodeStorageInfo[] locs,
long offset) throws IOException { long offset) throws IOException {
LocatedBlock lBlk = new LocatedBlock( LocatedBlock lBlk = new LocatedBlock(
@ -3304,7 +3302,7 @@ boolean abandonBlock(ExtendedBlock b, long fileId, String src, String holder)
return true; return true;
} }
protected INodeFile checkLease(String src, String holder, INode inode, private INodeFile checkLease(String src, String holder, INode inode,
long fileId) long fileId)
throws LeaseExpiredException, FileNotFoundException { throws LeaseExpiredException, FileNotFoundException {
assert hasReadLock(); assert hasReadLock();
@ -4422,7 +4420,7 @@ Lease reassignLeaseInternal(Lease lease, String src, String newHolder,
return leaseManager.reassignLease(lease, src, newHolder); return leaseManager.reassignLease(lease, src, newHolder);
} }
protected void commitOrCompleteLastBlock(final INodeFile fileINode, private void commitOrCompleteLastBlock(final INodeFile fileINode,
final Block commitBlock) throws IOException { final Block commitBlock) throws IOException {
assert hasWriteLock(); assert hasWriteLock();
Preconditions.checkArgument(fileINode.isUnderConstruction()); Preconditions.checkArgument(fileINode.isUnderConstruction());
@ -4818,7 +4816,6 @@ String getRegistrationID() {
* @return an array of datanode commands * @return an array of datanode commands
* @throws IOException * @throws IOException
*/ */
protected
HeartbeatResponse handleHeartbeat(DatanodeRegistration nodeReg, HeartbeatResponse handleHeartbeat(DatanodeRegistration nodeReg,
StorageReport[] reports, long cacheCapacity, long cacheUsed, StorageReport[] reports, long cacheCapacity, long cacheUsed,
int xceiverCount, int xmitsInProgress, int failedVolumes) int xceiverCount, int xmitsInProgress, int failedVolumes)
@ -4868,7 +4865,7 @@ void checkAvailableResources() {
* @param file * @param file
* @param logRetryCache * @param logRetryCache
*/ */
protected void persistBlocks(String path, INodeFile file, private void persistBlocks(String path, INodeFile file,
boolean logRetryCache) { boolean logRetryCache) {
assert hasWriteLock(); assert hasWriteLock();
Preconditions.checkArgument(file.isUnderConstruction()); Preconditions.checkArgument(file.isUnderConstruction());
@ -5300,7 +5297,7 @@ void setBalancerBandwidth(long bandwidth) throws IOException {
* @param path * @param path
* @param file * @param file
*/ */
protected void persistNewBlock(String path, INodeFile file) { private void persistNewBlock(String path, INodeFile file) {
Preconditions.checkArgument(file.isUnderConstruction()); Preconditions.checkArgument(file.isUnderConstruction());
getEditLog().logAddBlock(path, file); getEditLog().logAddBlock(path, file);
if (NameNode.stateChangeLog.isDebugEnabled()) { if (NameNode.stateChangeLog.isDebugEnabled()) {
@ -7178,7 +7175,7 @@ private void logReassignLease(String leaseHolder, String src,
* *
* @return true if delegation token operation is allowed * @return true if delegation token operation is allowed
*/ */
protected boolean isAllowedDelegationTokenOp() throws IOException { private boolean isAllowedDelegationTokenOp() throws IOException {
AuthenticationMethod authMethod = getConnectionAuthenticationMethod(); AuthenticationMethod authMethod = getConnectionAuthenticationMethod();
if (UserGroupInformation.isSecurityEnabled() if (UserGroupInformation.isSecurityEnabled()
&& (authMethod != AuthenticationMethod.KERBEROS) && (authMethod != AuthenticationMethod.KERBEROS)
@ -7345,13 +7342,7 @@ public String getLiveNodes() {
final List<DatanodeDescriptor> live = new ArrayList<DatanodeDescriptor>(); final List<DatanodeDescriptor> live = new ArrayList<DatanodeDescriptor>();
blockManager.getDatanodeManager().fetchDatanodes(live, null, true); blockManager.getDatanodeManager().fetchDatanodes(live, null, true);
for (DatanodeDescriptor node : live) { for (DatanodeDescriptor node : live) {
info.put(node.getHostName(), getLiveNodeInfo(node)); Map<String, Object> innerinfo = ImmutableMap.<String, Object>builder()
}
return JSON.toString(info);
}
protected Map<String, Object> getLiveNodeInfo(DatanodeDescriptor node) {
return ImmutableMap.<String, Object>builder()
.put("infoAddr", node.getInfoAddr()) .put("infoAddr", node.getInfoAddr())
.put("infoSecureAddr", node.getInfoSecureAddr()) .put("infoSecureAddr", node.getInfoSecureAddr())
.put("xferaddr", node.getXferAddr()) .put("xferaddr", node.getXferAddr())
@ -7369,6 +7360,10 @@ protected Map<String, Object> getLiveNodeInfo(DatanodeDescriptor node) {
.put("blockPoolUsedPercent", node.getBlockPoolUsedPercent()) .put("blockPoolUsedPercent", node.getBlockPoolUsedPercent())
.put("volfails", node.getVolumeFailures()) .put("volfails", node.getVolumeFailures())
.build(); .build();
info.put(node.getHostName(), innerinfo);
}
return JSON.toString(info);
} }
/** /**
@ -7655,12 +7650,13 @@ public ReentrantLock getLongReadLockForTests() {
} }
@VisibleForTesting @VisibleForTesting
public void setNNResourceChecker(NameNodeResourceChecker nnResourceChecker) { public SafeModeInfo getSafeModeInfoForTests() {
this.nnResourceChecker = nnResourceChecker; return safeMode;
} }
public SafeModeInfo getSafeModeInfo() { @VisibleForTesting
return safeMode; public void setNNResourceChecker(NameNodeResourceChecker nnResourceChecker) {
this.nnResourceChecker = nnResourceChecker;
} }
@Override @Override

View File

@ -223,7 +223,7 @@ public static FSEditLogOp createMkdirOp(String path) {
* if safemode is not running. * if safemode is not running.
*/ */
public static int getSafeModeSafeBlocks(NameNode nn) { public static int getSafeModeSafeBlocks(NameNode nn) {
SafeModeInfo smi = nn.getNamesystem().getSafeModeInfo(); SafeModeInfo smi = nn.getNamesystem().getSafeModeInfoForTests();
if (smi == null) { if (smi == null) {
return -1; return -1;
} }