@Test
public void testMetadata() throws Exception {

    long time = 0;
    metadata.update(Cluster.empty(), time);
    assertFalse("No update needed.", metadata.timeToNextUpdate(time) == 0);
    metadata.requestUpdate();
    assertFalse("Still no updated needed due to backoff", metadata.timeToNextUpdate(time) == 0);
    time += refreshBackoffMs;
    assertTrue("Update needed now that backoff time expired", metadata.timeToNextUpdate(time) == 0);
    String topic = "my-topic";
    Thread t1 = asyncFetch(topic);
    Thread t2 = asyncFetch(topic);
    assertTrue("Awaiting update", t1.isAlive());
    assertTrue("Awaiting update", t2.isAlive());
    // Perform metadata update when an update is requested on the async fetch thread
    // This simulates the metadata update sequence in KafkaProducer
    while (t1.isAlive() || t2.isAlive()) {
        if (metadata.timeToNextUpdate(time) == 0) {
            metadata.update(TestUtils.singletonCluster(topic, 1), time);
            time += refreshBackoffMs;
        }
        Thread.sleep(1);
    }
    t1.join();
    t2.join();
    assertFalse("No update needed.", metadata.timeToNextUpdate(time) == 0);
    time += metadataExpireMs;
    assertTrue("Update needed due to stale metadata.", metadata.timeToNextUpdate(time) == 0);
}