[BugFix] Avoid hdfs fs manager interrupting the thread on exception

Signed-off-by: xiangguangyxg <[email protected]>

fe/fe-core/src/main/java/com/starrocks/fs/hdfs/HdfsFsManager.java

@@ -49,6 +49,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;
@@ -707,6 +708,7 @@ private HdfsFs getFileSystemByCloudConfiguration(CloudConfiguration cloudConfigu
  tProperties.setCloud_configuration(tCloudConfiguration);
  }
  return fileSystem;
+
  } catch (Exception e) {
  LOG.error("errors while connect to " + path, e);
  throw new UserException(e);
@@ -1200,6 +1202,10 @@ public List<FileStatus> listFileMeta(String path, Map<String, String> properties
  } 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);
@@ -1246,6 +1252,10 @@ public List<TBrokerFileStatus> listPath(String path, boolean fileNameOnly, Map<S
  "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
  } catch (Exception e) {
  LOG.error("errors while get file status ", e);
  throw new UserException("Fail to get file status: " + e.getMessage(), e);
@@ -1259,6 +1269,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);
@@ -1288,6 +1302,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);
@@ -1300,6 +1319,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);
@@ -1317,6 +1340,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);
@@ -1329,6 +1356,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");
@@ -1340,6 +1372,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");
@@ -1367,6 +1404,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);
@@ -1383,6 +1424,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);
@@ -1404,6 +1449,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);
@@ -1420,6 +1469,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);
@@ -1433,6 +1487,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);