private Optional<Map<Integer, Long>> getStartOffsets() {
            if (!transitionTo(Status.ADVANCE_SYNC)) {
                TableWorkers.SCHEDULER.schedule(() -> eventLoop.submit(TopicPartitionsWorker.this::run), 100, TimeUnit.MILLISECONDS);
                return Optional.empty();
            }
            if (!fastNextSync.get()
                // The commit speed can't keep up with the advance sync rate.
                // We need to limit the number of writers to prevent excessive heap usage.
                || writers.size() >= WRITER_LIMIT
            ) {
                transitionTo(Status.IDLE);
                return Optional.empty();
            }
            if (writers.isEmpty()) {
                // Await for the first commit request to know the sync start offset
                transitionTo(Status.IDLE);
                return Optional.empty();
            }
            Writer lastWriter = writers.get(writers.size() - 1);
            Map<Integer, Long> startOffsets = new HashMap<>();
            lastWriter.getOffsets().forEach((partition, offset) -> startOffsets.put(partition, offset.end()));
            this.dirtyBytesCounter.set(0);
            return Optional.of(startOffsets);
        }
Try to sync the 'hot data' to the writer.
<p>
Note: it should run in event loop.
core/src/main/java/kafka/automq/table/worker/TopicPartitionsWorker.java