@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.idempotentCreateSnapshot(offset);
        log.debug("Updated last written offset of {} to {}.", tp, offset);
    }
@Test
    public void testUpdateLastWrittenOffset() {
        LogContext logContext = new LogContext();
        SnapshotRegistry snapshotRegistry = new SnapshotRegistry(logContext);
        SnapshottableCoordinator<MockCoordinatorShard, String> coordinator = new SnapshottableCoordinator<>(
            logContext,
            snapshotRegistry,
            new MockCoordinatorShard(snapshotRegistry, new MockCoordinatorTimer<>(new MockTime())),
            new TopicPartition("test-topic", 0)
        );

        assertTrue(coordinator.snapshotRegistry().hasSnapshot(0L));
        coordinator.updateLastWrittenOffset(100L);
        assertEquals(100L, coordinator.lastWrittenOffset());
        assertTrue(coordinator.snapshotRegistry().hasSnapshot(100L));
        assertTrue(coordinator.snapshotRegistry().hasSnapshot(0L));
    }