KAFKA-5792; Fix Transient failure in KafkaAdminClientTest.testHandleTimeout

Author: Colin P. Mccabe <cmccabe@confluent.io>

Reviewers: Ismael Juma <ismael@juma.me.uk>

Closes #3822 from cmccabe/KAFKA-5792

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

@@ -370,6 +370,10 @@ public class KafkaAdminClient extends AdminClient {
         thread.start();
     }
 
+    Time time() {
+        return time;
+    }
+
     @Override
     public void close(long duration, TimeUnit unit) {
         long waitTimeMs = unit.toMillis(duration);

clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java

@@ -324,6 +324,8 @@ public class KafkaAdminClientTest {
             env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
             env.kafkaClient().prepareMetadataUpdate(env.cluster(), Collections.<String>emptySet());
             env.kafkaClient().setNode(nodes.get(0));
+            assertEquals(time, env.time());
+            assertEquals(env.time(), ((KafkaAdminClient) env.adminClient()).time());
 
             // Make a request with an extremely short timeout.
             // Then wait for it to fail by not supplying any response.

clients/src/test/java/org/apache/kafka/clients/admin/MockKafkaAdminClientEnv.java

@@ -52,7 +52,7 @@ public class MockKafkaAdminClientEnv implements AutoCloseable {
     }
 
     public MockKafkaAdminClientEnv(Time time, Cluster cluster, Map<String, Object> config) {
-        this.time = Time.SYSTEM;
+        this.time = time;
         this.adminClientConfig = new AdminClientConfig(config);
         this.cluster = cluster;
         this.metadata = new Metadata(adminClientConfig.getLong(AdminClientConfig.RETRY_BACKOFF_MS_CONFIG),