@Override
    public void open(Configuration configuration) throws Exception {
        converter = FormatUtils.parseRowDataToJsonConverter(physicalRowDataType.getLogicalType());
        fieldGetters = FormatUtils.parseFieldGetters(physicalRowDataType.getLogicalType());
        AmazonS3 s3Client;
        if (s3Options.getAccessKeyId() != null && s3Options.getSecretKeyId() != null) {
            BasicAWSCredentials awsCreds =
                    new BasicAWSCredentials(s3Options.getAccessKeyId(), s3Options.getSecretKeyId());
            s3Client = AmazonS3ClientBuilder.standard().withCredentials(new AWSStaticCredentialsProvider(awsCreds))
                    .withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(
                            s3Options.getEndpoint(),
                            s3Options.getRegion()))
                    .build();
        } else {
            s3Client = AmazonS3ClientBuilder.standard().withEndpointConfiguration(
                    new AwsClientBuilder.EndpointConfiguration(s3Options.getEndpoint(), s3Options.getRegion())).build();
        }
        s3Helper = new S3Helper(s3Client, s3Options);
        this.scheduler = new ScheduledThreadPoolExecutor(1,
                new ExecutorThreadFactory("s3-dirty-sink"));
        this.scheduledFuture = this.scheduler.scheduleWithFixedDelay(() -> {
            if (!closed && !flushing) {
                flush();
            }
        }, s3Options.getBatchIntervalMs(), s3Options.getBatchIntervalMs(), TimeUnit.MILLISECONDS);
    }
S3 dirty sink that is used to sink dirty data to s3

@param <T>
inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/dirty/sink/s3/S3DirtySink.java