void activate(long newNextWriteOffset) {
        if (active()) {
            throw new RuntimeException("Can't activate already active OffsetControlManager.");
        }
        if (newNextWriteOffset < 0) {
            throw new RuntimeException("Invalid negative newNextWriteOffset " +
                    newNextWriteOffset + ".");
        }
        // Before switching to active, create an in-memory snapshot at the last committed
        // offset. This is required because the active controller assumes that there is always
        // an in-memory snapshot at the last committed offset.
        snapshotRegistry.getOrCreateSnapshot(lastStableOffset);
        this.nextWriteOffset = newNextWriteOffset;
        metrics.setActive(true);
    }
Called when the QuorumController becomes active.

@param newNextWriteOffset The new next write offset to use. Must be non-negative.
metadata/src/main/java/org/apache/kafka/controller/OffsetControlManager.java