@Override
    public synchronized void updateLastCommittedOffset(Long offset) {
        if (offset < lastCommittedOffset) {
            throw new IllegalStateException("New committed offset " + offset + " of " + tp +
                " must be greater than or equal to " + lastCommittedOffset + ".");
        }

        if (offset > lastWrittenOffset) {
            throw new IllegalStateException("New committed offset " + offset + " of " + tp +
                "must be less than or equal to " + lastWrittenOffset + ".");
        }

        lastCommittedOffset = offset;
        snapshotRegistry.deleteSnapshotsUpTo(offset);
        log.debug("Updated committed offset of {} to {}.", tp, offset);
    }
@Test
    public void testUpdateLastCommittedOffset() {
        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)
        );

        coordinator.updateLastWrittenOffset(100L);
        assertTrue(coordinator.snapshotRegistry().hasSnapshot(0L));
        assertTrue(coordinator.snapshotRegistry().hasSnapshot(100L));

        coordinator.updateLastCommittedOffset(100L);
        assertEquals(100L, coordinator.lastCommittedOffset());
        assertFalse(coordinator.snapshotRegistry().hasSnapshot(0L));
        assertTrue(coordinator.snapshotRegistry().hasSnapshot(100L));
    }