private void write0(WriteOptions options, String path, ByteBuf data, CompletableFuture<Void> attemptCf,
        CompletableFuture<Void> finalCf) {
        TimerUtil timerUtil = new TimerUtil();
        long objectSize = data.readableBytes();

        if (checkTimeout(options, finalCf)) {
            attemptCf.completeExceptionally(new TimeoutException());
            data.release();
            return;
        }

        CompletableFuture<Void> writeCf = doWrite(options, path, data);
        FutureUtil.propagate(writeCf, attemptCf);
        AtomicBoolean completedFlag = new AtomicBoolean(false);
        WriteOptions retryOptions = options.copy().retry(true);

        // Fast retry should only be triggered by the original request.
        long delayMillis = s3LatencyCalculator.valueAtPercentile(objectSize, 99);

        if (options.enableFastRetry() && delayMillis > 0 && !options.retry()) {
            data.retain();
            fastRetryTimer.newTimeout(timeout -> {
                if (writeCf != null && !writeCf.isDone() && fastRetryPermit.tryAcquire()) {
                    TimerUtil retryTimerUtil = new TimerUtil();

                    doWrite(retryOptions, path, data).thenAccept(nil -> {
                        recordWriteStats(path, objectSize, retryTimerUtil);

                        data.release();
                        fastRetryPermit.release();

                        if (completedFlag.compareAndSet(false, true)) {
                            finalCf.complete(null);
                            logger.info("Fast retry: put object {} with size {}, cost {}ms, delay {}ms", path, objectSize, retryTimerUtil.elapsedAs(TimeUnit.MILLISECONDS), delayMillis);
                        } else {
                            logger.info("Fast retry but duplicated: put object {} with size {}, cost {}ms, delay {}ms", path, objectSize, retryTimerUtil.elapsedAs(TimeUnit.MILLISECONDS), delayMillis);
                        }
                    }).exceptionally(ignore -> {
                        data.release();
                        fastRetryPermit.release();

                        // The fast retry request will not retry again.
                        S3OperationStats.getInstance().putObjectStats(objectSize, false).record(retryTimerUtil.elapsedAs(TimeUnit.NANOSECONDS));
                        return null;
                    });
                } else {
                    data.release();
                }
            }, delayMillis, TimeUnit.MILLISECONDS);
        }

        writeCf.thenAccept(nil -> {
            recordWriteStats(path, objectSize, timerUtil);
            data.release();
            if (completedFlag.compareAndSet(false, true)) {
                finalCf.complete(null);
            }
        }).exceptionally(ex -> {
            S3OperationStats.getInstance().putObjectStats(objectSize, false).record(timerUtil.elapsedAs(TimeUnit.NANOSECONDS));
            Pair<RetryStrategy, Throwable> strategyAndCause = toRetryStrategyAndCause(ex, S3Operation.PUT_OBJECT);
            RetryStrategy retryStrategy = strategyAndCause.getLeft();
            Throwable cause = strategyAndCause.getRight();

            if (retryStrategy == RetryStrategy.ABORT || checkS3ApiMode) {
                // no need to retry
                logger.error("PutObject for object {} fail", path, cause);
                data.release();
                if (completedFlag.compareAndSet(false, true)) {
                    finalCf.completeExceptionally(cause);
                }
                return null;
            }

            int retryCount = retryOptions.retryCountGetAndAdd();
            if (isThrottled(cause, retryCount)) {
                failedWriteMonitor.record(objectSize);
                logger.warn("PutObject for object {} fail, retry count {}, queued and retry later", path, retryCount, cause);
                queuedWrite0(retryOptions, path, data, finalCf);
            } else {
                int delay = retryDelay(S3Operation.PUT_OBJECT, retryCount);
                logger.warn("PutObject for object {} fail, retry count {}, retry in {}ms", path, retryCount, delay, cause);
                delayedWrite0(retryOptions, path, data, finalCf, delay);
            }
            return null;
        });
    }
Put an object to the specified path.

@param options   options (or context) about the write operation
@param path      the path to put the object
@param data      the data to put, it will be released once the finalCf is done
@param attemptCf the CompletableFuture to complete when a single attempt is done
@param finalCf   the CompletableFuture to complete when the write operation is done
s3stream/src/main/java/com/automq/stream/s3/operator/AbstractObjectStorage.java