void run() {
        if (status != Status.IDLE) {
            return;
        }
        if (partitions.isEmpty()) {
            LOGGER.info("[WORKER_EMPTY],{}", topic);
            return;
        }
        if (requireReset) {
            writers.clear();
            requireReset = false;
        }
        try {
            if (commitRequests.isEmpty() && !commit.starving) {
                advanceSync.sync();
            } else {
                commit.commit();
            }
        } catch (Throwable e) {
            transitionTo(Status.IDLE);
            requireReset = true;
            LOGGER.error("[WORKER_RUN_FAIL],{}", topic, e);
        }
    }
The flag to indicate the worker need reset. The worker status(writers) will be reset in the next round run.
core/src/main/java/kafka/automq/table/worker/TopicPartitionsWorker.java