private void uploadPart0(WriteOptions options, String path, String uploadId, int partNumber, ByteBuf data,
        CompletableFuture<ObjectStorageCompletedPart> attemptCf,
        CompletableFuture<ObjectStorageCompletedPart> finalCf) {
        if (checkTimeout(options, finalCf)) {
            attemptCf.completeExceptionally(new TimeoutException());
            return;
        }
        TimerUtil timerUtil = new TimerUtil();
        int size = data.readableBytes();
        CompletableFuture<ObjectStorageCompletedPart> uploadPartCf = doUploadPart(options, path, uploadId, partNumber, data);
        FutureUtil.propagate(uploadPartCf, attemptCf);
        uploadPartCf.thenAccept(part -> {
            S3OperationStats.getInstance().uploadSizeTotalStats.add(MetricsLevel.INFO, size);
            S3OperationStats.getInstance().uploadPartStats(size, true).record(timerUtil.elapsedAs(TimeUnit.NANOSECONDS));
            successWriteMonitor.record(size);
            finalCf.complete(part);
        }).exceptionally(ex -> {
            S3OperationStats.getInstance().uploadPartStats(size, false).record(timerUtil.elapsedAs(TimeUnit.NANOSECONDS));
            Pair<RetryStrategy, Throwable> strategyAndCause = toRetryStrategyAndCause(ex, S3Operation.UPLOAD_PART);
            RetryStrategy retryStrategy = strategyAndCause.getLeft();
            Throwable cause = strategyAndCause.getRight();

            if (retryStrategy == RetryStrategy.ABORT || checkS3ApiMode) {
                // no need to retry
                logger.error("UploadPart for object {}-{} fail", path, partNumber, cause);
                finalCf.completeExceptionally(cause);
                return null;
            }

            int retryCount = options.retryCountGetAndAdd();
            if (isThrottled(cause, retryCount)) {
                failedWriteMonitor.record(size);
                logger.warn("UploadPart for object {}-{} fail, retry count {}, queued and retry later", path, partNumber, retryCount, cause);
                queuedUploadPart0(options, path, uploadId, partNumber, data, finalCf);
            } else {
                int delay = retryDelay(S3Operation.UPLOAD_PART, retryCount);
                logger.warn("UploadPart for object {}-{} fail, retry count {}, retry in {}ms", path, partNumber, retryCount, delay, cause);
                delayedUploadPart0(options, path, uploadId, partNumber, data, finalCf, delay);
            }
            return null;
        });
    }
Upload a part of an object to the specified path.

@param options    options (or context) about the write operation
@param path       the path of the object where the part will be uploaded
@param uploadId   the upload ID of the multipart upload
@param partNumber the part number of the part to be uploaded
@param data       the data to be uploaded, 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 upload operation is done
s3stream/src/main/java/com/automq/stream/s3/operator/AbstractObjectStorage.java