This is an automated email from the ASF dual-hosted git repository.
hexiaoqiao pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/hadoop.git
The following commit(s) were added to refs/heads/trunk by this push:
new 3067a25 HDFS-15651. Client could not obtain block when DN
CommandProcessingThread exit. Contributed by Aiphago.
3067a25 is described below
commit 3067a25fa12a012709f43aa35cc606db6fb137f9
Author: He Xiaoqiao <[email protected]>
AuthorDate: Wed Nov 4 13:53:46 2020 +0800
HDFS-15651. Client could not obtain block when DN CommandProcessingThread
exit. Contributed by Aiphago.
Reviewed-by: He Xiaoqiao <[email protected]>
Reviewed-by: Yiqun Lin <[email protected]>
---
.../hdfs/server/datanode/BPServiceActor.java | 12 +++++++++++
.../hdfs/server/datanode/TestBPOfferService.java | 24 +++++++++++++++++++++-
2 files changed, 35 insertions(+), 1 deletion(-)
diff --git
a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BPServiceActor.java
b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BPServiceActor.java
index 49156c2..0cc9fac 100755
---
a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BPServiceActor.java
+++
b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BPServiceActor.java
@@ -1312,6 +1312,10 @@ class BPServiceActor implements Runnable {
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 @@ class BPServiceActor implements Runnable {
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 @@ class BPServiceActor implements Runnable {
dn.getMetrics().incrActorCmdQueueLength(1);
}
}
+
+ @VisibleForTesting
+ void stopCommandProcessingThread() {
+ if (commandProcessingThread != null) {
+ commandProcessingThread.interrupt();
+ }
+ }
}
\ No newline at end of file
diff --git
a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestBPOfferService.java
b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestBPOfferService.java
index 19881dc..45355d4 100644
---
a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestBPOfferService.java
+++
b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestBPOfferService.java
@@ -1211,4 +1211,26 @@ public class TestBPOfferService {
}
}
}
-}
+
+ @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();
+ }
+ }
+ }
+}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
