@Test
    public void testNodeIfOffline() {
        Map<String, Integer> partitionCounts = new HashMap<>();
        partitionCounts.put("topic-1", 1);
        Node node0 = new Node(0, "localhost", 9092);
        Node node1 = new Node(1, "localhost", 9093);

        MetadataResponse metadataResponse = RequestTestUtils.metadataUpdateWith("dummy", 2, Collections.emptyMap(), partitionCounts, _tp -> 99,
            (error, partition, leader, leaderEpoch, replicas, isr, offlineReplicas) ->
                new MetadataResponse.PartitionMetadata(error, partition, Optional.of(node0.id()), leaderEpoch,
                    Collections.singletonList(node0.id()), Collections.emptyList(),
                        Collections.singletonList(node1.id())), ApiKeys.METADATA.latestVersion(), Collections.emptyMap());
        metadata.updateWithCurrentRequestVersion(emptyMetadataResponse(), false, 0L);
        metadata.updateWithCurrentRequestVersion(metadataResponse, false, 10L);

        TopicPartition tp = new TopicPartition("topic-1", 0);

        assertOptional(metadata.fetch().nodeIfOnline(tp, 0), node -> assertEquals(node.id(), 0));
        assertFalse(metadata.fetch().nodeIfOnline(tp, 1).isPresent());
        assertEquals(metadata.fetch().nodeById(0).id(), 0);
        assertEquals(metadata.fetch().nodeById(1).id(), 1);
    }
Prior to Kafka version 2.4 (which coincides with Metadata version 9), the broker does not propagate leader epoch
information accurately while a reassignment is in progress, so we cannot rely on it. This is explained in more
detail in MetadataResponse's constructor.
clients/src/test/java/org/apache/kafka/clients/MetadataTest.java