You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by st...@apache.org on 2024/10/09 08:30:13 UTC
(hbase) branch branch-2.5 updated: HBASE-28721 AsyncFSWAL is broken when running against hadoop 3.4.0 (#6270)
This is an automated email from the ASF dual-hosted git repository.
stoty pushed a commit to branch branch-2.5
in repository https://gitbox.apache.org/repos/asf/hbase.git
The following commit(s) were added to refs/heads/branch-2.5 by this push:
new 586128742a7 HBASE-28721 AsyncFSWAL is broken when running against hadoop 3.4.0 (#6270)
586128742a7 is described below
commit 586128742a7538d5af67194c5a90b0462563d897
Author: Istvan Toth <st...@apache.org>
AuthorDate: Wed Oct 9 10:05:19 2024 +0200
HBASE-28721 AsyncFSWAL is broken when running against hadoop 3.4.0 (#6270)
Signed-off-by: Duo Zhang <zh...@apache.org>
(cherry picked from commit c18921a2868204c6a6268b7d7b7869a99e3df1bc)
---
.../io/asyncfs/FanOutOneBlockAsyncDFSOutput.java | 15 ++--
.../FanOutOneBlockAsyncDFSOutputHelper.java | 97 ++++++++++++++++++----
2 files changed, 87 insertions(+), 25 deletions(-)
diff --git a/hbase-asyncfs/src/main/java/org/apache/hadoop/hbase/io/asyncfs/FanOutOneBlockAsyncDFSOutput.java b/hbase-asyncfs/src/main/java/org/apache/hadoop/hbase/io/asyncfs/FanOutOneBlockAsyncDFSOutput.java
index 55a2f6c86ae..2a67d6e2742 100644
--- a/hbase-asyncfs/src/main/java/org/apache/hadoop/hbase/io/asyncfs/FanOutOneBlockAsyncDFSOutput.java
+++ b/hbase-asyncfs/src/main/java/org/apache/hadoop/hbase/io/asyncfs/FanOutOneBlockAsyncDFSOutput.java
@@ -60,6 +60,7 @@ import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.protocol.ClientProtocol;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
+import org.apache.hadoop.hdfs.protocol.HdfsFileStatus;
import org.apache.hadoop.hdfs.protocol.LocatedBlock;
import org.apache.hadoop.hdfs.protocol.datatransfer.PacketHeader;
import org.apache.hadoop.hdfs.protocol.datatransfer.PipelineAck;
@@ -121,7 +122,7 @@ public class FanOutOneBlockAsyncDFSOutput implements AsyncFSOutput {
private final String src;
- private final long fileId;
+ private HdfsFileStatus stat;
private final ExtendedBlock block;
@@ -354,14 +355,14 @@ public class FanOutOneBlockAsyncDFSOutput implements AsyncFSOutput {
}
FanOutOneBlockAsyncDFSOutput(Configuration conf, DistributedFileSystem dfs, DFSClient client,
- ClientProtocol namenode, String clientName, String src, long fileId, LocatedBlock locatedBlock,
- Encryptor encryptor, Map<Channel, DatanodeInfo> datanodeInfoMap, DataChecksum summer,
- ByteBufAllocator alloc, StreamSlowMonitor streamSlowMonitor) {
+ ClientProtocol namenode, String clientName, String src, HdfsFileStatus stat,
+ LocatedBlock locatedBlock, Encryptor encryptor, Map<Channel, DatanodeInfo> datanodeInfoMap,
+ DataChecksum summer, ByteBufAllocator alloc, StreamSlowMonitor streamSlowMonitor) {
this.conf = conf;
this.dfs = dfs;
this.client = client;
this.namenode = namenode;
- this.fileId = fileId;
+ this.stat = stat;
this.clientName = clientName;
this.src = src;
this.block = locatedBlock.getBlock();
@@ -592,7 +593,7 @@ public class FanOutOneBlockAsyncDFSOutput implements AsyncFSOutput {
buf = null;
}
closeDataNodeChannelsAndAwait();
- endFileLease(client, fileId);
+ endFileLease(client, stat);
RecoverLeaseFSUtils.recoverFileLease(dfs, new Path(src), conf,
reporter == null ? new CancelOnClose(client) : reporter);
}
@@ -607,7 +608,7 @@ public class FanOutOneBlockAsyncDFSOutput implements AsyncFSOutput {
state = State.CLOSED;
closeDataNodeChannelsAndAwait();
block.setNumBytes(ackedBlockLength);
- completeFile(client, namenode, src, clientName, block, fileId);
+ completeFile(client, namenode, src, clientName, block, stat);
}
@Override
diff --git a/hbase-asyncfs/src/main/java/org/apache/hadoop/hbase/io/asyncfs/FanOutOneBlockAsyncDFSOutputHelper.java b/hbase-asyncfs/src/main/java/org/apache/hadoop/hbase/io/asyncfs/FanOutOneBlockAsyncDFSOutputHelper.java
index 6ff200c4ac3..669727ea59e 100644
--- a/hbase-asyncfs/src/main/java/org/apache/hadoop/hbase/io/asyncfs/FanOutOneBlockAsyncDFSOutputHelper.java
+++ b/hbase-asyncfs/src/main/java/org/apache/hadoop/hbase/io/asyncfs/FanOutOneBlockAsyncDFSOutputHelper.java
@@ -141,9 +141,9 @@ public final class FanOutOneBlockAsyncDFSOutputHelper {
private interface LeaseManager {
- void begin(DFSClient client, long inodeId);
+ void begin(DFSClient client, HdfsFileStatus stat);
- void end(DFSClient client, long inodeId);
+ void end(DFSClient client, HdfsFileStatus stat);
}
private static final LeaseManager LEASE_MANAGER;
@@ -202,7 +202,58 @@ public final class FanOutOneBlockAsyncDFSOutputHelper {
};
}
- private static LeaseManager createLeaseManager() throws NoSuchMethodException {
+ private static LeaseManager createLeaseManager3_4() throws NoSuchMethodException {
+ Method beginFileLeaseMethod =
+ DFSClient.class.getDeclaredMethod("beginFileLease", String.class, DFSOutputStream.class);
+ beginFileLeaseMethod.setAccessible(true);
+ Method endFileLeaseMethod = DFSClient.class.getDeclaredMethod("endFileLease", String.class);
+ endFileLeaseMethod.setAccessible(true);
+ Method getConfigurationMethod = DFSClient.class.getDeclaredMethod("getConfiguration");
+ getConfigurationMethod.setAccessible(true);
+ Method getNamespaceMehtod = HdfsFileStatus.class.getDeclaredMethod("getNamespace");
+
+ return new LeaseManager() {
+
+ private static final String DFS_OUTPUT_STREAM_UNIQ_DEFAULT_KEY =
+ "dfs.client.output.stream.uniq.default.key";
+ private static final String DFS_OUTPUT_STREAM_UNIQ_DEFAULT_KEY_DEFAULT = "DEFAULT";
+
+ private String getUniqId(DFSClient client, HdfsFileStatus stat)
+ throws IllegalAccessException, IllegalArgumentException, InvocationTargetException {
+ // Copied from DFSClient in Hadoop 3.4.0
+ long fileId = stat.getFileId();
+ String namespace = (String) getNamespaceMehtod.invoke(stat);
+ if (namespace == null) {
+ Configuration conf = (Configuration) getConfigurationMethod.invoke(client);
+ String defaultKey = conf.get(DFS_OUTPUT_STREAM_UNIQ_DEFAULT_KEY,
+ DFS_OUTPUT_STREAM_UNIQ_DEFAULT_KEY_DEFAULT);
+ return defaultKey + "_" + fileId;
+ } else {
+ return namespace + "_" + fileId;
+ }
+ }
+
+ @Override
+ public void begin(DFSClient client, HdfsFileStatus stat) {
+ try {
+ beginFileLeaseMethod.invoke(client, getUniqId(client, stat), null);
+ } catch (IllegalAccessException | InvocationTargetException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public void end(DFSClient client, HdfsFileStatus stat) {
+ try {
+ endFileLeaseMethod.invoke(client, getUniqId(client, stat));
+ } catch (IllegalAccessException | InvocationTargetException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ };
+ }
+
+ private static LeaseManager createLeaseManager3() throws NoSuchMethodException {
Method beginFileLeaseMethod =
DFSClient.class.getDeclaredMethod("beginFileLease", long.class, DFSOutputStream.class);
beginFileLeaseMethod.setAccessible(true);
@@ -211,18 +262,18 @@ public final class FanOutOneBlockAsyncDFSOutputHelper {
return new LeaseManager() {
@Override
- public void begin(DFSClient client, long inodeId) {
+ public void begin(DFSClient client, HdfsFileStatus stat) {
try {
- beginFileLeaseMethod.invoke(client, inodeId, null);
+ beginFileLeaseMethod.invoke(client, stat.getFileId(), null);
} catch (IllegalAccessException | InvocationTargetException e) {
throw new RuntimeException(e);
}
}
@Override
- public void end(DFSClient client, long inodeId) {
+ public void end(DFSClient client, HdfsFileStatus stat) {
try {
- endFileLeaseMethod.invoke(client, inodeId);
+ endFileLeaseMethod.invoke(client, stat.getFileId());
} catch (IllegalAccessException | InvocationTargetException e) {
throw new RuntimeException(e);
}
@@ -230,6 +281,16 @@ public final class FanOutOneBlockAsyncDFSOutputHelper {
};
}
+ private static LeaseManager createLeaseManager() throws NoSuchMethodException {
+ try {
+ return createLeaseManager3_4();
+ } catch (NoSuchMethodException e) {
+ LOG.debug("DFSClient::beginFileLease wrong arguments, should be hadoop 3.3 or below");
+ }
+
+ return createLeaseManager3();
+ }
+
private static FileCreator createFileCreator3_3() throws NoSuchMethodException {
Method createMethod = ClientProtocol.class.getMethod("create", String.class, FsPermission.class,
String.class, EnumSetWritable.class, boolean.class, short.class, long.class,
@@ -320,12 +381,12 @@ public final class FanOutOneBlockAsyncDFSOutputHelper {
}
}
- static void beginFileLease(DFSClient client, long inodeId) {
- LEASE_MANAGER.begin(client, inodeId);
+ static void beginFileLease(DFSClient client, HdfsFileStatus stat) {
+ LEASE_MANAGER.begin(client, stat);
}
- static void endFileLease(DFSClient client, long inodeId) {
- LEASE_MANAGER.end(client, inodeId);
+ static void endFileLease(DFSClient client, HdfsFileStatus stat) {
+ LEASE_MANAGER.end(client, stat);
}
static DataChecksum createChecksum(DFSClient client) {
@@ -552,7 +613,7 @@ public final class FanOutOneBlockAsyncDFSOutputHelper {
throw new NameNodeException(e);
}
}
- beginFileLease(client, stat.getFileId());
+ beginFileLease(client, stat);
boolean succ = false;
LocatedBlock locatedBlock = null;
List<Future<Channel>> futureList = null;
@@ -576,8 +637,8 @@ public final class FanOutOneBlockAsyncDFSOutputHelper {
}
Encryptor encryptor = createEncryptor(conf, stat, client);
FanOutOneBlockAsyncDFSOutput output =
- new FanOutOneBlockAsyncDFSOutput(conf, dfs, client, namenode, clientName, src,
- stat.getFileId(), locatedBlock, encryptor, datanodes, summer, ALLOC, monitor);
+ new FanOutOneBlockAsyncDFSOutput(conf, dfs, client, namenode, clientName, src, stat,
+ locatedBlock, encryptor, datanodes, summer, ALLOC, monitor);
succ = true;
return output;
} catch (RemoteException e) {
@@ -616,7 +677,7 @@ public final class FanOutOneBlockAsyncDFSOutputHelper {
});
}
}
- endFileLease(client, stat.getFileId());
+ endFileLease(client, stat);
}
}
}
@@ -654,11 +715,11 @@ public final class FanOutOneBlockAsyncDFSOutputHelper {
}
static void completeFile(DFSClient client, ClientProtocol namenode, String src, String clientName,
- ExtendedBlock block, long fileId) {
+ ExtendedBlock block, HdfsFileStatus stat) {
for (int retry = 0;; retry++) {
try {
- if (namenode.complete(src, clientName, block, fileId)) {
- endFileLease(client, fileId);
+ if (namenode.complete(src, clientName, block, stat.getFileId())) {
+ endFileLease(client, stat);
return;
} else {
LOG.warn("complete file " + src + " not finished, retry = " + retry);