@Override
        public void run() throws Exception {
            try {
                if (!shutdown) {
                    log.debug("Node {}: beginning shutdown.", nodeId);
                    resign(leader.epoch());
                    for (MetaLogListenerData listenerData : listeners.values()) {
                        listenerData.beginShutdown();
                    }
                    shared.unregisterLogManager(LocalLogManager.this);
                }
            } catch (Exception e) {
                log.error("Unexpected exception while sending beginShutdown callbacks", e);
            }
            shutdown = true;
        }
The current leader, as seen by this log manager.
metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java