Repository: AutoMQ/automq (language: java)

Issue:
[Enhancement] Standardize and Enhance S3 API Testing for AutoMQ Compatibility
### Who is this for and what problem do they have today?
Currently, the test logic for the S3 API in `GenerateS3UrlCmd` and `DefaultS3Operator` is different and not comprehensive. We aim to standardize and enhance the testing for the S3 API, including adding verification for the returned results.

### Why is solving this problem impactful?
We hope that the object storage which passes the test using s3-test.sh can be considered compatible with AutoMQ. Therefore, it is necessary to improve and standardize the testing logic.

### Additional notes
* `com.automq.stream.utils.S3Utils#checkS3Access`
* `com.automq.stream.s3.operator.DefaultS3Operator#checkAvailable`



Discussion / hints:

Fix patch:
diff --git a/core/src/main/scala/kafka/admin/S3TestCommand.scala b/core/src/main/scala/kafka/admin/S3TestCommand.scala
index 0a648d7d88..5e19f4775d 100644
--- a/core/src/main/scala/kafka/admin/S3TestCommand.scala
+++ b/core/src/main/scala/kafka/admin/S3TestCommand.scala
@@ -11,7 +11,7 @@
 
 package kafka.admin
 
-import com.automq.stream.utils.S3Utils
+import com.automq.stream.utils.PingS3Helper
 import joptsimple.OptionSpec
 import kafka.utils.{Exit, Logging}
 import org.apache.kafka.server.util.{CommandDefaultOptions, CommandLineUtils}
@@ -48,16 +48,16 @@ object S3TestCommand extends Logging {
         val forcePathStyle = opts.has(opts.forcePathStyleOpt)
         val tagging = opts.has(opts.tagging)
 
-        val context = S3Utils.S3Context.builder()
-            .setEndpoint(s3Endpoint)
-            .setCredentialsProviders(util.List.of(StaticCredentialsProvider.create(AwsBasicCredentials.create(s3AccessKey, s3SecretKey))))
-            .setBucketName(s3Bucket)
-            .setRegion(s3Region)
-            .setForcePathStyle(forcePathStyle)
-            .setTagging(tagging)
-            .build()
-
-        S3Utils.checkS3Access(context)
+        val pingS3Helper = PingS3Helper.builder()
+          .endpoint(s3Endpoint)
+          .bucket(s3Bucket)
+          .region(s3Region)
+          .credentialsProviders(util.List.of(StaticCredentialsProvider.create(AwsBasicCredentials.create(s3AccessKey, s3SecretKey))))
+          .isForcePathStyle(forcePathStyle)
+          .tagging(tagging)
+          .needPrintToConsole(true)
+          .build()
+        pingS3Helper.pingS3()
     }
 
 
diff --git a/core/src/main/scala/kafka/log/stream/s3/DefaultS3Client.java b/core/src/main/scala/kafka/log/stream/s3/DefaultS3Client.java
index ae7259590e..37c36d793b 100644
--- a/core/src/main/scala/kafka/log/stream/s3/DefaultS3Client.java
+++ b/core/src/main/scala/kafka/log/stream/s3/DefaultS3Client.java
@@ -31,10 +31,13 @@
 import com.automq.stream.s3.wal.BlockWALService;
 import com.automq.stream.s3.wal.WriteAheadLog;
 import com.automq.stream.utils.LogContext;
+import com.automq.stream.utils.PingS3Helper;
 import com.automq.stream.utils.threads.S3StreamThreadPoolMonitor;
+
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
+
 import kafka.log.stream.s3.failover.DefaultFailoverFactory;
 import kafka.log.stream.s3.metadata.StreamMetadataManager;
 import kafka.log.stream.s3.network.ControllerRequestSender;
@@ -85,6 +88,17 @@ public DefaultS3Client(BrokerServer brokerServer, KafkaConfig kafkaConfig) {
                 config.networkBaselineBandwidth(), config.refillPeriodMs());
         List<AwsCredentialsProvider> credentialsProviders = List.of(CredentialsProviderHolder.getAwsCredentialsProvider(), EnvVariableCredentialsProvider.get());
         boolean forcePathStyle = this.config.forcePathStyle();
+        // check s3 availability
+        PingS3Helper pingS3Helper = PingS3Helper.builder()
+                .endpoint(endpoint)
+                .bucket(bucket)
+                .region(region)
+                .credentialsProviders(credentialsProviders)
+                .isForcePathStyle(forcePathStyle)
+                .tagging(config.objectTagging())
+                .needPrintToConsole(false)
+                .build();
+        pingS3Helper.pingS3();
         S3Operator s3Operator = DefaultS3Operator.builder().endpoint(endpoint).region(region).bucket(bucket).credentialsProviders(credentialsProviders).tagging(config.objectTagging())
                 .inboundLimiter(networkInboundLimiter).outboundLimiter(networkOutboundLimiter).readWriteIsolate(true).forcePathStyle(forcePathStyle).build();
         S3Operator compactionS3Operator = DefaultS3Operator.builder().endpoint(endpoint).region(region).bucket(bucket).credentialsProviders(credentialsProviders).tagging(config.objectTagging())
diff --git a/s3stream/src/main/java/com/automq/stream/s3/operator/DefaultS3Operator.java b/s3stream/src/main/java/com/automq/stream/s3/operator/DefaultS3Operator.java
index 59952e28ac..f97b93d77f 100644
--- a/s3stream/src/main/java/com/automq/stream/s3/operator/DefaultS3Operator.java
+++ b/s3stream/src/main/java/com/automq/stream/s3/operator/DefaultS3Operator.java
@@ -21,7 +21,6 @@
 import com.automq.stream.s3.network.AsyncNetworkBandwidthLimiter;
 import com.automq.stream.s3.network.ThrottleStrategy;
 import com.automq.stream.utils.FutureUtil;
-import com.automq.stream.utils.S3Utils;
 import com.automq.stream.utils.ThreadUtils;
 import com.automq.stream.utils.Threads;
 import com.automq.stream.utils.Utils;
@@ -31,32 +30,7 @@
 import io.netty.handler.ssl.OpenSsl;
 import io.netty.util.HashedWheelTimer;
 import io.netty.util.Timeout;
-import java.net.URI;
-import java.nio.charset.StandardCharsets;
-import java.time.Duration;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Optional;
-import java.util.Set;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.Semaphore;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.function.Consumer;
-import java.util.stream.Collectors;
 import org.apache.commons.lang3.StringUtils;
-import org.apache.commons.lang3.exception.ExceptionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider;
@@ -66,7 +40,6 @@
 import software.amazon.awssdk.awscore.AwsRequestOverrideConfiguration;
 import software.amazon.awssdk.core.async.AsyncRequestBody;
 import software.amazon.awssdk.core.async.AsyncResponseTransformer;
-import software.amazon.awssdk.core.exception.SdkClientException;
 import software.amazon.awssdk.http.HttpStatusCode;
 import software.amazon.awssdk.http.async.SdkAsyncHttpClient;
 import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient;
@@ -83,7 +56,6 @@
 import software.amazon.awssdk.services.s3.model.DeleteObjectsResponse;
 import software.amazon.awssdk.services.s3.model.DeletedObject;
 import software.amazon.awssdk.services.s3.model.GetObjectRequest;
-import software.amazon.awssdk.services.s3.model.NoSuchBucketException;
 import software.amazon.awssdk.services.s3.model.ObjectIdentifier;
 import software.amazon.awssdk.services.s3.model.PutObjectRequest;
 import software.amazon.awssdk.services.s3.model.S3Error;
