HOTFIX: fix npe in StreamsMetadataState when onChange has not been called

If some StreamsMetadataState methods are called before the onChange method is called a NullPointerException was being thrown. Added null check for cluster in isInitialized method

Author: Damian Guy <damian.guy@gmail.com>

Reviewers: Guozhang Wang <wangguoz@gmail.com>

Closes #1920 from dguy/fix-npe-streamsmetadata

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsMetadataState.java

@@ -234,7 +234,7 @@ public class StreamsMetadataState {
     }
 
     private boolean isInitialized() {
-        return !clusterMetadata.topics().isEmpty();
+        return clusterMetadata != null && !clusterMetadata.topics().isEmpty();
     }
 
     private class SourceTopicsInfo {

streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsMetadataStateTest.java

@@ -114,6 +114,11 @@ public class StreamsMetadataStateTest {
         discovery.onChange(hostToPartitions, cluster);
     }
 
+    @Test
+    public void shouldNotThrowNPEWhenOnChangeNotCalled() throws Exception {
+        new StreamsMetadataState(builder).getAllMetadataForStore("store");
+    }
+
     @Test
     public void shouldGetAllStreamInstances() throws Exception {
         final StreamsMetadata one = new StreamsMetadata(hostOne, Utils.mkSet("table-one", "table-two", "merged-table"),