@Override
    public synchronized void close() {
        super.close();
        notifyAll();
    }
@Test
    public void testMetadataEquivalentResponsesBackoff() throws InterruptedException {
        long time = 0;
        metadata.updateWithCurrentRequestVersion(responseWithCurrentTopics(), false, time);
        assertTrue(metadata.timeToNextUpdate(time) > 0, "No update needed");
        metadata.requestUpdate(false);
        assertTrue(metadata.timeToNextUpdate(time) > 0, "Still no update needed due to backoff");
        time += (long) (refreshBackoffMs * (1 + CommonClientConfigs.RETRY_BACKOFF_JITTER));
        metadata.updateWithCurrentRequestVersion(responseWithCurrentTopics(), false, time);
        assertTrue(metadata.timeToNextUpdate(time) > 0, "No update needed after equivalent metadata response");
        metadata.requestUpdate(false);
        assertTrue(metadata.timeToNextUpdate(time) > 0, "Still no update needed due to backoff");
        assertTrue(metadata.timeToNextUpdate(time + refreshBackoffMs) > 0, "Still no updated needed due to exponential backoff");
        time += (long) (refreshBackoffMs * CommonClientConfigs.RETRY_BACKOFF_EXP_BASE * (1 + CommonClientConfigs.RETRY_BACKOFF_JITTER));
        assertEquals(0, metadata.timeToNextUpdate(time), "Update needed now that backoff time expired");
        String topic = "my-topic";
        metadata.close();
        Thread t1 = asyncFetch(topic, 500);
        t1.join();
        assertEquals(KafkaException.class, backgroundError.get().getClass());
        assertTrue(backgroundError.get().toString().contains("Requested metadata update after close"));
        clearBackgroundError();
    }