synchronized void revertLastWrittenOffset(
        long offset
    ) {
        if (offset > lastWrittenOffset) {
            throw new IllegalStateException("New offset " + offset + " of " + tp +
                " must be smaller than " + lastWrittenOffset + ".");
        }

        log.debug("Revert last written offset of {} to {}.", tp, offset);
        lastWrittenOffset = offset;
        snapshotRegistry.revertToSnapshot(offset);
    }
Reverts the last written offset. This also reverts the snapshot
registry to this offset. All the changes applied after the offset
are lost.

@param offset The offset to revert to.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/SnapshottableCoordinator.java