@Override
    public void resign(int epoch) {
        if (epoch < 0) {
            throw new IllegalArgumentException("Attempt to resign from an invalid negative epoch " + epoch);
        }

        LeaderAndEpoch leaderAndEpoch = leaderAndEpoch();
        int currentEpoch = leaderAndEpoch.epoch();

        if (epoch > currentEpoch) {
            throw new IllegalArgumentException("Attempt to resign from epoch " + epoch +
                    " which is larger than the current epoch " + currentEpoch);
        } else if (epoch < currentEpoch) {
            // If the passed epoch is smaller than the current epoch, then it might mean
            // that the listener has not been notified about a leader change that already
            // took place. In this case, we consider the call as already fulfilled and
            // take no further action.
            log.debug("Ignoring call to resign from epoch {} since it is smaller than the " +
                    "current epoch {}", epoch, currentEpoch);
            return;
        }

        LeaderAndEpoch nextLeader = new LeaderAndEpoch(OptionalInt.empty(), currentEpoch + 1);
        try {
            shared.tryAppend(nodeId,
                    currentEpoch,
                    OptionalLong.empty(),
                    new LeaderChangeBatch(nextLeader));
        } catch (NotLeaderException exp) {
            // the leader epoch has already advanced. resign is a no op.
            log.debug("Ignoring call to resign from epoch {}. Either we are not the leader or the provided epoch is " +
                    "smaller than the current epoch {}", epoch, currentEpoch);
        }
    }
Shutdown the log manager.

Even though the API suggests a non-blocking shutdown, this method always returns a completed
future. This means that shutdown is a blocking operation.
metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java