[BugFix] Avoid hdfs fs manager interrupting the thread when exception…

… occurs (#48403) Signed-off-by: xiangguangyxg <[email protected]> (cherry picked from commit 3fcafac) # Conflicts: # fe/fe-core/src/main/java/com/starrocks/fs/hdfs/HdfsFsManager.java
StarRocks · Jul 22, 2024 · 0f2362a · 0f2362a
1 parent d02b772
commit 0f2362a
Showing 1 changed file with 87 additions and 0 deletions.
diff --git a/fe/fe-core/src/main/java/com/starrocks/fs/hdfs/HdfsFsManager.java b/fe/fe-core/src/main/java/com/starrocks/fs/hdfs/HdfsFsManager.java
@@ -46,6 +46,7 @@
 
 import java.io.FileNotFoundException;
 import java.io.IOException;
+import java.io.InterruptedIOException;
 import java.security.PrivilegedExceptionAction;
 import java.util.ArrayList;
 import java.util.Enumeration;
@@ -1077,6 +1078,29 @@ public void getTProperties(String path, Map<String, String> loadProperties, THdf
  getFileSystem(path, loadProperties, tProperties);
  }
 
+<<<<<<< HEAD
+=======
+ public List<FileStatus> listFileMeta(String path, Map<String, String> properties) throws UserException {
+ WildcardURI pathUri = new WildcardURI(path);
+ HdfsFs fileSystem = getFileSystem(path, properties, null);
+ Path pathPattern = new Path(pathUri.getPath());
+ try {
+ FileStatus[] files = fileSystem.getDFSFileSystem().globStatus(pathPattern);
+ return Lists.newArrayList(files);
+ } catch (FileNotFoundException e) {
+ LOG.info("file not found: " + path, e);
+ throw new UserException("file not found: " + path, e);
+ } catch (InterruptedIOException e) {
+ Thread.interrupted(); // clear interrupted flag
+ LOG.error("Interrupted while get file status: " + path, e);
+ throw new UserException("Failed to get file status: " + path, e); // throw unified user exception
+ } catch (Exception e) {
+ LOG.error("errors while get file status ", e);
+ throw new UserException("Fail to get file status: " + e.getMessage(), e);
+ }
+ }
+
+>>>>>>> 3fcafac8b9 ([BugFix] Avoid hdfs fs manager interrupting the thread when exception occurs (#48403))
  public List<TBrokerFileStatus> listPath(String path, boolean fileNameOnly, Map<String, String> loadProperties)
  throws UserException {
  List<TBrokerFileStatus> resultFileStatus = null;
@@ -1110,8 +1134,22 @@ public List<TBrokerFileStatus> listPath(String path, boolean fileNameOnly, Map<S
  resultFileStatus.add(brokerFileStatus);
  }
  } catch (FileNotFoundException e) {
+<<<<<<< HEAD
  LOG.info("file not found: " + e.getMessage());
  throw new UserException("file not found: " + e.getMessage());
+=======
+ LOG.info("file not found: " + path, e);
+ throw new UserException("file not found: " + path, e);
+ } catch (IllegalArgumentException e) {
+ LOG.error("The arguments of blob store(S3/Azure) may be wrong. You can check " +
+ "the arguments like region, IAM, instance profile and so on.");
+ throw new UserException("The arguments of blob store(S3/Azure) may be wrong. " +
+ "You can check the arguments like region, IAM, instance profile and so on.", e);
+ } catch (InterruptedIOException e) {
+ Thread.interrupted(); // clear interrupted flag
+ LOG.error("Interrupted while list path: " + path, e);
+ throw new UserException("Failed to list path: " + path, e); // throw unified user exception
+>>>>>>> 3fcafac8b9 ([BugFix] Avoid hdfs fs manager interrupting the thread when exception occurs (#48403))
  } catch (Exception e) {
  LOG.error("errors while get file status ", e);
  throw new UserException("unknown error when get file status: " + e.getMessage());
@@ -1125,6 +1163,10 @@ public void deletePath(String path, Map<String, String> loadProperties) throws U
  Path filePath = new Path(pathUri.getPath());
  try {
  fileSystem.getDFSFileSystem().delete(filePath, true);
+ } catch (InterruptedIOException e) {
+ Thread.interrupted(); // clear interrupted flag
+ LOG.error("Interrupted while delete path: " + path, e);
+ throw new UserException("Failed to delete path: " + path, e); // throw unified user exception
  } catch (IOException e) {
  LOG.error("errors while delete path " + path, e);
  throw new UserException("delete path " + path + "error", e);
@@ -1155,6 +1197,11 @@ public void renamePath(String srcPath, String destPath, Map<String, String> load
  if (!isRenameSuccess) {
  throw new UserException("failed to rename path from " + srcPath + " to " + destPath);
  }
+ } catch (InterruptedIOException e) {
+ Thread.interrupted(); // clear interrupted flag
+ LOG.error("Interrupted while rename path from " + srcPath + " to " + destPath, e);
+ // throw unified user exception
+ throw new UserException("Failed to rename path from " + srcPath + " to " + destPath, e);
  } catch (IOException e) {
  LOG.error("errors while rename path from " + srcPath + " to " + destPath, e);
  throw new UserException("errors while rename " + srcPath + "to " + destPath, e);
@@ -1167,6 +1214,10 @@ public boolean checkPathExist(String path, Map<String, String> loadProperties) t
  Path filePath = new Path(pathUri.getPath());
  try {
  return fileSystem.getDFSFileSystem().exists(filePath);
+ } catch (InterruptedIOException e) {
+ Thread.interrupted(); // clear interrupted flag
+ LOG.error("Interrupted while check path exist: " + path, e);
+ throw new UserException("Failed to check path exist: " + path, e); // throw unified user exception
  } catch (IOException e) {
  LOG.error("errors while check path exist: " + path, e);
  throw new UserException("errors while check if path " + path + " exist", e);
@@ -1184,6 +1235,10 @@ public TBrokerFD openReader(String path, long startOffset, Map<String, String> l
  TBrokerFD fd = parseUUIDToFD(uuid);
  ioStreamManager.putNewInputStream(fd, fsDataInputStream, fileSystem);
  return fd;
+ } catch (InterruptedIOException e) {
+ Thread.interrupted(); // clear interrupted flag
+ LOG.error("Interrupted while open file " + path, e);
+ throw new UserException("Failed to open file " + path, e); // throw unified user exception
  } catch (IOException e) {
  LOG.error("errors while open path", e);
  throw new UserException("could not open file " + path, e);
@@ -1196,6 +1251,11 @@ public byte[] pread(TBrokerFD fd, long offset, long length) throws UserException
  long currentStreamOffset;
  try {
  currentStreamOffset = fsDataInputStream.getPos();
+ } catch (InterruptedIOException e) {
+ Thread.interrupted(); // clear interrupted flag
+ LOG.error("Interrupted while get file pos from output stream", e);
+ // throw unified user exception
+ throw new UserException("Failed to get file pos from output stream", e);
  } catch (IOException e) {
  LOG.error("errors while get file pos from output stream", e);
  throw new UserException("errors while get file pos from output stream");
@@ -1207,6 +1267,11 @@ public byte[] pread(TBrokerFD fd, long offset, long length) throws UserException
  + offset + " seek to it");
  try {
  fsDataInputStream.seek(offset);
+ } catch (InterruptedIOException e) {
+ Thread.interrupted(); // clear interrupted flag
+ LOG.error("Interrupted while seek file pos from output stream", e);
+ // throw unified user exception
+ throw new UserException("Failed to seek file pos from output stream", e);
  } catch (IOException e) {
  throw new UserException("current read offset " + currentStreamOffset + " is not equal to "
  + offset + ", and could not seek to it");
@@ -1234,6 +1299,10 @@ public byte[] pread(TBrokerFD fd, long offset, long length) throws UserException
  System.arraycopy(buf, 0, smallerBuf, 0, readLength);
  return smallerBuf;
  }
+ } catch (InterruptedIOException e) {
+ Thread.interrupted(); // clear interrupted flag
+ LOG.error("Interrupted while read data from stream", e);
+ throw new UserException("Failed to read data from stream", e); // throw unified user exception
  } catch (IOException e) {
  LOG.error("errors while read data from stream", e);
  throw new UserException("errors while read data from stream", e);
@@ -1250,6 +1319,10 @@ public void closeReader(TBrokerFD fd) throws UserException {
  synchronized (fsDataInputStream) {
  try {
  fsDataInputStream.close();
+ } catch (InterruptedIOException e) {
+ Thread.interrupted(); // clear interrupted flag
+ LOG.error("Interrupted while close file input stream", e);
+ throw new UserException("Failed to close file input stream", e); // throw unified user exception
  } catch (IOException e) {
  LOG.error("errors while close file input stream", e);
  throw new UserException("errors while close file input stream", e);
@@ -1271,6 +1344,10 @@ public TBrokerFD openWriter(String path, Map<String, String> loadProperties) thr
  LOG.info("finish a open writer request. fd: " + fd);
  ioStreamManager.putNewOutputStream(fd, fsDataOutputStream, fileSystem);
  return fd;
+ } catch (InterruptedIOException e) {
+ Thread.interrupted(); // clear interrupted flag
+ LOG.error("Interrupted while open file " + path, e);
+ throw new UserException("Failed to open file " + path, e); // throw unified user exception
  } catch (IOException e) {
  LOG.error("errors while open path", e);
  throw new UserException("could not open file " + path, e);
@@ -1287,6 +1364,11 @@ public void pwrite(TBrokerFD fd, long offset, byte[] data) throws UserException
  }
  try {
  fsDataOutputStream.write(data);
+ } catch (InterruptedIOException e) {
+ Thread.interrupted(); // clear interrupted flag
+ LOG.error("Interrupted while write file " + fd + " to output stream", e);
+ // throw unified user exception
+ throw new UserException("Failed to write file " + fd + " to output stream", e);
  } catch (IOException e) {
  LOG.error("errors while write file " + fd + " to output stream", e);
  throw new UserException("errors while write data to output stream", e);
@@ -1300,6 +1382,11 @@ public void closeWriter(TBrokerFD fd) throws UserException {
  try {
  fsDataOutputStream.hsync();
  fsDataOutputStream.close();
+ } catch (InterruptedIOException e) {
+ Thread.interrupted(); // clear interrupted flag
+ LOG.error("Interrupted while close file " + fd + " output stream", e);
+ // throw unified user exception
+ throw new UserException("Failed to close file " + fd + " output stream", e);
  } catch (IOException e) {
  LOG.error("errors while close file " + fd + " output stream", e);
  throw new UserException("errors while close file output stream", e);