void plan() {
            Map<Integer, OffsetBound> partition2bound = new HashMap<>();
            for (Map.Entry<Integer, Long> entry : startOffsets.entrySet()) {
                int partitionId = entry.getKey();
                Partition partition = partitions.get(partitionId);
                OffsetBound offsetBound = new OffsetBound(partition, startOffsets.get(partitionId), partition.log().get().highWatermark());
                partition2bound.put(partitionId, offsetBound);
            }
            double totalDirtyBytes = partition2bound.values().stream().mapToLong(bound -> bound.end - bound.start).sum() * avgRecordSize * decompressedRatio;
            int parallel = Math.max((int) (Math.ceil(totalDirtyBytes / config.microSyncBatchSize())), 1);
            parallel = Math.min(Math.min(parallel, eventLoops.size()), partition2bound.size());
            moreQuota = tryAcquireMoreCommitQuota(parallel - 1);
            parallel = moreQuota.value() + 1;

            List<Integer> partitionSortedBySize = partition2bound.entrySet().stream().sorted((e1, e2) -> {
                long size1 = e1.getValue().end - e1.getValue().start;
                long size2 = e2.getValue().end - e2.getValue().start;
                return Long.compare(size2, size1);
            }).map(Map.Entry::getKey).collect(Collectors.toList());

            for (int i = 0; i < parallel; i++) {
                EventLoops.EventLoopRef workerEventLoop = eventLoops.leastLoadEventLoop();
                Writer taskWriter = writerFactory.newWriter();
                writers.add(taskWriter);
                PartitionWriteTaskContext ctx = new PartitionWriteTaskContext(taskWriter, workerEventLoop, flushExecutors, config, priority);
                microSyncBatchTasks.add(new MicroSyncBatchTask(logContext, ctx));
            }
            for (int i = 0; i < partitionSortedBySize.size(); i++) {
                int partitionId = partitionSortedBySize.get(i);
                microSyncBatchTasks.get(i % parallel).addPartition(partition2bound.get(partitionId));
            }
            for (MicroSyncBatchTask task : microSyncBatchTasks) {
                task.startOffsets(startOffsets);
                task.endOffsets(config.microSyncBatchSize(), avgRecordSize * decompressedRatio);
                task.offsetBounds().forEach(bound -> startOffsets.put(bound.partition.partitionId(), bound.end));
                if (task.hasMoreData()) {
                    hasMoreData = true;
                }
            }
        }
Split sync task to parallel micro sync batch tasks.
core/src/main/java/kafka/automq/table/worker/TopicPartitionsWorker.java