public synchronized void add(String topic, long nowMs) {
        Objects.requireNonNull(topic, "topic cannot be null");
        if (topics.put(topic, nowMs + metadataIdleMs) == null) {
            newTopics.add(topic);
            requestUpdateForNewTopics();
        }
    }
@Test
    public void testMetadata() throws Exception {
        long time = Time.SYSTEM.milliseconds();
        String topic = "my-topic";
        metadata.add(topic, time);

        metadata.updateWithCurrentRequestVersion(responseWithTopics(Collections.emptySet()), false, time);
        assertTrue(metadata.timeToNextUpdate(time) > 0, "No update needed.");
        metadata.requestUpdate(true);
        assertTrue(metadata.timeToNextUpdate(time) > 0, "Still no updated needed due to backoff");
        time += (long) (refreshBackoffMs * (1 + CommonClientConfigs.RETRY_BACKOFF_JITTER));
        assertEquals(0, metadata.timeToNextUpdate(time), "Update needed now that backoff time expired");
        Thread t1 = asyncFetch(topic, 500);
        Thread t2 = asyncFetch(topic, 500);
        assertTrue(t1.isAlive(), "Awaiting update");
        assertTrue(t2.isAlive(), "Awaiting update");
        // 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.updateWithCurrentRequestVersion(responseWithCurrentTopics(), false, time);
                time += (long) (refreshBackoffMs * (1 + CommonClientConfigs.RETRY_BACKOFF_JITTER));
            }
            Thread.sleep(1);
        }
        t1.join();
        t2.join();
        assertTrue(metadata.timeToNextUpdate(time) > 0, "No update needed.");
        time += metadataExpireMs;
        assertEquals(0, metadata.timeToNextUpdate(time), "Update needed due to stale metadata.");
    }