synchronized void electLeaderIfNeeded() {
            if (leader.leaderId().isPresent() || logManagers.isEmpty()) {
                return;
            }
            int nextLeaderIndex = ThreadLocalRandom.current().nextInt(logManagers.size());
            Iterator<Integer> iter = logManagers.keySet().iterator();
            Integer nextLeaderNode = null;
            for (int i = 0; i <= nextLeaderIndex; i++) {
                nextLeaderNode = iter.next();
            }
            LeaderAndEpoch newLeader = new LeaderAndEpoch(OptionalInt.of(nextLeaderNode), leader.epoch() + 1);
            log.info("Elected new leader: {}.", newLeader);
            append(OptionalLong.empty(), new LeaderChangeBatch(newLeader));
        }
The initial max read offset which LocalLog instances will be configured with.
metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java