public CompletableFuture<MicroSyncBatchTask> run() {
            List<CompletableFuture<Void>> partitionCfList = new ArrayList<>();
            offsetBounds.forEach(bound -> partitionCfList.add(runPartitionWriteTask(bound.partition, bound.start, bound.end)));
            return CompletableFuture.allOf(partitionCfList.toArray(new CompletableFuture[0]))
                .thenComposeAsync(nil -> {
                    if (ctx.requireReset) {
                        return CompletableFuture.completedFuture(null);
                    } else {
                        return ctx.writer.flush(FlushMode.COMPLETE, ctx.flushExecutors, ctx.eventLoop)
                            .exceptionally(ex -> {
                                ctx.requireReset = true;
                                LOGGER.error("[FLUSH_FAIL],{}", logContext, ex);
                                return null;
                            });
                    }
                }, ctx.eventLoop).thenApply(nil -> this).whenComplete((nil, ex) -> {
                    ctx.eventLoop.release();
                });
        }
The micro sync batch task syncs the records in partition - [start, end) to iceberg.
How to use:
1. #addPartition: add the partition to the task.
2. #startOffsets: set start offsets for partition syncing.
3. #endOffset: adjust the end offsets for partition syncing according to the batch size and record size.
4. #run: run the task.
core/src/main/java/kafka/automq/table/worker/TopicPartitionsWorker.java