@Override
    public synchronized void close() throws IOException {
        if (!closed) {
            closed = true;
            if (this.scheduledFuture != null) {
                scheduledFuture.cancel(false);
                this.scheduler.shutdown();
            }
            try {
                flush();
            } catch (Exception e) {
                LOGGER.warn("Writing records to s3 failed.", e);
                throw new RuntimeException("Writing records to s3 failed.", e);
            }
        }
    }
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