public void endOffsets(int microSyncBatchSize, double avgRecordSize) {
            int avgRecordCountPerPartition = Math.max((int) Math.ceil(microSyncBatchSize / avgRecordSize / offsetBounds.size()), 1);
            long[] fixedEndOffsets = new long[offsetBounds.size()];
            hasMoreData = offsetBounds.stream().mapToLong(bound -> bound.end - bound.start).sum() * avgRecordSize > microSyncBatchSize;
            if (!hasMoreData) {
                return;
            }
            int remaining = 0;
            for (int i = 0; i < offsetBounds.size(); i++) {
                OffsetBound offsetBound = offsetBounds.get(i);
                int count = (int) Math.min(offsetBound.end - offsetBound.start, avgRecordCountPerPartition);
                fixedEndOffsets[i] = offsetBound.start + count;
                remaining += avgRecordCountPerPartition - count;
            }
            for (int i = 0; i < offsetBounds.size() && remaining > 0; i++) {
                OffsetBound offsetBound = offsetBounds.get(i);
                int add = (int) Math.min(offsetBound.end - fixedEndOffsets[i], remaining);
                fixedEndOffsets[i] += add;
                remaining -= add;
            }
            for (int i = 0; i < offsetBounds.size(); i++) {
                offsetBounds.get(i).end = fixedEndOffsets[i];
            }
        }
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