public CompletableFuture<SyncTask> sync() {
            TimerUtil timer = new TimerUtil();
            plan();
            return CompletableFuture.allOf(microSyncBatchTasks
                    .stream()
                    .map(t -> t.run().thenApply(rst -> {
                        if (rst.ctx.requireReset) {
                            TopicPartitionsWorker.this.requireReset = true;
                        }
                        updateDecompressedRatio();
                        return rst;
                    })).toArray(CompletableFuture[]::new)
                )
                .thenApply(nil -> this)
                .whenComplete((rst, ex) -> {
                    moreQuota.release();
                    if (ex != null) {
                        LOGGER.error("[SYNC_TASK_FAIL],{}", logContext, ex);
                        TopicPartitionsWorker.this.requireReset = true;
                    } else {
                        TopicPartitionsWorker.this.writers.addAll(writers);
                        List<OffsetBound> offsetBounds = microSyncBatchTasks.stream().flatMap(t -> t.offsetBounds().stream()).collect(Collectors.toList());
                        double size = avgRecordSize * offsetBounds.stream().mapToLong(bound -> bound.end - bound.start).sum();
                        LOGGER.info("[SYNC_TASK],{},size={},decompressedRatio={},parallel={},partitions={},cost={}ms", logContext, size, decompressedRatio, microSyncBatchTasks.size(), offsetBounds, timer.elapsedAs(TimeUnit.MILLISECONDS));
                    }
                });
        }
Split sync task to parallel micro sync batch tasks.
core/src/main/java/kafka/automq/table/worker/TopicPartitionsWorker.java