@@ -93,6 +65,30 @@
 import software.amazon.awssdk.services.s3.model.UploadPartRequest;
 import software.amazon.awssdk.services.s3.model.UploadPartResponse;
 
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+
 import static com.automq.stream.s3.metadata.ObjectUtils.tagging;
 import static com.automq.stream.utils.FutureUtil.cause;
 
@@ -133,17 +129,18 @@ public class DefaultS3Operator implements S3Operator {
         ThreadUtils.createThreadFactory("s3-timeout-detect", true), 1, TimeUnit.SECONDS, 100);
 
     private boolean deleteObjectsReturnSuccessKeys;
+    private boolean checkS3ApiModel = false;
 
     public DefaultS3Operator(String endpoint, String region, String bucket, boolean forcePathStyle,
         List<AwsCredentialsProvider> credentialsProviders, boolean tagging) {
-        this(endpoint, region, bucket, forcePathStyle, credentialsProviders, tagging, null, null, false);
+        this(endpoint, region, bucket, forcePathStyle, credentialsProviders, tagging, null, null, false, false);
     }
 
     public DefaultS3Operator(String endpoint, String region, String bucket, boolean forcePathStyle,
         List<AwsCredentialsProvider> credentialsProviders,
         boolean tagging,
         AsyncNetworkBandwidthLimiter networkInboundBandwidthLimiter,
-        AsyncNetworkBandwidthLimiter networkOutboundBandwidthLimiter, boolean readWriteIsolate) {
+        AsyncNetworkBandwidthLimiter networkOutboundBandwidthLimiter, boolean readWriteIsolate, boolean checkS3ApiModel) {
         this.currentIndex = INDEX.incrementAndGet();
         this.maxMergeReadSparsityRate = Utils.getMaxMergeReadSparsityRate();
         this.networkInboundBandwidthLimiter = networkInboundBandwidthLimiter;
@@ -155,17 +152,10 @@ public DefaultS3Operator(String endpoint, String region, String bucket, boolean
         this.inflightWriteLimiter = new Semaphore(maxS3Concurrency);
         this.inflightReadLimiter = readWriteIsolate ? new Semaphore(maxS3Concurrency) : inflightWriteLimiter;
         this.bucket = bucket;
+        this.checkS3ApiModel = checkS3ApiModel;
         scheduler.scheduleWithFixedDelay(this::tryMergeRead, 1, 1, TimeUnit.MILLISECONDS);
         checkConfig();
-        S3Utils.S3Context s3Context = S3Utils.S3Context.builder()
-            .setEndpoint(endpoint)
-            .setRegion(region)
-            .setBucketName(bucket)
-            .setForcePathStyle(forcePathStyle)
-            .setCredentialsProviders(credentialsProviders)
-            .build();
-        LOGGER.info("You are using s3Context: {}, max concurrency: {}", s3Context, maxS3Concurrency);
-        checkAvailable(s3Context);
+        setDeleteObjectsMode();
         S3StreamMetricsManager.registerInflightS3ReadQuotaSupplier(inflightReadLimiter::availablePermits, currentIndex);
         S3StreamMetricsManager.registerInflightS3WriteQuotaSupplier(inflightWriteLimiter::availablePermits, currentIndex);
     }
@@ -340,7 +330,7 @@ void mergedRangeRead0(String path, long start, long end, CompletableFuture<ByteB
         long size = end - start + 1;
         GetObjectRequest request = GetObjectRequest.builder().bucket(bucket).key(path).range(range(start, end)).build();
         Consumer<Throwable> failHandler = ex -> {
-            if (isUnrecoverable(ex)) {
+            if (isUnrecoverable(ex) || checkS3ApiModel) {
                 LOGGER.error("GetObject for object {} [{}, {}) fail", path, start, end, ex);
                 cf.completeExceptionally(ex);
             } else {
@@ -417,7 +407,7 @@ private void write0(String path, ByteBuf data, CompletableFuture<Void> cf) {
             cf.complete(null);
         }).exceptionally(ex -> {
             S3OperationStats.getInstance().putObjectStats(objectSize, false).record(timerUtil.elapsedAs(TimeUnit.NANOSECONDS));
-            if (isUnrecoverable(ex)) {
+            if (isUnrecoverable(ex) || checkS3ApiModel) {
                 LOGGER.error("PutObject for object {} fail", path, ex);
                 cf.completeExceptionally(ex);
                 data.release();
@@ -559,7 +549,7 @@ void createMultipartUpload0(String path, CompletableFuture<String> cf) {
             cf.complete(createMultipartUploadResponse.uploadId());
         }).exceptionally(ex -> {
             S3OperationStats.getInstance().createMultiPartUploadStats(false).record(timerUtil.elapsedAs(TimeUnit.NANOSECONDS));
-            if (isUnrecoverable(ex)) {
+            if (isUnrecoverable(ex) || checkS3ApiModel) {
                 LOGGER.error("CreateMultipartUpload for object {} fail", path, ex);
                 cf.completeExceptionally(ex);
             } else {
@@ -608,7 +598,7 @@ private void uploadPart0(String path, String uploadId, int partNumber, ByteBuf p
             cf.complete(completedPart);
         }).exceptionally(ex -> {
             S3OperationStats.getInstance().uploadPartStats(size, false).record(timerUtil.elapsedAs(TimeUnit.NANOSECONDS));
-            if (isUnrecoverable(ex)) {
+            if (isUnrecoverable(ex) || checkS3ApiModel) {
                 LOGGER.error("UploadPart for object {}-{} fail", path, partNumber, ex);
                 part.release();
                 cf.completeExceptionally(ex);
@@ -648,7 +638,7 @@ private void uploadPartCopy0(String sourcePath, String path, long start, long en
             cf.complete(completedPart);
         }).exceptionally(ex -> {
             S3OperationStats.getInstance().uploadPartCopyStats(false).record(timerUtil.elapsedAs(TimeUnit.NANOSECONDS));
-            if (isUnrecoverable(ex)) {
+            if (isUnrecoverable(ex) || checkS3ApiModel) {
                 LOGGER.warn("UploadPartCopy for object {}-{} [{}, {}] fail", path, partNumber, start, end, ex);
                 cf.completeExceptionally(ex);
             } else {
@@ -682,7 +672,7 @@ public void completeMultipartUpload0(String path, String uploadId, List<Complete
             cf.complete(null);
         }).exceptionally(ex -> {
             S3OperationStats.getInstance().completeMultiPartUploadStats(false).record(timerUtil.elapsedAs(TimeUnit.NANOSECONDS));
-            if (isUnrecoverable(ex)) {
+            if (isUnrecoverable(ex) || checkS3ApiModel) {
                 LOGGER.error("CompleteMultipartUpload for object {} fail", path, ex);
                 cf.completeExceptionally(ex);
             } else if (!checkPartNumbers(request.multipartUpload())) {
@@ -718,7 +708,7 @@ private void checkConfig() {
         }
     }
 
-    private CompletableFuture<Boolean> asyncCheckDeleteObjectsReturnSuccessDeleteKeys(S3Utils.S3Context s3Context) {
+    private CompletableFuture<Boolean> asyncCheckDeleteObjectsReturnSuccessDeleteKeys() {
         byte[] content = new Date().toString().getBytes(StandardCharsets.UTF_8);
         String path1 = String.format("check_available/deleteObjectsMode/%d", System.nanoTime());
         String path2 = String.format("check_available/deleteObjectsMode/%d", System.nanoTime() + 1);
@@ -767,57 +757,12 @@ static boolean checkIfDeleteObjectsWillReturnSuccessDeleteKeys(List<String> path
         throw exception;
     }
 
-    private void checkAvailable(S3Utils.S3Context s3Context) {
-        byte[] content = new Date().toString().getBytes(StandardCharsets.UTF_8);
-        String path = String.format("check_available/%d", System.nanoTime());
-        String multipartPath = String.format("check_available_multipart/%d", System.nanoTime());
+    private void setDeleteObjectsMode() {
         try {
-            // Check network and bucket
-//            readS3Client.getBucketAcl(b -> b.bucket(bucket)).get(3, TimeUnit.SECONDS);
-
-            // Simple write/read/delete
-            this.write(path, Unpooled.wrappedBuffer(content)).get(30, TimeUnit.SECONDS);
-            ByteBuf read = this.rangeRead(path, 0, content.length).get(30, TimeUnit.SECONDS);
-            read.release();
-            this.delete(path).get(30, TimeUnit.SECONDS);
-
-            // Multipart write/read/delete
-            Writer writer = this.writer(multipartPath);
-            writer.write(Unpooled.wrappedBuffer(content));
-            writer.close().get(30, TimeUnit.SECONDS);
-            read = this.rangeRead(multipartPath, 0, content.length).get(30, TimeUnit.SECONDS);
-            read.release();
-            this.delete(multipartPath).get(30, TimeUnit.SECONDS);
-
-            // Check if oss provider deleteObjects will return successDeleteKeys in deleted.
-            this.deleteObjectsReturnSuccessKeys = asyncCheckDeleteObjectsReturnSuccessDeleteKeys(s3Context).get(30, TimeUnit.SECONDS);
+            this.deleteObjectsReturnSuccessKeys = asyncCheckDeleteObjectsReturnSuccessDeleteKeys().get(30, TimeUnit.SECONDS);
         } catch (Throwable e) {
-            LOGGER.error("Failed to write/read/delete object on S3 ", e);
-            String exceptionMsg = String.format("Failed to write/read/delete object on S3. You are using s3Context: %s.", s3Context);
-
-            Throwable cause = e.getCause() != null ? e.getCause() : e;
-            if (cause instanceof SdkClientException) {
-                if (cause.getMessage().contains("UnknownHostException")) {
-                    Throwable rootCause = ExceptionUtils.getRootCause(cause);
-                    exceptionMsg += "\nUnable to resolve Host \"" + rootCause.getMessage() + "\". Please check your S3 endpoint.";
-                } else if (cause.getMessage().startsWith("Unable to execute HTTP request")) {
-                    exceptionMsg += "\nUnable to execute HTTP request. Please check your network connection and make sure you can access S3.";
-                }
-            }
-
-            if (e instanceof TimeoutException || cause instanceof TimeoutException) {
-                exceptionMsg += "\nConnection timeout. Please check your network connection and make sure you can access S3.";
-            }
-
-            if (cause instanceof NoSuchBucketException) {
-                exceptionMsg += "\nBucket \"" + bucket + "\" not found. Please check your bucket name.";
-            }
-
-            List<String> advices = s3Context.advices();
-            if (!advices.isEmpty()) {
-                exceptionMsg += "\nHere are some advices: \n" + String.join("\n", advices);
-            }
-            throw new RuntimeException(exceptionMsg, e);
+            LOGGER.error("Failed to check if the s3 `deleteObjects` api will return deleteKeys", e);
+            throw new RuntimeException(e);
         }
     }
 
@@ -1061,6 +1006,7 @@ public static class Builder {
         private boolean readWriteIsolate;
         private int maxReadConcurrency = 50;
         private int maxWriteConcurrency = 50;
+        private boolean checkS3ApiModel = false;
 
         public Builder endpoint(String endpoint) {
             this.endpoint = endpoint;
@@ -1107,9 +1053,14 @@ public Builder readWriteIsolate(boolean readWriteIsolate) {
             return this;
         }
 
+        public Builder checkS3ApiModel(boolean checkS3ApiModel) {
+            this.checkS3ApiModel = checkS3ApiModel;
+            return this;
+        }
+
         public DefaultS3Operator build() {
             return new DefaultS3Operator(endpoint, region, bucket, forcePathStyle, credentialsProviders, tagging,
-                inboundLimiter, outboundLimiter, readWriteIsolate);
+                inboundLimiter, outboundLimiter, readWriteIsolate, checkS3ApiModel);
         }
     }
 }
diff --git a/s3stream/src/main/java/com/automq/stream/utils/PingS3Helper.java b/s3stream/src/main/java/com/automq/stream/utils/PingS3Helper.java
new file mode 100644
index 0000000000..89d142dbac
--- /dev/null
+++ b/s3stream/src/main/java/com/automq/stream/utils/PingS3Helper.java
@@ -0,0 +1,439 @@
+/*
+ * Copyright 2024, AutoMQ CO.,LTD.
+ *
+ * Use of this software is governed by the Business Source License
+ * included in the file BSL.md
+ *
+ * As of the Change Date specified in that file, in accordance with
+ * the Business Source License, use of this software will be governed
+ * by the Apache License, Version 2.0
+ */
+
+package com.automq.stream.utils;
+
+
+import com.automq.stream.s3.operator.DefaultS3Operator;
+import com.automq.stream.s3.operator.S3Operator;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProviderChain;
+import software.amazon.awssdk.core.exception.SdkClientException;
+import software.amazon.awssdk.services.s3.model.CompletedPart;
+import software.amazon.awssdk.services.s3.model.NoSuchBucketException;
+import software.amazon.awssdk.services.s3.model.NoSuchKeyException;
+import software.amazon.awssdk.services.s3.model.S3Exception;
+
+import java.nio.charset.StandardCharsets;
+import java.util.AbstractMap;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import java.util.Locale;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.ExecutionException;
+
+public class PingS3Helper {
+    private static final Logger LOGGER = LoggerFactory.getLogger(PingS3Helper.class);
+    private S3Operator s3Operator;
+    private String endpoint;
+    private String region;
+    private String bucket;
+    private boolean isForcePathStyle;
+    private List<AwsCredentialsProvider> credentialsProviders;
+    private boolean tagging;
+    private final boolean needPrintToConsole;
+
+    private PingS3Helper(Builder builder) {
+        this.needPrintToConsole = builder.needPrintToConsole;
+    }
+
+    public static Builder builder() {
+        return new Builder();
+    }
+
+    public static class Builder {
+        private boolean needPrintToConsole;
+        private String endpoint;
+        private String region;
+        private String bucket;
+        private boolean isForcePathStyle;
+        private List<AwsCredentialsProvider> credentialsProviders;
+        private boolean tagging;
+
+        public Builder needPrintToConsole(boolean needPrintToConsole) {
+            this.needPrintToConsole = needPrintToConsole;
+            return this;
+        }
+
+        public Builder endpoint(String endpoint) {
+            this.endpoint = endpoint;
+            return this;
+        }
+
+        public Builder region(String region) {
+            this.region = region;
+            return this;
+        }
+
+        public Builder bucket(String bucket) {
+            this.bucket = bucket;
+            return this;
+        }
+
+        public Builder isForcePathStyle(boolean isForcePathStyle) {
+            this.isForcePathStyle = isForcePathStyle;
+            return this;
+        }
+
+        public Builder credentialsProviders(List<AwsCredentialsProvider> credentialsProviders) {
+            this.credentialsProviders = credentialsProviders;
+            return this;
+        }
+
+        public Builder tagging(boolean tagging) {
+            this.tagging = tagging;
+            return this;
+        }
+
+        public PingS3Helper build() {
+            PingS3Helper pingS3Helper = new PingS3Helper(this);
+            pingS3Helper.endpoint = this.endpoint;
+            pingS3Helper.region = this.region;
+            pingS3Helper.bucket = this.bucket;
+            pingS3Helper.isForcePathStyle = this.isForcePathStyle;
+            pingS3Helper.credentialsProviders = this.credentialsProviders;
+            pingS3Helper.tagging = this.tagging;
+            return pingS3Helper;
+        }
+    }
+
+    @Override
+    public String toString() {
+        return "s3 parameters{" +
+            "endpoint='" + endpoint + '\'' +
+            ", region='" + region + '\'' +
+            ", bucket='" + bucket + '\'' +
+            ", isForcePathStyle=" + isForcePathStyle +
+            ", credentialsProviders=" + credentialsProviders +
+            ", tagging=" + tagging +
+            '}';
+    }
+
+    public void pingS3() {
+        try {
+            s3Operator = DefaultS3Operator.builder()
+                .endpoint(endpoint)
+                .region(region)
+                .bucket(bucket)
+                .forcePathStyle(isForcePathStyle)
+                .credentialsProviders(credentialsProviders)
+                .tagging(tagging)
+                .checkS3ApiModel(true)
+                .build();
+        } catch (Exception e) {
+            handleException(e, "Delete objects");
+        }
+        try {
+            checkSimpleObjOperation();
+            checkMultipartUploadOperation();
+            checkUploadPartCopy().join();
+            checkDeleteObjs().join();
+        } finally {
+            s3Operator.close();
+        }
+    }
+
+    private void checkSimpleObjOperation() {
+        byte[] content = new Date().toString().getBytes(StandardCharsets.UTF_8);
+        String path = String.format("check_simple_obj_available/%d", System.nanoTime());
+
+        CompletableFuture<Void> future = checkWrite(path, Unpooled.wrappedBuffer(content))
+            .thenCompose(v -> checkRead(path, Unpooled.wrappedBuffer(content)))
+            .thenCompose(v -> checkDelete(path));
+
+        future.join();
+    }
+
+    private void checkMultipartUploadOperation() {
+        String path = String.format("check_multipart_obj_available/%d", System.nanoTime());
+        CompletableFuture<Void> future = checkCreateMultipartUpload(path)
+            .thenCompose(respUploadId -> checkUploadPart(path, respUploadId)
+                .thenApply(completedPart -> new AbstractMap.SimpleEntry<>(respUploadId, completedPart)))
+            .thenCompose(entry ->
+                checkCompleteMultipartUpload(path, entry.getKey(), entry.getValue())
+            ).thenRun(() -> s3Operator.delete(path))
+            .exceptionally(ex -> {
+                handleException(ex, "Delete object");
+                return null;
+            });
+
+        future.join();
+    }
+
+    private CompletableFuture<Void> checkWrite(String path, ByteBuf data) {
+        return s3Operator.write(path, data).thenRun(() -> {
+            LOGGER.info("Successfully write object to s3");
+            printOperationStatus("Write object", true);
+        }).exceptionally(ex -> handleException(ex, "Write object"));
+    }
+
+    private CompletableFuture<Void> checkRead(String path, ByteBuf data) {
+        return s3Operator.rangeRead(path, 0, data.readableBytes()).thenAccept(buf -> {
+
+            if (data.equals(buf)) {
+                LOGGER.info("Successfully rangeRead object");
+                printOperationStatus("RangeRead object", true);
+                buf.release();
+            } else {
+                String exceptionMsg = "Failed to rangeRead object. The read data is empty/wrong";
+                handleErrorResponse(exceptionMsg, "RangeRead object");
+            }
+        }).exceptionally(this::handleRangeReadException);
+    }
+
+    private CompletableFuture<Void> checkDelete(String path) {
+        return s3Operator.delete(path).thenRun(() -> {
+            LOGGER.info("Successfully delete object to s3");
+            printOperationStatus("Delete object", true);
+        }).exceptionally(ex -> handleException(ex, "Delete object"));
+    }
+
+    private CompletableFuture<String> checkCreateMultipartUpload(String path) {
+        return s3Operator.createMultipartUpload(path)
+            .thenApply(respUploadId -> {
+                if (respUploadId != null && !respUploadId.isEmpty()) {
+                    printOperationStatus("CreateMultipartUpload", true);
+                    LOGGER.info("Successfully createMultipartUpload, uploadId: {}", respUploadId);
+                } else {
+                    String exceptionMsg = "Failed to createMultipartUpload, uploadId is empty";
+                    handleErrorResponse(exceptionMsg, "CreateMultipartUpload");
+                }
+                return respUploadId;
+            })
+            .exceptionally(ex -> {
+                handleException(ex, "CreateMultipartUpload");
+                return null;
+            });
+    }
+
+    private CompletableFuture<CompletedPart> checkUploadPart(String path, String uploadId) {
+        byte[] content = new Date().toString().getBytes(StandardCharsets.UTF_8);
+        return s3Operator.uploadPart(path, uploadId, 1, Unpooled.wrappedBuffer(content))
+            .thenApply(completedPart -> {
+                if (completedPart.eTag() != null && !completedPart.eTag().isEmpty() && completedPart.partNumber() != null) {
+                    LOGGER.info("Successfully upload to s3, eTag: {}", completedPart.eTag());
+                    printOperationStatus("UploadPart", true);
+                } else {
+                    String exceptionMsg = String.format("Failed to uploadPart to s3, eTag: %s, partNumber: %s", completedPart.eTag(), completedPart.partNumber());
+                    handleErrorResponse(exceptionMsg, "UploadPart");
+                }
+                return completedPart;
+            })
+            .exceptionally(ex -> {
+                handleException(ex, "UploadPart");
+                return null;
+            });
+    }
+
+    private CompletionStage<Void> checkCompleteMultipartUpload(String path, String uploadId, CompletedPart completedPart) {
+        List<CompletedPart> parts = new ArrayList<>();
+        parts.add(completedPart);
+        return s3Operator.completeMultipartUpload(path, uploadId, parts).thenRun(() -> {
+            LOGGER.info("Successfully CompleteMultipartUpload");
+            printOperationStatus("CompleteMultipartUpload", true);
+        }).exceptionally(ex -> handleException(ex, "completeMultipartUpload"));
+    }
+
+    private CompletableFuture<Void> checkUploadPartCopy() {
+        String sourcePath = String.format("check_upload_part_copy_available/%d", System.nanoTime());
+        String path = String.format("check_upload_part_copy_available/%d", System.nanoTime());
+
+        byte[] content = new Date().toString().getBytes(StandardCharsets.UTF_8);
+        ByteBuf data = Unpooled.wrappedBuffer(content);
+
+        return s3Operator.write(sourcePath, data)
+            .thenCompose(aVoid -> s3Operator.createMultipartUpload(path))
+            .thenCompose(uploadId ->
+                s3Operator.uploadPartCopy(sourcePath, path, 0, data.readableBytes(), uploadId, 1)
+                    .thenCompose(uploadPartCopyResponse -> {
+                        if (uploadPartCopyResponse != null && uploadPartCopyResponse.eTag() != null && !uploadPartCopyResponse.eTag().isEmpty()) {
+                            LOGGER.info("Successfully uploadPartCopy");
+                            printOperationStatus("UploadPartCopy", true);
+                            CompletedPart completedPart = CompletedPart.builder().partNumber(1).eTag(uploadPartCopyResponse.eTag()).build();
+                            ArrayList<CompletedPart> completedParts = new ArrayList<>();
+                            completedParts.add(completedPart);
+                            return s3Operator.completeMultipartUpload(path, uploadId, completedParts);
+                        } else {
+                            handleErrorResponse("Failed to uploadPartCopy. The response was wrong", "UploadPartCopy");
+                        }
+                        return null;
+                    })
+                    .exceptionally(ex -> handleException(ex, "UploadPartCopy")))
+            .thenCompose(aVoid -> {
+                CompletableFuture<Void> deletePath = s3Operator.delete(path);
+                CompletableFuture<Void> deleteSourcePath = s3Operator.delete(sourcePath);
+                return CompletableFuture.allOf(deletePath, deleteSourcePath);
+            });
+    }
+
+    private CompletableFuture<Object> checkDeleteObjs() {
+        byte[] content = new Date().toString().getBytes(StandardCharsets.UTF_8);
+        String path1 = String.format("check_available/deleteObjects/%d", System.nanoTime());
+        String path2 = String.format("check_available/deleteObjects/%d", System.nanoTime() + 1);
+        List<String> paths = List.of(path1, path2);
+        CompletableFuture<Void> writeFuture = CompletableFuture.allOf(
+            s3Operator.write(path1, Unpooled.wrappedBuffer(content)),
+            s3Operator.write(path2, Unpooled.wrappedBuffer(content))
+        );
+
+        return writeFuture.thenCompose(v ->
+            s3Operator.delete(paths)
+                .thenCompose(response -> {
+                    if (response != null && response.size() == 2) {
+                        LOGGER.info("Successfully deleted objects");
+                        printOperationStatus("Delete objects", true);
+                        return CompletableFuture.completedFuture(null);
+                    } else {
+                        String exceptionMsg = "Failed to delete objects. The response was wrong";
+                        handleErrorResponse(exceptionMsg, "Delete objects");
+                    }
+                    return null;
+                }).exceptionally(ex -> handleException(ex, "deleteObjects"))
+        );
+    }
+
+    private void handleErrorResponse(String exceptionMsg, String operation) {
+        printOperationStatus(operation, false);
+        LOGGER.error(exceptionMsg);
+        throw new RuntimeException(exceptionMsg);
+    }
+
+    private Void handleException(Throwable ex, String operation) {
+        Throwable cause = ex.getCause() != null ? ex.getCause() : ex;
+        cause = ex.getCause();
+        String exceptionMsg = String.format("Failed to %s on S3. Here are your parameters about S3: %s.", operation, this);
+        if (cause instanceof SdkClientException) {
+            if (cause.getMessage().contains("UnknownHostException")) {
+                Throwable rootCause = ExceptionUtils.getRootCause(cause);
+                exceptionMsg += "\nUnable to resolve Host \"" + rootCause.getMessage() + "\". Please check your S3 endpoint.";
+            } else if (cause.getMessage().startsWith("Unable to execute HTTP request")) {
+                exceptionMsg += "\nUnable to execute HTTP request. Please check your network connection and make sure you can access S3.";
+            }
+        } else if (cause instanceof ExecutionException) {
+            Throwable realCause = cause.getCause();
+            if (realCause instanceof NoSuchBucketException) {
+                exceptionMsg += "\nBucket \"" + bucket + "\" not found. Please check your bucket name.";
+            }
+        } else if (cause instanceof S3Exception) {
+            if (cause instanceof NoSuchBucketException) {
+                exceptionMsg += "\nBucket \"" + bucket + "\" not found. Please check your bucket name.";
+            }
+        }
+        printOperationStatus(operation, false);
+        List<String> advices = advices();
+        if (!advices.isEmpty()) {
+            exceptionMsg += "\nHere are some advices: \n" + String.join("\n", advices);
+        }
+        throw new RuntimeException(exceptionMsg, ex);
+    }
+
+    private Void handleRangeReadException(Throwable ex) {
+        Throwable cause = ex.getCause() != null ? ex.getCause() : ex;
+        if (cause instanceof NoSuchKeyException) {
+            printOperationStatus("read object", false);
+            throw new RuntimeException("Failed to rangeRead object. No such key.", ex);
+        } else {
+            return handleException(ex, "read");
+        }
+    }
+
+    private void printOperationStatus(String operation, boolean isSuccess) {
+        if (needPrintToConsole) {
+            if (isSuccess) {
+                System.out.println("[ OK ] " + operation);
+            } else {
+                System.out.println("[ FAILED ] " + operation);
+            }
+        }
+    }
+
+    private List<String> advices() {
+        List<String> advises = new ArrayList<>();
+        checkBucketName(advises);
+        checkEndpoint(advises);
+        checkCredentialsProvider(advises);
+        checkRegion(advises);
+        checkForcePathStyle(advises);
+        checkTagging(advises);
+        return advises;
+    }
+
+    private void checkBucketName(List<String> advises) {
+        if (StringUtils.isBlank(bucket)) {
+            advises.add("bucketName is blank. Please supply a valid bucketName.");
+        }
+    }
+
+    private void checkEndpoint(List<String> advises) {
+        if (StringUtils.isBlank(endpoint)) {
+            advises.add("endpoint is blank. Please supply a valid endpoint.");
+        } else {
+            validateEndpoint(advises);
+        }
+    }
+
+    private void validateEndpoint(List<String> advises) {
+        if (endpoint.startsWith("https")) {
+            advises.add("You are using https endpoint. Please make sure your object storage service supports https.");
+        }
+        String[] splits = endpoint.split("//");
+        if (splits.length < 2) {
+            advises.add("endpoint is invalid. Please supply a valid endpoint.");
+        } else {
+            String[] dotSplits = splits[1].split("\\.");
+            if (dotSplits.length == 0 || StringUtils.isBlank(dotSplits[0])) {
+                advises.add("endpoint is invalid. Please supply a valid endpoint.");
+            } else if (!StringUtils.isBlank(bucket) && Objects.equals(bucket.toLowerCase(Locale.ENGLISH), dotSplits[0].toLowerCase(Locale.ENGLISH))) {
+                advises.add("bucket name should not be included in endpoint.");
+            }
+        }
+    }
+
+    private void checkCredentialsProvider(List<String> advises) {
+        if (credentialsProviders == null || credentialsProviders.isEmpty()) {
+            advises.add("no credentials provider is supplied. Please supply a credentials provider.");
+        } else {
+            try {
+                AwsCredentialsProviderChain chain = AwsCredentialsProviderChain.builder().credentialsProviders(credentialsProviders).build();
+                chain.resolveCredentials();
+            } catch (SdkClientException e) {
+                advises.add("all provided credentials providers are invalid. Please supply a valid credentials provider. Error msg: " + e.getMessage());
+            }
+        }
+    }
+
+    private void checkRegion(List<String> advises) {
+        if (StringUtils.isBlank(region)) {
+            advises.add("region is blank. Please supply a valid region.");
+        }
+    }
+
+    private void checkForcePathStyle(List<String> advises) {
+        if (!isForcePathStyle) {
+            advises.add("forcePathStyle is set as false. Please set it as true if you are using minio.");
+        }
+    }
+
+    private void checkTagging(List<String> advises) {
+        if (tagging) {
+            advises.add("currently, it's only supported in AWS S3. Please make sure your object storage service supports tagging.");
+        }
+    }
+}
diff --git a/s3stream/src/main/java/com/automq/stream/utils/S3Utils.java b/s3stream/src/main/java/com/automq/stream/utils/S3Utils.java
deleted file mode 100644
index 575162d45e..0000000000
--- a/s3stream/src/main/java/com/automq/stream/utils/S3Utils.java
+++ /dev/null
@@ -1,489 +0,0 @@
-/*
- * Copyright 2024, AutoMQ CO.,LTD.
- *
- * Use of this software is governed by the Business Source License
- * included in the file BSL.md
- *
- * As of the Change Date specified in that file, in accordance with
- * the Business Source License, use of this software will be governed
- * by the Apache License, Version 2.0
- */
-
-package com.automq.stream.utils;
-
-import com.automq.stream.s3.ByteBufAlloc;
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.CompositeByteBuf;
-import io.netty.buffer.Unpooled;
-import java.net.URI;
-import java.nio.ByteBuffer;
-import java.nio.charset.StandardCharsets;
-import java.time.Duration;
-import java.util.ArrayList;
-import java.util.Date;
-import java.util.List;
-import java.util.Locale;
-import java.util.Objects;
-import java.util.Random;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.commons.lang3.exception.ExceptionUtils;
-import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
-import software.amazon.awssdk.auth.credentials.AwsCredentialsProviderChain;
-import software.amazon.awssdk.core.async.AsyncRequestBody;
-import software.amazon.awssdk.core.async.AsyncResponseTransformer;
-import software.amazon.awssdk.core.exception.SdkClientException;
-import software.amazon.awssdk.regions.Region;
-import software.amazon.awssdk.services.s3.S3AsyncClient;
-import software.amazon.awssdk.services.s3.S3AsyncClientBuilder;
-import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest;
-import software.amazon.awssdk.services.s3.model.CompletedMultipartUpload;
-import software.amazon.awssdk.services.s3.model.CompletedPart;
-import software.amazon.awssdk.services.s3.model.CreateMultipartUploadRequest;
-import software.amazon.awssdk.services.s3.model.DeleteObjectRequest;
-import software.amazon.awssdk.services.s3.model.GetObjectRequest;
-import software.amazon.awssdk.services.s3.model.PutObjectRequest;
-import software.amazon.awssdk.services.s3.model.S3Exception;
-import software.amazon.awssdk.services.s3.model.Tag;
-import software.amazon.awssdk.services.s3.model.Tagging;
-import software.amazon.awssdk.services.s3.model.UploadPartRequest;
-import software.amazon.awssdk.services.s3.model.UploadPartResponse;
-
-public class S3Utils {
-
-    /**
-     * Check s3 access with context.
-     * This method is generally used to help users figure out problems in using S3.
-     *
-     * @param context s3 context.
-     */
-    public static void checkS3Access(S3Context context) {
-        try (ObjectOperationTask task = new ObjectOperationTask(context)) {
-            task.run();
-        } catch (Throwable e) {
-            System.out.println("ERROR: " + ExceptionUtils.getRootCause(e));
-            System.exit(1);
-        }
-
-        try (S3MultipartUploadTestTask task = new S3MultipartUploadTestTask(context)) {
-            task.run();
-        } catch (Throwable e) {
-            System.out.println("ERROR: " + ExceptionUtils.getRootCause(e));
-            System.exit(1);
-        }
-    }
-
-    private static String range(long start, long end) {
-        if (end == -1L) {
-            return "bytes=" + start + "-";
-        }
-        return "bytes=" + start + "-" + end;
-    }
-
-    private static S3AsyncClient newS3AsyncClient(String endpoint, String region, boolean forcePathStyle,
-        List<AwsCredentialsProvider> credentialsProviders) {
-        S3AsyncClientBuilder builder = S3AsyncClient.builder().region(Region.of(region));
-        if (StringUtils.isNotBlank(endpoint)) {
-            builder.endpointOverride(URI.create(endpoint));
-        }
-        builder.serviceConfiguration(c -> c.pathStyleAccessEnabled(forcePathStyle));
-        builder.credentialsProvider(AwsCredentialsProviderChain.builder().credentialsProviders(credentialsProviders).build());
-        builder.overrideConfiguration(b -> b.apiCallTimeout(Duration.ofMinutes(1))
-            .apiCallAttemptTimeout(Duration.ofSeconds(30)));
-        return builder.build();
-    }
-
-    private static abstract class S3CheckTask implements AutoCloseable {
-        protected final S3AsyncClient client;
-        protected final String bucketName;
-        protected final Tagging tagging;
-        private final String taskName;
-
-        public S3CheckTask(S3Context context, String taskName) {
-            this.client = newS3AsyncClient(context.endpoint, context.region, context.forcePathStyle, context.credentialsProviders);
-            this.bucketName = context.bucketName;
-            this.tagging = context.tagging ? Tagging.builder().tagSet(Tag.builder().key("task").value(taskName).build()).build() : null;
-            this.taskName = taskName;
-        }
-
-        protected static void showErrorInfo(Exception e) {
-            if (e.getCause() instanceof S3Exception) {
-                S3Exception se = (S3Exception) e.getCause();
-                // Do not use system.err because automq admin tool suppress system.err
-                System.out.println("get S3 exception: ");
-                se.printStackTrace(System.out);
-            } else {
-                System.out.println("get other exception: ");
-                e.printStackTrace(System.out);
-            }
-        }
-
-        protected void run() {
-        }
-
-        public String getTaskName() {
-            return taskName;
-        }
-
-        @Override
-        public void close() {
-            if (this.client != null) {
-                client.close();
-            }
-        }
-    }
-
-    // This task is used to test s3 multipart upload
-    private static class S3MultipartUploadTestTask extends ObjectOperationTask {
-        private Random random = new Random();
-
-        public S3MultipartUploadTestTask(S3Context context) {
-            super(context, S3MultipartUploadTestTask.class.getSimpleName());
-        }
-
-        @Override
-        public void run() {
-            ByteBuf byteBuf = null;
-            try {
-                // Simple write/read/delete
-                String uploadId = createMultipartUpload(client, bucketName, path).get();
-                List<CompletedPart> parts = new ArrayList<>();
-                int data1Size = 1024 * 1024 * 5;
-                int data2Size = 1024;
-                int totalSize = data1Size + data2Size;
-
-                byte[] randomBytes = new byte[data1Size];
-                random.nextBytes(randomBytes);
-                ByteBuf data1 = Unpooled.wrappedBuffer(randomBytes);
-                writePart(uploadId, path, bucketName, data1, 1).thenAccept(parts::add).get();
-
-                byte[] randomBytes2 = new byte[data2Size];
-                random.nextBytes(randomBytes2);
-                ByteBuf data2 = Unpooled.wrappedBuffer(randomBytes2);
-                writePart(uploadId, path, bucketName, data2, 2).thenAccept(parts::add).get();
-
-                System.out.println("[ OK ] Write S3 object");
-
-                completeMultipartUpload(client, path, bucketName, uploadId, parts).get();
-                System.out.println("[ OK ] Upload s3 multipart object");
-
-                CompletableFuture<ByteBuf> readCf = new CompletableFuture<>();
-                readRange(client, path, readCf, bucketName, 0, -1);
-                byteBuf = readCf.get();
-                if (byteBuf == null) {
-                    System.out.println("[ FAILED ] Read s3 multipart object");
-                    throw new RuntimeException("read multipart object " + path + " fail. got null");
-                } else if (byteBuf.readableBytes() != totalSize) {
-                    System.out.println("[ FAILED ] Read s3 multipart object");
-                    throw new RuntimeException("read multipart object " + path + " fail. expected size " + totalSize + ", actual size " + byteBuf.readableBytes());
-                }
-                System.out.println("[ OK ] Read s3 multipart object");
-            } catch (ExecutionException | InterruptedException e) {
-                showErrorInfo(e);
-                throw new RuntimeException(e);
-            } finally {
-                if (byteBuf != null) {
-                    byteBuf.release();
-                }
-            }
-        }
-
-        private CompletableFuture<String> createMultipartUpload(S3AsyncClient writeS3Client, String bucketName,
-            String path) {
-            CompletableFuture<String> cf = new CompletableFuture<>();
-            CreateMultipartUploadRequest.Builder builder = CreateMultipartUploadRequest.builder().bucket(bucketName).key(path);
-            if (null != tagging) {
-                builder.tagging(tagging);
-            }
-            CreateMultipartUploadRequest request = builder.build();
-            writeS3Client.createMultipartUpload(request).thenAccept(createMultipartUploadResponse -> {
-                cf.complete(createMultipartUploadResponse.uploadId());
-            }).exceptionally(ex -> {
-                System.out.println("[ FAILED ] Upload s3 multipart object");
-                cf.completeExceptionally(ex);
-                return null;
-            });
-            return cf;
-        }
-
-        public CompletableFuture<Void> completeMultipartUpload(S3AsyncClient writeS3Client, String path, String bucket,
-            String uploadId, List<CompletedPart> parts) {
-            CompletableFuture<Void> cf = new CompletableFuture<>();
-            CompletedMultipartUpload multipartUpload = CompletedMultipartUpload.builder().parts(parts).build();
-            CompleteMultipartUploadRequest request = CompleteMultipartUploadRequest.builder().bucket(bucket).key(path).uploadId(uploadId).multipartUpload(multipartUpload).build();
-
-            writeS3Client.completeMultipartUpload(request).thenAccept(completeMultipartUploadResponse -> {
-                cf.complete(null);
-            }).exceptionally(ex -> {
-                System.out.println("[ FAILED ] Upload s3 multipart object, upload id is " + uploadId);
-                cf.completeExceptionally(ex);
-                return null;
-            });
-            return cf;
-        }
-
-        private CompletableFuture<CompletedPart> writePart(String uploadId, String path, String bucket, ByteBuf data,
-            int partNum) {
-            CompletableFuture<CompletedPart> cf = new CompletableFuture<>();
-            uploadPart(client, cf, path, uploadId, partNum, bucket, data);
-            return cf;
-        }
-
-        private void uploadPart(S3AsyncClient writeS3Client, CompletableFuture<CompletedPart> cf, String path,
-            String uploadId, int partNumber, String bucket, ByteBuf part) {
-            AsyncRequestBody body = AsyncRequestBody.fromByteBuffersUnsafe(part.nioBuffers());
-            UploadPartRequest request = UploadPartRequest.builder().bucket(bucket).key(path).uploadId(uploadId)
-                .partNumber(partNumber).build();
-            CompletableFuture<UploadPartResponse> uploadPartCf = writeS3Client.uploadPart(request, body);
-            uploadPartCf.thenAccept(uploadPartResponse -> {
-                CompletedPart completedPart = CompletedPart.builder().partNumber(partNumber).eTag(uploadPartResponse.eTag()).build();
-                cf.complete(completedPart);
-            }).exceptionally(ex -> {
-                cf.completeExceptionally(ex);
-                return null;
-            });
-            cf.whenComplete((rst, ex) -> part.release());
-        }
-    }
-
-    private static class ObjectOperationTask extends S3CheckTask {
-        protected final String path;
-
-        public ObjectOperationTask(S3Context context) {
-            this(context, ObjectOperationTask.class.getSimpleName());
-        }
-
-        protected ObjectOperationTask(S3Context context, String taskName) {
-            super(context, taskName);
-            this.path = String.format("%d/%s", System.nanoTime(), getTaskName());
-        }
-
-        @Override
-        public void run() {
-            byte[] content = new Date().toString().getBytes(StandardCharsets.UTF_8);
-            ByteBuf byteBuf = null;
-            try {
-                // Simple write/read/delete
-                CompletableFuture<Void> writeCf = new CompletableFuture<>();
-                writeObject(client, path, ByteBuffer.wrap(content), writeCf, bucketName);
-                writeCf.get();
-                System.out.println("[ OK ] Write s3 object");
-
-                CompletableFuture<ByteBuf> readCf = new CompletableFuture<>();
-                readRange(client, path, readCf, bucketName, 0, -1);
-                byteBuf = readCf.get();
-                if (byteBuf == null) {
-                    System.out.println("[ Failed ] Read s3 object");
-                    throw new RuntimeException("read object " + path + " fail. got null");
-                } else if (byteBuf.readableBytes() != content.length) {
-                    System.out.println("[ Failed ] Read s3 object");
-                    throw new RuntimeException("read object " + path + " fail. expected size " + content.length + ", actual size " + byteBuf.readableBytes());
-                }
-                byte[] readContent = new byte[byteBuf.readableBytes()];
-                byteBuf.readBytes(readContent);
-                if (!StringUtils.equals(new String(readContent, StandardCharsets.UTF_8), new String(content, StandardCharsets.UTF_8))) {
-                    System.out.println("[ Failed ] Read s3 object");
-                    throw new RuntimeException("read object " + path + " fail. expected content " + new String(content, StandardCharsets.UTF_8) + ", actual content " + new String(readContent, StandardCharsets.UTF_8));
-                }
-                System.out.println("[ OK ] Read s3 object");
-            } catch (ExecutionException | InterruptedException e) {
-                showErrorInfo(e);
-                throw new RuntimeException(e);
-            } finally {
-                if (byteBuf != null) {
-                    byteBuf.release();
-                }
-            }
-        }
-
-        private void writeObject(S3AsyncClient writeS3Client, String path, ByteBuffer data, CompletableFuture<Void> cf,
-            String bucket) {
-            PutObjectRequest.Builder builder = PutObjectRequest.builder().bucket(bucket).key(path);
-            if (null != tagging) {
-                builder.tagging(tagging);
-            }
-            PutObjectRequest request = builder.build();
-            AsyncRequestBody body = AsyncRequestBody.fromByteBuffersUnsafe(data);
-            writeS3Client.putObject(request, body).thenAccept(putObjectResponse -> {
-                cf.complete(null);
-            }).exceptionally(ex -> {
-                System.out.printf("[ Failed ] Write s3 object. PutObject for object %s fail with msg %s %n", path, ex.getMessage());
-                cf.completeExceptionally(ex);
-                return null;
-            });
-        }
-
-        protected void readRange(S3AsyncClient readS3Client, String path, CompletableFuture<ByteBuf> cf, String bucket,
-            long start, long end) {
-            GetObjectRequest request = GetObjectRequest.builder().bucket(bucket).key(path).range(range(start, end)).build();
-            readS3Client.getObject(request, AsyncResponseTransformer.toPublisher())
-                .thenAccept(responsePublisher -> {
-                    CompositeByteBuf buf = ByteBufAlloc.compositeByteBuffer();
-                    responsePublisher.subscribe(bytes -> {
-                        // the aws client will copy DefaultHttpContent to heap ByteBuffer
-                        buf.addComponent(true, Unpooled.wrappedBuffer(bytes));
-                    }).thenAccept(v -> {
-                        cf.complete(buf);
-                    });
-                }).exceptionally(ex -> {
-                    cf.completeExceptionally(ex);
-                    return null;
-                });
-        }
-
-        protected void deleteObject(S3AsyncClient deleteS3Client, String path, CompletableFuture<Void> cf,
-            String bucket) {
-            DeleteObjectRequest request = DeleteObjectRequest.builder().bucket(bucket).key(path).build();
-            deleteS3Client.deleteObject(request).thenAccept(deleteObjectResponse -> {
-                cf.complete(null);
-            }).exceptionally(ex -> {
-                System.out.printf("[ FAILED ] Delete s3 object. Delete object %s fail with msg %s %n", path, ex.getMessage());
-                cf.completeExceptionally(ex);
-                return null;
-            });
-        }
-
-        @Override
-        public void close() {
-            try {
-                CompletableFuture<Void> deleteCf = new CompletableFuture<>();
-                deleteObject(client, path, deleteCf, bucketName);
-                deleteCf.get();
-            } catch (InterruptedException | ExecutionException e) {
-
-                System.out.println("[ FAILED ] Delete s3 object. NOTICE: please delete object " + path + " manually!!!");
-                showErrorInfo(e);
-                throw new RuntimeException(e);
-            } finally {
-                super.close();
-            }
-            System.out.println("[ OK ] Delete s3 object");
-
-        }
-    }
-
-    public static class S3Context {
-        private final String endpoint;
-        private final List<AwsCredentialsProvider> credentialsProviders;
-        private final String bucketName;
-        private final String region;
-        private final boolean forcePathStyle;
-        private final boolean tagging;
-
-        public S3Context(String endpoint, List<AwsCredentialsProvider> credentialsProviders, String bucketName,
-            String region,
-            boolean forcePathStyle,
-            boolean tagging) {
-            this.endpoint = endpoint;
-            this.credentialsProviders = credentialsProviders;
-            this.bucketName = bucketName;
-            this.region = region;
-            this.forcePathStyle = forcePathStyle;
-            this.tagging = tagging;
-        }
-
-        public static Builder builder() {
-            return new Builder();
-        }
-
-        public List<String> advices() {
-            List<String> advises = new ArrayList<>();
-            if (StringUtils.isBlank(bucketName)) {
-                advises.add("bucketName is blank. Please supply a valid bucketName.");
-            }
-            if (StringUtils.isBlank(endpoint)) {
-                advises.add("endpoint is blank. Please supply a valid endpoint.");
-            } else {
-                if (endpoint.startsWith("https")) {
-                    advises.add("You are using https endpoint. Please make sure your object storage service supports https.");
-                }
-                String[] splits = endpoint.split("//");
-                if (splits.length < 2) {
-                    advises.add("endpoint is invalid. Please supply a valid endpoint.");
-                } else {
-                    String[] dotSplits = splits[1].split("\\.");
-                    if (dotSplits.length == 0 || StringUtils.isBlank(dotSplits[0])) {
-                        advises.add("endpoint is invalid. Please supply a valid endpoint.");
-                    } else if (!StringUtils.isBlank(bucketName) && Objects.equals(bucketName.toLowerCase(Locale.ENGLISH), dotSplits[0].toLowerCase(Locale.ENGLISH))) {
-                        advises.add("bucket name should not be included in endpoint.");
-                    }
-                }
-            }
-            if (credentialsProviders == null || credentialsProviders.isEmpty()) {
-                advises.add("no credentials provider is supplied. Please supply a credentials provider.");
-            }
-            try (AwsCredentialsProviderChain chain = AwsCredentialsProviderChain.builder().credentialsProviders(credentialsProviders).build()) {
-                chain.resolveCredentials();
-            } catch (SdkClientException e) {
-                advises.add("all provided credentials providers are invalid. Please supply a valid credentials provider. Error msg: " + e.getMessage());
-            }
-            if (StringUtils.isBlank(region)) {
-                advises.add("region is blank. Please supply a valid region.");
-            }
-            if (!forcePathStyle) {
-                advises.add("forcePathStyle is set as false. Please set it as true if you are using minio.");
-            }
-            if (tagging) {
-                advises.add("currently, it's only supported in AWS S3. Please make sure your object storage service supports tagging.");
-            }
-            return advises;
-        }
-
-        @Override
-        public String toString() {
-            return "S3CheckContext{" +
-                "endpoint='" + endpoint + '\'' +
-                ", credentialsProviders=" + credentialsProviders +
-                ", bucketName='" + bucketName + '\'' +
-                ", region='" + region + '\'' +
-                ", forcePathStyle=" + forcePathStyle +
-                ", tagging=" + tagging +
-                '}';
-        }
-
-        public static class Builder {
-            private String endpoint;
-            private List<AwsCredentialsProvider> credentialsProviders;
-            private String bucketName;
-            private String region;
-            private boolean forcePathStyle;
-            private boolean tagging;
-
-            public Builder setEndpoint(String endpoint) {
-                this.endpoint = endpoint;
-                return this;
-            }
-
-            public Builder setCredentialsProviders(List<AwsCredentialsProvider> credentialsProviders) {
-                this.credentialsProviders = credentialsProviders;
-                return this;
-            }
-
-            public Builder setBucketName(String bucketName) {
-                this.bucketName = bucketName;
-                return this;
-            }
-
-            public Builder setRegion(String region) {
-                this.region = region;
-                return this;
-            }
-
-            public Builder setForcePathStyle(boolean forcePathStyle) {
-                this.forcePathStyle = forcePathStyle;
-                return this;
-            }
-
-            public Builder setTagging(boolean tagging) {
-                this.tagging = tagging;
-                return this;
-            }
-
-            public S3Context build() {
-                return new S3Context(endpoint, credentialsProviders, bucketName, region, forcePathStyle, tagging);
-            }
-        }
-
-    }
-}
diff --git a/tools/src/main/java/org/apache/kafka/tools/automq/GenerateS3UrlCmd.java b/tools/src/main/java/org/apache/kafka/tools/automq/GenerateS3UrlCmd.java
index f961e2fe41..2c177a1baf 100644
--- a/tools/src/main/java/org/apache/kafka/tools/automq/GenerateS3UrlCmd.java
+++ b/tools/src/main/java/org/apache/kafka/tools/automq/GenerateS3UrlCmd.java
@@ -17,7 +17,8 @@
 package org.apache.kafka.tools.automq;
 
 import com.automq.s3shell.sdk.model.EndpointProtocol;
-import com.automq.stream.utils.S3Utils;
+import com.automq.stream.utils.PingS3Helper;
+
 import java.util.List;
 import net.sourceforge.argparse4j.inf.ArgumentParser;
 import net.sourceforge.argparse4j.inf.Namespace;
@@ -150,14 +151,16 @@ public String run() {
         System.out.println();
 
         // precheck
-        var context = S3Utils.S3Context.builder()
-            .setEndpoint(parameter.endpointProtocol.getName() + "://" + parameter.s3Endpoint)
-            .setCredentialsProviders(List.of(() -> AwsBasicCredentials.create(parameter.s3AccessKey, parameter.s3SecretKey)))
-            .setBucketName(parameter.s3DataBucket)
-            .setRegion(parameter.s3Region)
-            .setForcePathStyle(parameter.s3PathStyle)
-            .build();
-        S3Utils.checkS3Access(context);
+        PingS3Helper pingS3Helper = PingS3Helper.builder()
+                .endpoint(parameter.endpointProtocol.getName() + "://" + parameter.s3Endpoint)
+                .bucket(parameter.s3DataBucket)
+                .region(parameter.s3Region)
+                .credentialsProviders(List.of(() -> AwsBasicCredentials.create(parameter.s3AccessKey, parameter.s3SecretKey)))
+                .isForcePathStyle(parameter.s3PathStyle)
+                .tagging(false)
+                .needPrintToConsole(true)
+                .build();
+        pingS3Helper.pingS3();
 
         String s3Url = buildS3Url();
         System.out.println("############  String of s3url ################");


Test patch: