Collection<CompletableFuture<StreamObject>> groupAndSplitStreamDataBlocks(S3ObjectMetadata objectMetadata,
        List<StreamDataBlock> streamDataBlocks) {
        List<Pair<List<StreamDataBlock>, CompletableFuture<StreamObject>>> groupedDataBlocks = new ArrayList<>();
        List<List<StreamDataBlock>> groupedStreamDataBlocks = CompactionUtils.groupStreamDataBlocks(streamDataBlocks, new GroupByOffsetPredicate());
        for (List<StreamDataBlock> group : groupedStreamDataBlocks) {
            groupedDataBlocks.add(new ImmutablePair<>(group, new CompletableFuture<>()));
        }
        logger.info("Force split object {}, expect to generate {} stream objects", objectMetadata.objectId(), groupedDataBlocks.size());

        int index = 0;
        while (index < groupedDataBlocks.size()) {
            List<Pair<List<StreamDataBlock>, CompletableFuture<StreamObject>>> batchGroup = new ArrayList<>();
            long readSize = 0;
            while (index < groupedDataBlocks.size()) {
                Pair<List<StreamDataBlock>, CompletableFuture<StreamObject>> group = groupedDataBlocks.get(index);
                List<StreamDataBlock> groupedStreamDataBlock = group.getLeft();
                long size = groupedStreamDataBlock.get(groupedStreamDataBlock.size() - 1).getBlockEndPosition() -
                    groupedStreamDataBlock.get(0).getBlockStartPosition();
                if (readSize + size > compactionCacheSize) {
                    break;
                }
                readSize += size;
                batchGroup.add(group);
                index++;
            }
            if (batchGroup.isEmpty()) {
                logger.error("Force split object failed, not be able to read any data block, maybe compactionCacheSize is too small");
                return new ArrayList<>();
            }
            // prepare N stream objects at one time
            objectManager.prepareObject(batchGroup.size(), TimeUnit.MINUTES.toMillis(CompactionConstants.S3_OBJECT_TTL_MINUTES))
                .thenComposeAsync(objectId -> {
                    List<StreamDataBlock> blocksToRead = batchGroup.stream().flatMap(p -> p.getLeft().stream()).collect(Collectors.toList());
                    DataBlockReader reader = new DataBlockReader(objectMetadata, objectStorage, compactionBucket, bucketCallbackScheduledExecutor);
                    // batch read
                    reader.readBlocks(blocksToRead, Math.min(CompactionConstants.S3_OBJECT_MAX_READ_BATCH, networkBandwidth));

                    List<CompletableFuture<Void>> cfs = new ArrayList<>();
                    for (Pair<List<StreamDataBlock>, CompletableFuture<StreamObject>> pair : batchGroup) {
                        List<StreamDataBlock> blocks = pair.getLeft();
                        DataBlockWriter writer = new DataBlockWriter(objectId, objectStorage, config.objectPartSize());
                        CompletableFuture<Void> cf = CompactionUtils.chainWriteDataBlock(writer, blocks, forceSplitThreadPool);
                        long finalObjectId = objectId;
                        cfs.add(cf.thenCompose(nil -> writer.close()).whenComplete((ret, ex) -> {
                            if (ex != null) {
                                logger.error("write to stream object {} failed", finalObjectId, ex);
                                writer.release();
                                blocks.forEach(StreamDataBlock::release);
                                return;
                            }
                            StreamObject streamObject = new StreamObject();
                            streamObject.setObjectId(finalObjectId);
                            streamObject.setStreamId(blocks.get(0).getStreamId());
                            streamObject.setStartOffset(blocks.get(0).getStartOffset());
                            streamObject.setEndOffset(blocks.get(blocks.size() - 1).getEndOffset());
                            streamObject.setObjectSize(writer.size());
                            streamObject.setAttributes(ObjectAttributes.builder().bucket(writer.bucketId()).build().attributes());
                            pair.getValue().complete(streamObject);
                        }));
                        objectId++;
                    }
                    return CompletableFuture.allOf(cfs.toArray(new CompletableFuture[0]));
                }, forceSplitThreadPool)
                .exceptionally(ex -> {
                    logger.error("Force split object {} failed", objectMetadata.objectId(), ex);
                    for (Pair<List<StreamDataBlock>, CompletableFuture<StreamObject>> pair : groupedDataBlocks) {
                        pair.getValue().completeExceptionally(ex);
                    }
                    throw new IllegalStateException(String.format("Force split object %d failed", objectMetadata.objectId()), ex);
                }).join();
        }

        return groupedDataBlocks.stream().map(Pair::getValue).collect(Collectors.toList());
    }
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