@Override
    public synchronized void updateLastWrittenOffset(Long offset) {
        if (offset <= lastWrittenOffset) {
            throw new IllegalStateException("New last written offset " + offset + " of " + tp +
                " must be greater than " + lastWrittenOffset + ".");
        }

        lastWrittenOffset = offset;
        snapshotRegistry.getOrCreateSnapshot(offset);
        log.debug("Updated last written offset of {} to {}.", tp, offset);
    }
Updates the last written offset. This also create a new snapshot
in the snapshot registry.

@param offset The new last written offset.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/SnapshottableCoordinator.java