big-data/hadoop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

HDFS-17024. Potential data race introduced by HDFS-15865 (#6223)

Browse Source
This commit is contained in:
Hiroaki Segawa 2023-10-27 14:25:00 +09:00 committed by GitHub
parent 652908519e
commit 93a3c6e2cd
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 8 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DataStreamer.java
View File

@ -476,6 +476,7 @@ boolean doWaitForRestart() {
private DataOutputStream blockStream; private DataOutputStream blockStream;
private DataInputStream blockReplyStream; private DataInputStream blockReplyStream;
private ResponseProcessor response = null; private ResponseProcessor response = null;
private final Object nodesLock = new Object();
private volatile DatanodeInfo[] nodes = null; // list of targets for current block private volatile DatanodeInfo[] nodes = null; // list of targets for current block
private volatile StorageType[] storageTypes = null; private volatile StorageType[] storageTypes = null;
private volatile String[] storageIDs = null; private volatile String[] storageIDs = null;
@ -619,7 +620,9 @@ private void setPipeline(LocatedBlock lb) {
private void setPipeline(DatanodeInfo[] nodes, StorageType[] storageTypes, private void setPipeline(DatanodeInfo[] nodes, StorageType[] storageTypes,
String[] storageIDs) { String[] storageIDs) {
synchronized (nodesLock) {
this.nodes = nodes; this.nodes = nodes;
}
this.storageTypes = storageTypes; this.storageTypes = storageTypes;
this.storageIDs = storageIDs; this.storageIDs = storageIDs;
} }
@ -916,7 +919,10 @@ void waitForAckedSeqno(long seqno) throws IOException {
try (TraceScope ignored = dfsClient.getTracer(). try (TraceScope ignored = dfsClient.getTracer().
newScope("waitForAckedSeqno")) { newScope("waitForAckedSeqno")) {
LOG.debug("{} waiting for ack for: {}", this, seqno); LOG.debug("{} waiting for ack for: {}", this, seqno);
int dnodes = nodes != null ? nodes.length : 3; int dnodes;
synchronized (nodesLock) {
dnodes = nodes != null ? nodes.length : 3;
}
int writeTimeout = dfsClient.getDatanodeWriteTimeout(dnodes); int writeTimeout = dfsClient.getDatanodeWriteTimeout(dnodes);
long begin = Time.monotonicNow(); long begin = Time.monotonicNow();
try { try {