HDFS-15651. Client could not obtain block when DN CommandProcessingThread exit. Contributed by Aiphago.
Reviewed-by: He Xiaoqiao <hexiaoqiao@apache.org>
Reviewed-by: Yiqun Lin <yqlin@apache.org>
(cherry picked from commit 3067a25fa1
)
Conflicts:
hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestBPOfferService.java
This commit is contained in:
parent
872839416b
commit
ab451b8e42
@ -1312,6 +1312,10 @@ public void run() {
|
||||
processQueue();
|
||||
} catch (Throwable t) {
|
||||
LOG.error("{} encountered fatal exception and exit.", getName(), t);
|
||||
runningState = RunningState.FAILED;
|
||||
} finally {
|
||||
LOG.warn("Ending command processor service for: " + this);
|
||||
shouldServiceRun = false;
|
||||
}
|
||||
}
|
||||
|
||||
@ -1327,6 +1331,7 @@ private void processQueue() {
|
||||
dn.getMetrics().incrNumProcessedCommands();
|
||||
} catch (InterruptedException e) {
|
||||
LOG.error("{} encountered interrupt and exit.", getName());
|
||||
Thread.currentThread().interrupt();
|
||||
// ignore unless thread was specifically interrupted.
|
||||
if (Thread.interrupted()) {
|
||||
break;
|
||||
@ -1398,4 +1403,11 @@ void enqueue(DatanodeCommand[] cmds) throws InterruptedException {
|
||||
dn.getMetrics().incrActorCmdQueueLength(1);
|
||||
}
|
||||
}
|
||||
|
||||
@VisibleForTesting
|
||||
void stopCommandProcessingThread() {
|
||||
if (commandProcessingThread != null) {
|
||||
commandProcessingThread.interrupt();
|
||||
}
|
||||
}
|
||||
}
|
@ -1216,4 +1216,26 @@ public void testCommandProcessingThread() throws Exception {
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Test(timeout = 5000)
|
||||
public void testCommandProcessingThreadExit() throws Exception {
|
||||
Configuration conf = new HdfsConfiguration();
|
||||
MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf).
|
||||
numDataNodes(1).build();
|
||||
try {
|
||||
List<DataNode> datanodes = cluster.getDataNodes();
|
||||
DataNode dataNode = datanodes.get(0);
|
||||
List<BPOfferService> allBpOs = dataNode.getAllBpOs();
|
||||
BPOfferService bpos = allBpOs.get(0);
|
||||
waitForInitialization(bpos);
|
||||
BPServiceActor actor = bpos.getBPServiceActors().get(0);
|
||||
// Stop and wait util actor exit.
|
||||
actor.stopCommandProcessingThread();
|
||||
GenericTestUtils.waitFor(() -> !actor.isAlive(), 100, 3000);
|
||||
} finally {
|
||||
if (cluster != null) {
|
||||
cluster.shutdown();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
Loading…
Reference in New Issue
Block a user