private void transitionTo(
            CoordinatorState newState
        ) {
            if (!newState.canTransitionFrom(state)) {
                throw new IllegalStateException("Cannot transition from " + state + " to " + newState);
            }
            CoordinatorState oldState = state;

            log.debug("Transition from {} to {}.", state, newState);
            switch (newState) {
                case LOADING:
                    state = CoordinatorState.LOADING;
                    SnapshotRegistry snapshotRegistry = new SnapshotRegistry(logContext);
                    coordinator = new SnapshottableCoordinator<>(
                        logContext,
                        snapshotRegistry,
                        coordinatorShardBuilderSupplier
                            .get()
                            .withLogContext(logContext)
                            .withSnapshotRegistry(snapshotRegistry)
                            .withTime(time)
                            .withTimer(timer)
                            .withCoordinatorMetrics(coordinatorMetrics)
                            .withTopicPartition(tp)
                            .build(),
                        tp
                    );
                    load();
                    break;

                case ACTIVE:
                    state = CoordinatorState.ACTIVE;
                    highWatermarklistener = new HighWatermarkListener();
                    partitionWriter.registerListener(tp, highWatermarklistener);
                    coordinator.onLoaded(metadataImage);
                    break;

                case FAILED:
                    state = CoordinatorState.FAILED;
                    unload();
                    break;

                case CLOSED:
                    state = CoordinatorState.CLOSED;
                    unload();
                    break;

                default:
                    throw new IllegalArgumentException("Transitioning to " + newState + " is not supported.");
            }

            runtimeMetrics.recordPartitionStateChange(oldState, state);
        }
Transitions to the new state.

@param newState The new state.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntime.java