@SuppressWarnings("checkstyle:CyclomaticComplexity")
    void executeCompactionPlans(CommitStreamSetObjectRequest request, List<CompactionPlan> compactionPlans,
        List<S3ObjectMetadata> s3ObjectMetadata)
        throws CompletionException {
        if (compactionPlans.isEmpty()) {
            return;
        }
        Map<Long, S3ObjectMetadata> s3ObjectMetadataMap = s3ObjectMetadata.stream()
            .collect(Collectors.toMap(S3ObjectMetadata::objectId, e -> e));
        List<StreamDataBlock> sortedStreamDataBlocks = new ArrayList<>();
        for (int i = 0; i < compactionPlans.size(); i++) {
            if (!running.get()) {
                logger.info("Compaction manager is shutdown, abort compaction progress");
                return;
            }
            // iterate over each compaction plan
            CompactionPlan compactionPlan = compactionPlans.get(i);
            long totalSize = compactionPlan.streamDataBlocksMap().values().stream().flatMap(List::stream)
                .mapToLong(StreamDataBlock::getBlockSize).sum();
            logger.info("Compaction progress {}/{}, read from {} stream set objects, total size: {}", i + 1, compactionPlans.size(),
                compactionPlan.streamDataBlocksMap().size(), totalSize);
            for (Map.Entry<Long, List<StreamDataBlock>> streamDataBlocEntry : compactionPlan.streamDataBlocksMap().entrySet()) {
                S3ObjectMetadata metadata = s3ObjectMetadataMap.get(streamDataBlocEntry.getKey());
                List<StreamDataBlock> streamDataBlocks = streamDataBlocEntry.getValue();
                DataBlockReader reader = new DataBlockReader(metadata, objectStorage, compactionBucket, bucketCallbackScheduledExecutor);
                reader.readBlocks(streamDataBlocks, Math.min(CompactionConstants.S3_OBJECT_MAX_READ_BATCH, networkBandwidth));
            }
            List<CompletableFuture<StreamObject>> streamObjectCfList = new ArrayList<>();
            CompletableFuture<Void> streamSetObjectChainWriteCf = CompletableFuture.completedFuture(null);
            for (CompactedObject compactedObject : compactionPlan.compactedObjects()) {
                if (compactedObject.type() == CompactionType.COMPACT) {
                    sortedStreamDataBlocks.addAll(compactedObject.streamDataBlocks());
                    streamSetObjectChainWriteCf = uploader.chainWriteStreamSetObject(streamSetObjectChainWriteCf, compactedObject);
                } else {
                    streamObjectCfList.add(uploader.writeStreamObject(compactedObject));
                }
            }

            List<CompletableFuture<?>> cfList = new ArrayList<>();
            cfList.add(streamSetObjectChainWriteCf);
            cfList.addAll(streamObjectCfList);
            synchronized (this) {
                if (!running.get()) {
                    logger.info("Compaction manager is shutdown, skip waiting for uploading objects");
                    return;
                }
                // wait for all stream objects and stream set object part to be uploaded
                compactionCf = new CompletableFuture<>();
                CompletableFuture.allOf(cfList.toArray(new CompletableFuture[0]))
                    .whenComplete((v, uploadException) -> {
                        if (uploadException != null) {
                            logger.error("Error while uploading compaction objects", uploadException);
                        }
                        FutureUtil.exec(uploader::forceUploadStreamSetObject, logger, "force upload sso")
                            .whenComplete((vv, forceUploadException) -> {
                                if (forceUploadException != null) {
                                    logger.error("Error while force uploading stream set object", uploadException);
                                }
                                if (uploadException != null || forceUploadException != null) {
                                    FutureUtil.exec(uploader::release, logger, "release uploader").whenComplete((vvv, releaseException) -> {
                                        if (releaseException != null) {
                                            logger.error("Unexpected exception while release uploader");
                                        }
                                        for (CompactedObject compactedObject : compactionPlan.compactedObjects()) {
                                            compactedObject.streamDataBlocks().forEach(StreamDataBlock::release);
                                        }
                                        if (uploadException != null) {
                                            compactionCf.completeExceptionally(new CompletionException("Uploading failed", uploadException));
                                        } else {
                                            compactionCf.completeExceptionally(new CompletionException("Force uploading sso failed", forceUploadException));
                                        }
                                    });
                                } else {
                                    compactionCf.complete(null);
                                }
                            });
                    });
            }
            try {
                compactionCf.join();
            } catch (CancellationException ex) {
                // release all data blocks on shutdown
                for (CompactedObject compactedObject : compactionPlan.compactedObjects()) {
                    compactedObject.streamDataBlocks().forEach(StreamDataBlock::release);
                }
                logger.warn("Compaction progress {}/{} is cancelled", i + 1, compactionPlans.size());
                return;
            }
            compactionCf = null;

            streamObjectCfList.stream().map(CompletableFuture::join).forEach(request::addStreamObject);

            if (ByteBufAlloc.getPolicy().isPooled()) {
                // Check if all blocks are released after each iteration
                List<CompactedObject> compactedObjects = compactionPlan.compactedObjects();
                for (CompactedObject compactedObject : compactedObjects) {
                    for (StreamDataBlock block : compactedObject.streamDataBlocks()) {
                        if (block.getDataCf().join().refCnt() > 0) {
                            logger.error("Block {} is not released after compaction, compact type: {}", block, compactedObject.type());
                        }
                    }
                }
            }
        }
        List<ObjectStreamRange> objectStreamRanges = CompactionUtils.buildObjectStreamRangeFromGroup(
            CompactionUtils.groupStreamDataBlocks(sortedStreamDataBlocks, new GroupByOffsetPredicate()));
        objectStreamRanges.forEach(request::addStreamRange);
        request.setObjectId(uploader.getStreamSetObjectId());
        // set stream set object id to be the first object id of compacted objects
        request.setOrderId(s3ObjectMetadata.get(0).objectId());
        request.setObjectSize(uploader.complete());
        request.setAttributes(ObjectAttributes.builder().bucket(uploader.bucketId()).build().attributes());
    }
Split specified stream set object into stream objects.

@param objectMetadata     stream set object to split
@param cfs                List of CompletableFuture of StreamObject
@return true if split succeed, false otherwise
s3stream/src/main/java/com/automq/stream/s3/compact/CompactionManager.java