KAFKA-2599: Fix Metadata.getClusterForCurrentTopics throws NPE

…h null checking

Author: Edward Ribeiro <edward.ribeiro@gmail.com>

Reviewers: Ismael Juma, Guozhang Wang

Closes #262 from eribeiro/KAFKA-2599

clients/src/main/java/org/apache/kafka/clients/Metadata.java

@@ -14,10 +14,12 @@ package org.apache.kafka.clients;
 
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
 import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Node;
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.errors.TimeoutException;
 import org.slf4j.Logger;
@@ -237,11 +239,13 @@ public final class Metadata {
 
     private Cluster getClusterForCurrentTopics(Cluster cluster) {
         Collection<PartitionInfo> partitionInfos = new ArrayList<>();
+        List<Node> nodes = Collections.emptyList();
         if (cluster != null) {
             for (String topic : this.topics) {
                 partitionInfos.addAll(cluster.partitionsForTopic(topic));
             }
+            nodes = cluster.nodes();
         }
-        return new Cluster(cluster.nodes(), partitionInfos);
+        return new Cluster(nodes, partitionInfos);
     }
 }

clients/src/test/java/org/apache/kafka/clients/MetadataTest.java

@@ -111,6 +111,9 @@ public class MetadataTest {
         assertEquals(100, metadata.timeToNextUpdate(1100));
         assertEquals(100, metadata.lastSuccessfulUpdate());
 
+        metadata.needMetadataForAllTopics(true);
+        metadata.update(null, time);
+        assertEquals(100, metadata.timeToNextUpdate(1000));
     }
 
     @Test