CommitStreamSetObjectRequest buildCompactRequest(List<StreamMetadata> streamMetadataList,
        List<S3ObjectMetadata> objectsToCompact)
        throws CompletionException {
        CommitStreamSetObjectRequest request = new CommitStreamSetObjectRequest();
        request.setObjectId(-1L);

        Set<Long> compactedObjectIds = new HashSet<>();
        logger.info("{} stream set objects as compact candidates, total compaction size: {}",
            objectsToCompact.size(), objectsToCompact.stream().mapToLong(S3ObjectMetadata::objectSize).sum());
        Map<Long, List<StreamDataBlock>> streamDataBlockMapToCompact = new HashMap<>();
        for (S3ObjectMetadata objectMetadata : objectsToCompact) {
            List<StreamDataBlock> streamDataBlocks = streamDataBlockMap.get(objectMetadata.objectId());
            if (streamDataBlocks == null) {
                continue;
            }
            streamDataBlockMapToCompact.put(objectMetadata.objectId(), streamDataBlocks);
        }

        long now = System.currentTimeMillis();
        Set<Long> excludedObjectIds = new HashSet<>();
        List<CompactionPlan> compactionPlans = this.compactionAnalyzer.analyze(streamDataBlockMapToCompact, excludedObjectIds);
        logger.info("Analyze compaction plans complete, cost {}ms", System.currentTimeMillis() - now);
        logCompactionPlans(compactionPlans, excludedObjectIds);
        objectsToCompact = objectsToCompact.stream().filter(e -> !excludedObjectIds.contains(e.objectId())).collect(Collectors.toList());
        executeCompactionPlans(request, compactionPlans, objectsToCompact);

        if (!running.get()) {
            logger.info("Compaction manager is shutdown, skip constructing compaction request");
            return null;
        }

        compactionPlans.forEach(c -> c.streamDataBlocksMap().values().forEach(v -> v.forEach(b -> compactedObjectIds.add(b.getObjectId()))));

        // compact out-dated objects directly
        streamDataBlockMapToCompact.entrySet().stream().filter(e -> e.getValue().isEmpty()).forEach(e -> {
            logger.info("Object {} is out of date, will be deleted after compaction", e.getKey());
            compactedObjectIds.add(e.getKey());
        });

        request.setCompactedObjectIds(new ArrayList<>(compactedObjectIds));
        if (isSanityCheckFailed(streamMetadataList, request)) {
            logger.error("Sanity check failed, compaction result is illegal");
            return null;
        }

        return request;
    }
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