@Test
    public void testClusterListenerGetsNotifiedOfUpdate() {
        MockClusterResourceListener mockClusterListener = new MockClusterResourceListener();
        ClusterResourceListeners listeners = new ClusterResourceListeners();
        listeners.maybeAdd(mockClusterListener);
        metadata = new Metadata(refreshBackoffMs, refreshBackoffMaxMs,
                metadataExpireMs, new LogContext(), listeners);

        String hostName = "www.example.com";
        metadata.bootstrap(Collections.singletonList(new InetSocketAddress(hostName, 9002)));
        assertFalse(MockClusterResourceListener.IS_ON_UPDATE_CALLED.get(),
            "ClusterResourceListener should not called when metadata is updated with bootstrap Cluster");

        Map<String, Integer> partitionCounts = new HashMap<>();
        partitionCounts.put("topic", 1);
        partitionCounts.put("topic1", 1);
        MetadataResponse metadataResponse = RequestTestUtils.metadataUpdateWith("dummy", 1, partitionCounts);
        metadata.updateWithCurrentRequestVersion(metadataResponse, false, 100);

        assertEquals("dummy", mockClusterListener.clusterResource().clusterId(),
            "MockClusterResourceListener did not get cluster metadata correctly");
        assertTrue(MockClusterResourceListener.IS_ON_UPDATE_CALLED.get(),
            "MockClusterResourceListener should be called when metadata is updated with non-bootstrap Cluster");
    }
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