@Test
    public void testMetadataTopicExpiry() throws Exception {
        long offset = 0;
        client.updateMetadata(RequestTestUtils.metadataUpdateWith(1, Collections.singletonMap("test", 2)));

        Future<RecordMetadata> future = appendToAccumulator(tp0);
        sender.runOnce();
        assertTrue(metadata.containsTopic(tp0.topic()), "Topic not added to metadata");
        client.updateMetadata(RequestTestUtils.metadataUpdateWith(1, Collections.singletonMap("test", 2)));
        sender.runOnce();  // send produce request
        client.respond(produceResponse(tp0, offset, Errors.NONE, 0));
        sender.runOnce();
        assertEquals(0, client.inFlightRequestCount(), "Request completed.");
        assertFalse(client.hasInFlightRequests());
        assertEquals(0, sender.inFlightBatches(tp0).size());
        sender.runOnce();
        assertTrue(future.isDone(), "Request should be completed");

        assertTrue(metadata.containsTopic(tp0.topic()), "Topic not retained in metadata list");
        time.sleep(TOPIC_IDLE_MS);
        client.updateMetadata(RequestTestUtils.metadataUpdateWith(1, Collections.singletonMap("test", 2)));
        assertFalse(metadata.containsTopic(tp0.topic()), "Unused topic has not been expired");
        future = appendToAccumulator(tp0);
        sender.runOnce();
        assertTrue(metadata.containsTopic(tp0.topic()), "Topic not added to metadata");
        client.updateMetadata(RequestTestUtils.metadataUpdateWith(1, Collections.singletonMap("test", 2)));
        sender.runOnce();  // send produce request
        client.respond(produceResponse(tp0, offset + 1, Errors.NONE, 0));
        sender.runOnce();
        assertEquals(0, client.inFlightRequestCount(), "Request completed.");
        assertFalse(client.hasInFlightRequests());
        assertEquals(0, sender.inFlightBatches(tp0).size());
        sender.runOnce();
        assertTrue(future.isDone(), "Request should be completed");
    }
Tests that topics are added to the metadata list when messages are available to send
and expired if not used during a metadata refresh interval.
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java