修改minio连接

seatunnel-datasource/seatunnel-datasource-plugins/datasource-s3/pom.xml

@@ -86,6 +86,13 @@
 
             </exclusions>
         </dependency>
+
+        <dependency>
+            <groupId>io.quarkiverse.minio</groupId>
+            <artifactId>minio-client</artifactId>
+            <version>0.2.0</version>
+        </dependency>
+
         <!--        <dependency>-->
         <!--            <groupId>org.slf4j</groupId>-->
         <!--            <artifactId>slf4j-reload4j</artifactId>-->

seatunnel-datasource/seatunnel-datasource-plugins/datasource-s3/src/main/java/org/apache/seatunnel/datasource/plugin/s3/S3ClientService.java

+package org.apache.seatunnel.datasource.plugin.s3;
+
+import com.amazonaws.services.dynamodbv2.xspec.S;
+import io.minio.MinioClient;
+import io.minio.errors.MinioException;
+
+public class S3ClientService {
+    private String ENDPOINT;
+    private String PROVIDER;
+    private String USERNAME;
+    private String PASSWORD;
+    private String BUCKET;
+
+    private Integer PORT;
+    private final String clientId = "Client" + (int) (Math.random() * 100000000);
+    private MinioClient minioClient;
+
+    private String TOPIC;
+
+
+    public S3ClientService(
+            String endpoint,
+            String provider,
+            String username,
+            String password,
+            String bucket,
+            Integer port) throws MinioException {
+        this.ENDPOINT = endpoint;
+        this.PROVIDER = provider;
+        this.USERNAME = username;
+        this.PASSWORD = password;
+        this.BUCKET = bucket;
+        this.PORT = port;
+        setMinioClient(endpoint, provider, username, password, bucket, port);
+    }
+
+    public MinioClient getMinioClient() {
+        return minioClient;
+    }
+
+
+
+    public void setMinioClient(
+            String endpoint, String provider,
+            String username, String password, String bucket, Integer port)
+            throws MinioException {
+        minioClient = new MinioClient.Builder()
+                .endpoint(endpoint, port, true)
+                .credentials(username, password)
+                .build();
+
+    }
+
+
+
+//
+//    /** 关闭MQTT连接 */
+//    public void close() throws MqttException {
+//        mqttClient.close();
+//        mqttClient.disconnect();
+//    }
+}

seatunnel-datasource/seatunnel-datasource-plugins/datasource-s3/src/main/java/org/apache/seatunnel/datasource/plugin/s3/S3DatasourceChannel.java

@@ -17,7 +17,10 @@
 
 package org.apache.seatunnel.datasource.plugin.s3;
 
+import io.minio.MinioClient;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.common.utils.SeaTunnelException;
 import org.apache.seatunnel.datasource.plugin.api.DataSourceChannel;
 import org.apache.seatunnel.datasource.plugin.api.DataSourcePluginException;
 import org.apache.seatunnel.datasource.plugin.api.model.TableField;
@@ -69,14 +72,24 @@ public class S3DatasourceChannel implements DataSourceChannel {
     @Override
     public boolean checkDataSourceConnectivity(
             @NonNull String pluginName, @NonNull Map<String, String> requestParams) {
-        Configuration conf = HadoopS3AConfiguration.getConfiguration(requestParams);
-        try (FileSystem fs = FileSystem.get(conf)) {
-            fs.listStatus(new Path("/"));
+        if (requestParams.isEmpty()) {
+            throw new SeaTunnelException("requestParmas 为空！");
+        }
+        try {
+            S3ClientService s3Client = createS3Client(requestParams);
             return true;
-        } catch (IOException e) {
+        } catch (Exception ex) {
             throw new DataSourcePluginException(
-                    String.format("check s3 connectivity failed, config is: %s", requestParams), e);
+                    "check s3 connectivity failed, " + ex.getMessage(), ex);
         }
+//        try (FileSystem fs = FileSystem.get(conf)) {
+////            fs.listStatus(new Path("/"));
+//
+//            return true;
+//        } catch (IOException e) {
+//            throw new DataSourcePluginException(
+//                    String.format("check s3 connectivity failed, config is: %s", requestParams), e);
+//        }
     }
 
     @Override
@@ -98,4 +111,20 @@ public class S3DatasourceChannel implements DataSourceChannel {
         throw new UnsupportedOperationException(
                 "getTableFields is not supported for S3 datasource");
     }
+
+    private S3ClientService createS3Client(Map<String, String> requestParams) {
+        String endpoint = requestParams.get("fs.s3a.endpoint").split(":")[0] + "";
+        Integer port = Integer.valueOf(requestParams.get("fs.s3a.endpoint").split(":")[1]);
+        String provider = requestParams.get("fs.s3a.aws.credentials.provider") + "";
+        String username = requestParams.get("access_key") + "";
+        String password = requestParams.get("secret_key") + "";
+        String bucket = requestParams.get("bucket") + "";
+        try {
+            S3ClientService s3ClientService = new S3ClientService(endpoint, provider, username, password, bucket, port);
+
+            return s3ClientService;
+        } catch (Exception e) {
+            throw new SeaTunnelException("创建Mqtt客户端错误！");
+        }
+    }
 }