HADOOP-19650. [ABFS] Fixing NPE when close() called on uninitialized AzureBlobFileSystem (#7880) (#7888)

Contributed by Anuj Modi

Author: anujmodi2021

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AzureBlobFileSystem.java

@@ -131,6 +131,7 @@
 import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.DATA_BLOCKS_BUFFER_DEFAULT;
 import static org.apache.hadoop.fs.azurebfs.constants.InternalConstants.CAPABILITY_SAFE_READAHEAD;
 import static org.apache.hadoop.fs.azurebfs.services.AbfsErrors.ERR_CREATE_ON_ROOT;
+import static org.apache.hadoop.fs.azurebfs.services.AbfsErrors.ERR_INVALID_ABFS_STATE;
 import static org.apache.hadoop.fs.azurebfs.services.AbfsErrors.UNAUTHORIZED_SAS;
 import static org.apache.hadoop.fs.impl.PathCapabilitiesSupport.validatePathCapabilityArgs;
 import static org.apache.hadoop.fs.statistics.IOStatisticsLogging.logIOStatisticsAtLevel;
@@ -148,7 +149,11 @@ public class AzureBlobFileSystem extends FileSystem
   private URI uri;
   private Path workingDir;
   private AzureBlobFileSystemStore abfsStore;
-  private boolean isClosed;
+
+  /**
+   * Flag to indicate whether the file system is closed or not initiated.
+   */
+  private boolean isClosed = true;
   private final String fileSystemId = UUID.randomUUID().toString();
 
   private boolean delegationTokenEnabled = false;
@@ -311,6 +316,7 @@ public void initialize(URI uri, Configuration configuration)
     }
 
     rateLimiting = RateLimitingFactory.create(abfsConfiguration.getRateLimit());
+    isClosed = false;
     LOG.debug("Initializing AzureBlobFileSystem for {} complete", uri);
   }
 
@@ -328,8 +334,8 @@ public String toString() {
     final StringBuilder sb = new StringBuilder(
         "AzureBlobFileSystem{");
     sb.append("uri=").append(fullPathUri);
-    sb.append(", user='").append(abfsStore.getUser()).append('\'');
-    sb.append(", primaryUserGroup='").append(abfsStore.getPrimaryGroup()).append('\'');
+    sb.append(", user='").append(getAbfsStore().getUser()).append('\'');
+    sb.append(", primaryUserGroup='").append(getAbfsStore().getPrimaryGroup()).append('\'');
     sb.append("[" + CAPABILITY_SAFE_READAHEAD + "]");
     sb.append('}');
     return sb.toString();
@@ -353,7 +359,7 @@ public FSDataInputStream open(final Path path, final int bufferSize) throws IOEx
     // bufferSize is unused.
     LOG.debug(
         "AzureBlobFileSystem.open path: {} bufferSize as configured in 'fs.azure.read.request.size': {}",
-        path, abfsStore.getAbfsConfiguration().getReadBufferSize());
+        path, getAbfsStore().getAbfsConfiguration().getReadBufferSize());
     return open(path, Optional.empty());
   }
 
