oss avoid listBuckets permission in bucket check (#14414)

---------

Co-authored-by: kuantian.zhang <kuantian.zhang@inceptio.ai>

dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/log/remote/OssRemoteLogHandler.java

@@ -31,7 +31,6 @@ import java.io.IOException;
 import lombok.extern.slf4j.Slf4j;
 
 import com.aliyun.oss.OSS;
-import com.aliyun.oss.model.Bucket;
 import com.aliyun.oss.model.GetObjectRequest;
 import com.aliyun.oss.model.PutObjectRequest;
 
@@ -104,17 +103,13 @@ public class OssRemoteLogHandler implements RemoteLogHandler, Closeable {
             throw new IllegalArgumentException(Constants.REMOTE_LOGGING_OSS_BUCKET_NAME + " is empty");
         }
 
-        Bucket existsBucket = ossClient.listBuckets()
+        boolean existsBucket = ossClient.doesBucketExist(bucketName);
-                .stream()
+        if (!existsBucket) {
-                .filter(
+            throw new IllegalArgumentException(
-                        bucket -> bucket.getName().equals(bucketName))
+                    "bucketName: " + bucketName + " is not exists, you need to create them by yourself");
-                .findFirst()
+        }
-                .orElseThrow(() -> {
-                    return new IllegalArgumentException(
-                            "bucketName: " + bucketName + " does not exist, you need to create them by yourself");
-                });
 
-        log.info("bucketName: {} has been found", existsBucket.getName());
+        log.info("bucketName: {} has been found", bucketName);
     }
 
     private String readOssAccessKeyId() {

dolphinscheduler-storage-plugin/dolphinscheduler-storage-oss/src/main/java/org/apache/dolphinscheduler/plugin/storage/oss/OssStorageOperator.java

@@ -59,7 +59,6 @@ import lombok.extern.slf4j.Slf4j;
 import com.aliyun.oss.OSS;
 import com.aliyun.oss.OSSException;
 import com.aliyun.oss.ServiceException;
-import com.aliyun.oss.model.Bucket;
 import com.aliyun.oss.model.DeleteObjectsRequest;
 import com.aliyun.oss.model.ListObjectsV2Request;
 import com.aliyun.oss.model.ListObjectsV2Result;
@@ -509,17 +508,13 @@ public class OssStorageOperator implements Closeable, StorageOperate {
             throw new IllegalArgumentException("resource.alibaba.cloud.oss.bucket.name is empty");
         }
 
-        Bucket existsBucket = ossClient.listBuckets()
+        boolean existsBucket = ossClient.doesBucketExist(bucketName);
-                .stream()
+        if (!existsBucket) {
-                .filter(
+            throw new IllegalArgumentException(
-                        bucket -> bucket.getName().equals(bucketName))
+                    "bucketName: " + bucketName + " is not exists, you need to create them by yourself");
-                .findFirst()
+        }
-                .orElseThrow(() -> {
-                    return new IllegalArgumentException(
-                            "bucketName: " + bucketName + " does not exist, you need to create them by yourself");
-                });
 
-        log.info("bucketName: {} has been found, the current regionName is {}", existsBucket.getName(), region);
+        log.info("bucketName: {} has been found, the current regionName is {}", bucketName, region);
     }
 
     protected void deleteDir(String directoryName) {