@Override
        public synchronized void onChange(final Thread thread,
                                          final ThreadStateTransitionValidator abstractNewState,
                                          final ThreadStateTransitionValidator abstractOldState) {
            // StreamThreads first
            if (thread instanceof StreamThread) {
                final StreamThread.State newState = (StreamThread.State) abstractNewState;
                threadState.put(thread.getId(), newState);

                if (newState == StreamThread.State.PARTITIONS_REVOKED || newState == StreamThread.State.PARTITIONS_ASSIGNED) {
                    setState(State.REBALANCING);
                } else if (newState == StreamThread.State.RUNNING) {
                    maybeSetRunning();
                }
            } else if (thread instanceof GlobalStreamThread) {
                // global stream thread has different invariants
                final GlobalStreamThread.State newState = (GlobalStreamThread.State) abstractNewState;
                globalThreadState = newState;

                if (newState == GlobalStreamThread.State.RUNNING) {
                    maybeSetRunning();
                } else if (newState == GlobalStreamThread.State.DEAD) {
                    if (state != State.PENDING_SHUTDOWN) {
                        log.error("Global thread has died. The streams application or client will now close to ERROR.");
                        closeToError();
                    }
                }
            }
        }
If all threads are up, including the global thread, set to RUNNING
streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java