HBASE-28721 AsyncFSWAL is broken when running against hadoop 3.4.0 (#6270)

Signed-off-by: Duo Zhang <zhangduo@apache.org>
(cherry picked from commit c18921a)

Author: stoty

hbase-asyncfs/src/main/java/org/apache/hadoop/hbase/io/asyncfs/FanOutOneBlockAsyncDFSOutput.java

@@ -60,6 +60,7 @@
 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 @@ private void setupReceiver(int timeoutMs) {
   }
 
   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 void recoverAndClose(CancelableProgressable reporter) throws IOException
       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 void close() throws IOException {
     state = State.CLOSED;
     closeDataNodeChannelsAndAwait();
     block.setNumBytes(ackedBlockLength);
-    completeFile(client, namenode, src, clientName, block, fileId);
+    completeFile(client, namenode, src, clientName, block, stat);
   }
 
   @Override

hbase-asyncfs/src/main/java/org/apache/hadoop/hbase/io/asyncfs/FanOutOneBlockAsyncDFSOutputHelper.java

@@ -141,9 +141,9 @@ private 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 boolean isClientRunning(DFSClient client) {
     };
   }
 
-  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,25 +262,35 @@ private static LeaseManager createLeaseManager() throws NoSuchMethodException {
     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);
         }
       }
     };
   }
 
+  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 boolean progress() {
     }
   }
 
-  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 @@ private static FanOutOneBlockAsyncDFSOutput createOutput(DistributedFileSystem d
           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 @@ private static FanOutOneBlockAsyncDFSOutput createOutput(DistributedFileSystem d
         }
         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 void operationComplete(Future<Channel> future) throws Exception {
               });
             }
           }
-          endFileLease(client, stat.getFileId());
+          endFileLease(client, stat);
         }
       }
     }
@@ -654,11 +715,11 @@ public static boolean shouldRetryCreate(RemoteException e) {
   }
 
   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);