@Test
    public void testPartialMetadataUpdate() {
        Time time = new MockTime();

        metadata = new Metadata(refreshBackoffMs, refreshBackoffMaxMs, metadataExpireMs, new LogContext(), new ClusterResourceListeners()) {
                @Override
                protected MetadataRequest.Builder newMetadataRequestBuilderForNewTopics() {
                    return newMetadataRequestBuilder();
                }
            };

        assertFalse(metadata.updateRequested());

        // Request a metadata update. This must force a full metadata update request.
        metadata.requestUpdate(true);
        Metadata.MetadataRequestAndVersion versionAndBuilder = metadata.newMetadataRequestAndVersion(time.milliseconds());
        assertFalse(versionAndBuilder.isPartialUpdate);
        metadata.update(versionAndBuilder.requestVersion,
                RequestTestUtils.metadataUpdateWith(1, Collections.singletonMap("topic", 1)), false, time.milliseconds());
        assertFalse(metadata.updateRequested());

        // Request a metadata update for a new topic. This should perform a partial metadata update.
        metadata.requestUpdateForNewTopics();
        versionAndBuilder = metadata.newMetadataRequestAndVersion(time.milliseconds());
        assertTrue(versionAndBuilder.isPartialUpdate);
        metadata.update(versionAndBuilder.requestVersion,
                RequestTestUtils.metadataUpdateWith(1, Collections.singletonMap("topic", 1)), true, time.milliseconds());
        assertFalse(metadata.updateRequested());

        // Request both types of metadata updates. This should always perform a full update.
        metadata.requestUpdate(true);
        metadata.requestUpdateForNewTopics();
        versionAndBuilder = metadata.newMetadataRequestAndVersion(time.milliseconds());
        assertFalse(versionAndBuilder.isPartialUpdate);
        metadata.update(versionAndBuilder.requestVersion,
                RequestTestUtils.metadataUpdateWith(1, Collections.singletonMap("topic", 1)), false, time.milliseconds());
        assertFalse(metadata.updateRequested());

        // Request only a partial metadata update, but elapse enough time such that a full refresh is needed.
        metadata.requestUpdateForNewTopics();
        final long refreshTimeMs = time.milliseconds() + metadata.metadataExpireMs();
        versionAndBuilder = metadata.newMetadataRequestAndVersion(refreshTimeMs);
        assertFalse(versionAndBuilder.isPartialUpdate);
        metadata.update(versionAndBuilder.requestVersion,
                RequestTestUtils.metadataUpdateWith(1, Collections.singletonMap("topic", 1)), true, refreshTimeMs);
        assertFalse(metadata.updateRequested());

        // Request two partial metadata updates that are overlapping.
        metadata.requestUpdateForNewTopics();
        versionAndBuilder = metadata.newMetadataRequestAndVersion(time.milliseconds());
        assertTrue(versionAndBuilder.isPartialUpdate);
        metadata.requestUpdateForNewTopics();
        Metadata.MetadataRequestAndVersion overlappingVersionAndBuilder = metadata.newMetadataRequestAndVersion(time.milliseconds());
        assertTrue(overlappingVersionAndBuilder.isPartialUpdate);
        assertTrue(metadata.updateRequested());
        metadata.update(versionAndBuilder.requestVersion,
                RequestTestUtils.metadataUpdateWith(1, Collections.singletonMap("topic-1", 1)), true, time.milliseconds());
        assertTrue(metadata.updateRequested());
        metadata.update(overlappingVersionAndBuilder.requestVersion,
                RequestTestUtils.metadataUpdateWith(1, Collections.singletonMap("topic-2", 1)), true, time.milliseconds());
        assertFalse(metadata.updateRequested());
    }
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