@@ -516,7 +522,7 @@ public FSDataOutputStream append(final Path f, final int bufferSize, final Progr
       TracingContext tracingContext = new TracingContext(clientCorrelationId,
           fileSystemId, FSOperationType.APPEND, tracingHeaderFormat,
           listener);
-      OutputStream outputStream = abfsStore
+      OutputStream outputStream = getAbfsStore()
           .openFileForWrite(qualifiedPath, statistics, false, tracingContext);
       return new FSDataOutputStream(outputStream, statistics);
     } catch (AzureBlobFileSystemException ex) {
@@ -781,7 +787,7 @@ public boolean mkdirs(final Path f, final FsPermission permission) throws IOExce
       TracingContext tracingContext = new TracingContext(clientCorrelationId,
           fileSystemId, FSOperationType.MKDIR, false, tracingHeaderFormat,
           listener);
-      abfsStore.createDirectory(qualifiedPath,
+      getAbfsStore().createDirectory(qualifiedPath,
           permission == null ? FsPermission.getDirDefault() : permission,
           FsPermission.getUMask(getConf()), tracingContext);
       statIncrement(DIRECTORIES_CREATED);
@@ -794,10 +800,10 @@ public boolean mkdirs(final Path f, final FsPermission permission) throws IOExce
 
   @Override
   public synchronized void close() throws IOException {
-    if (isClosed) {
+    if (isClosed()) {
       return;
     }
-    if (abfsStore.getClient().isMetricCollectionEnabled()) {
+    if (getAbfsStore().getClient().isMetricCollectionEnabled()) {
       TracingContext tracingMetricContext = new TracingContext(
               clientCorrelationId,
               fileSystemId, FSOperationType.GET_ATTR, true,
@@ -818,7 +824,7 @@ public synchronized void close() throws IOException {
               IOSTATISTICS_LOGGING_LEVEL_DEFAULT);
       logIOStatisticsAtLevel(LOG, iostatisticsLoggingLevel, getIOStatistics());
     }
-    IOUtils.cleanupWithLogger(LOG, abfsStore, delegationTokenManager,
+    IOUtils.cleanupWithLogger(LOG, getAbfsStore(), delegationTokenManager,
         getAbfsClient());
     this.isClosed = true;
     if (LOG.isDebugEnabled()) {
@@ -865,7 +871,7 @@ public void breakLease(final Path f) throws IOException {
       TracingContext tracingContext = new TracingContext(clientCorrelationId,
           fileSystemId, FSOperationType.BREAK_LEASE, tracingHeaderFormat,
           listener);
-      abfsStore.breakLease(qualifiedPath, tracingContext);
+      getAbfsStore().breakLease(qualifiedPath, tracingContext);
     } catch (AzureBlobFileSystemException ex) {
       checkException(f, ex);
     }
@@ -883,6 +889,8 @@ public void breakLease(final Path f) throws IOException {
    */
   @Override
   public Path makeQualified(Path path) {
+    // Every API works on qualified paths. If store is null better to fail early.
+    Preconditions.checkState(getAbfsStore() != null);
     // To support format: abfs://{dfs.nameservices}/file/path,
     // path need to be first converted to URI, then get the raw path string,
     // during which {dfs.nameservices} will be omitted.
@@ -916,7 +924,7 @@ public String getScheme() {
   public Path getHomeDirectory() {
     return makeQualified(new Path(
         FileSystemConfigurations.USER_HOME_DIRECTORY_PREFIX
-            + "/" + abfsStore.getUser()));
+            + "/" + getAbfsStore().getUser()));
   }
 
   /**
@@ -938,7 +946,7 @@ public BlockLocation[] getFileBlockLocations(FileStatus file,
     if (file.getLen() < start) {
       return new BlockLocation[0];
     }
-    final String blobLocationHost = abfsStore.getAbfsConfiguration().getAzureBlockLocationHost();
+    final String blobLocationHost = getAbfsStore().getAbfsConfiguration().getAzureBlockLocationHost();
 
     final String[] name = {blobLocationHost};
     final String[] host = {blobLocationHost};
@@ -972,15 +980,15 @@ protected void finalize() throws Throwable {
    * @return the short name of the user who instantiated the FS
    */
   public String getOwnerUser() {
-    return abfsStore.getUser();
+    return getAbfsStore().getUser();
   }
 
   /**
    * Get the group name of the owner of the FS.
    * @return primary group name
    */
   public String getOwnerUserPrimaryGroup() {
-    return abfsStore.getPrimaryGroup();
+    return getAbfsStore().getPrimaryGroup();
   }
 
   private boolean deleteRoot() throws IOException {
@@ -1052,7 +1060,7 @@ public void setOwner(final Path path, final String owner, final String group)
     Path qualifiedPath = makeQualified(path);
 
     try {
-      abfsStore.setOwner(qualifiedPath,
+      getAbfsStore().setOwner(qualifiedPath,
           owner,
           group,
           tracingContext);
@@ -1089,15 +1097,15 @@ public void setXAttr(final Path path,
       TracingContext tracingContext = new TracingContext(clientCorrelationId,
           fileSystemId, FSOperationType.SET_ATTR, true, tracingHeaderFormat,
           listener);
-      Hashtable<String, String> properties = abfsStore
+      Hashtable<String, String> properties = getAbfsStore()
           .getPathStatus(qualifiedPath, tracingContext);
       String xAttrName = ensureValidAttributeName(name);
       boolean xAttrExists = properties.containsKey(xAttrName);
       XAttrSetFlag.validate(name, xAttrExists, flag);
 
-      String xAttrValue = abfsStore.decodeAttribute(value);
+      String xAttrValue = getAbfsStore().decodeAttribute(value);
       properties.put(xAttrName, xAttrValue);
-      abfsStore.setPathProperties(qualifiedPath, properties, tracingContext);
+      getAbfsStore().setPathProperties(qualifiedPath, properties, tracingContext);
     } catch (AzureBlobFileSystemException ex) {
       checkException(path, ex);
     }
@@ -1129,12 +1137,12 @@ public byte[] getXAttr(final Path path, final String name)
       TracingContext tracingContext = new TracingContext(clientCorrelationId,
           fileSystemId, FSOperationType.GET_ATTR, true, tracingHeaderFormat,
           listener);
-      Hashtable<String, String> properties = abfsStore
+      Hashtable<String, String> properties = getAbfsStore()
           .getPathStatus(qualifiedPath, tracingContext);
       String xAttrName = ensureValidAttributeName(name);
       if (properties.containsKey(xAttrName)) {
         String xAttrValue = properties.get(xAttrName);
-        value = abfsStore.encodeAttribute(xAttrValue);
+        value = getAbfsStore().encodeAttribute(xAttrValue);
       }
     } catch (AzureBlobFileSystemException ex) {
       checkException(path, ex);
@@ -1172,7 +1180,7 @@ public void setPermission(final Path path, final FsPermission permission)
     Path qualifiedPath = makeQualified(path);
 
     try {
-      abfsStore.setPermission(qualifiedPath, permission, tracingContext);
+      getAbfsStore().setPermission(qualifiedPath, permission, tracingContext);
     } catch (AzureBlobFileSystemException ex) {
       checkException(path, ex);
     }
@@ -1209,7 +1217,7 @@ public void modifyAclEntries(final Path path, final List<AclEntry> aclSpec)
     Path qualifiedPath = makeQualified(path);
 
     try {
-      abfsStore.modifyAclEntries(qualifiedPath, aclSpec, tracingContext);
+      getAbfsStore().modifyAclEntries(qualifiedPath, aclSpec, tracingContext);
     } catch (AzureBlobFileSystemException ex) {
       checkException(path, ex);
     }
@@ -1244,7 +1252,7 @@ public void removeAclEntries(final Path path, final List<AclEntry> aclSpec)
     Path qualifiedPath = makeQualified(path);
 
     try {
-      abfsStore.removeAclEntries(qualifiedPath, aclSpec, tracingContext);
+      getAbfsStore().removeAclEntries(qualifiedPath, aclSpec, tracingContext);
     } catch (AzureBlobFileSystemException ex) {
       checkException(path, ex);
     }
@@ -1272,7 +1280,7 @@ public void removeDefaultAcl(final Path path) throws IOException {
     Path qualifiedPath = makeQualified(path);
 
     try {
-      abfsStore.removeDefaultAcl(qualifiedPath, tracingContext);
+      getAbfsStore().removeDefaultAcl(qualifiedPath, tracingContext);
     } catch (AzureBlobFileSystemException ex) {
       checkException(path, ex);
     }
@@ -1302,7 +1310,7 @@ public void removeAcl(final Path path) throws IOException {
     Path qualifiedPath = makeQualified(path);
 
     try {
-      abfsStore.removeAcl(qualifiedPath, tracingContext);
+      getAbfsStore().removeAcl(qualifiedPath, tracingContext);
     } catch (AzureBlobFileSystemException ex) {
       checkException(path, ex);
     }
@@ -1339,7 +1347,7 @@ public void setAcl(final Path path, final List<AclEntry> aclSpec)
     Path qualifiedPath = makeQualified(path);
 
     try {
-      abfsStore.setAcl(qualifiedPath, aclSpec, tracingContext);
+      getAbfsStore().setAcl(qualifiedPath, aclSpec, tracingContext);
     } catch (AzureBlobFileSystemException ex) {
       checkException(path, ex);
     }
@@ -1367,7 +1375,7 @@ public AclStatus getAclStatus(final Path path) throws IOException {
     Path qualifiedPath = makeQualified(path);
 
     try {
-      return abfsStore.getAclStatus(qualifiedPath, tracingContext);
+      return getAbfsStore().getAclStatus(qualifiedPath, tracingContext);
     } catch (AzureBlobFileSystemException ex) {
       checkException(path, ex);
       return null;
@@ -1394,7 +1402,7 @@ public void access(final Path path, final FsAction mode) throws IOException {
       TracingContext tracingContext = new TracingContext(clientCorrelationId,
           fileSystemId, FSOperationType.ACCESS, tracingHeaderFormat,
           listener);
-      this.abfsStore.access(qualifiedPath, mode, tracingContext);
+      this.getAbfsStore().access(qualifiedPath, mode, tracingContext);
     } catch (AzureBlobFileSystemException ex) {
       checkCheckAccessException(path, ex);
     }
@@ -1416,11 +1424,11 @@ public boolean exists(Path f) throws IOException {
   public RemoteIterator<FileStatus> listStatusIterator(Path path)
       throws IOException {
     LOG.debug("AzureBlobFileSystem.listStatusIterator path : {}", path);
-    if (abfsStore.getAbfsConfiguration().enableAbfsListIterator()) {
+    if (getAbfsStore().getAbfsConfiguration().enableAbfsListIterator()) {
       TracingContext tracingContext = new TracingContext(clientCorrelationId,
           fileSystemId, FSOperationType.LISTSTATUS, true, tracingHeaderFormat, listener);
       AbfsListStatusRemoteIterator abfsLsItr =
-          new AbfsListStatusRemoteIterator(path, abfsStore,
+          new AbfsListStatusRemoteIterator(path, getAbfsStore(),
               tracingContext);
       return RemoteIterators.typeCastingRemoteIterator(abfsLsItr);
     } else {
@@ -1502,7 +1510,7 @@ private boolean fileSystemExists() throws IOException {
     try {
       TracingContext tracingContext = new TracingContext(clientCorrelationId,
           fileSystemId, FSOperationType.TEST_OP, tracingHeaderFormat, listener);
-      abfsStore.getFilesystemProperties(tracingContext);
+      getAbfsStore().getFilesystemProperties(tracingContext);
     } catch (AzureBlobFileSystemException ex) {
       try {
         checkException(null, ex);
@@ -1521,7 +1529,7 @@ private void createFileSystem(TracingContext tracingContext) throws IOException
     LOG.debug(
         "AzureBlobFileSystem.createFileSystem uri: {}", uri);
     try {
-      abfsStore.createFilesystem(tracingContext);
+      getAbfsStore().createFilesystem(tracingContext);
     } catch (AzureBlobFileSystemException ex) {
       checkException(null, ex);
     }
@@ -1744,14 +1752,21 @@ public boolean failed() {
 
   @VisibleForTesting
   public AzureBlobFileSystemStore getAbfsStore() {
+    if (abfsStore == null) {
+      throw new IllegalStateException(ERR_INVALID_ABFS_STATE);
+    }
     return abfsStore;
   }
 
   @VisibleForTesting
   AbfsClient getAbfsClient() {
-    return abfsStore.getClient();
+    return getAbfsStore().getClient();
   }
 
+  @VisibleForTesting
+  boolean isClosed() {
+    return isClosed;
+  }
   /**
    * Get any Delegation Token manager created by the filesystem.
    * @return the DT manager or null.

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsErrors.java

@@ -77,5 +77,6 @@ public final class AbfsErrors {
   public static final String ERR_BLOB_LIST_PARSING = "Parsing of XML List Response Failed in BlobClient.";
   public static final String ERR_DFS_LIST_PARSING = "Parsing of Json List Response Failed in DfsClient.";
   public static final String INCORRECT_INGRESS_TYPE = "Ingress Type Cannot be DFS for Blob endpoint configured filesystem.";
+  public static final String ERR_INVALID_ABFS_STATE = "Invalid state for AzureBlobFilesystem. Either Filesystem was closed or not initialized.";
   private AbfsErrors() {}
 }