### Question:
StreamsMetadataState { public synchronized <K> StreamsMetadata getMetadataWithKey(final String storeName, final K key, final Serializer<K> keySerializer) { Objects.requireNonNull(keySerializer, "keySerializer can't be null"); Objects.requireNonNull(storeName, "storeName can't be null"); Objects.requireNonNull(key, "key can't be null"); if (!isInitialized()) { return StreamsMetadata.NOT_AVAILABLE; } if (globalStores.contains(storeName)) { if (thisHost == UNKNOWN_HOST) { return allMetadata.get(0); } return myMetadata; } final SourceTopicsInfo sourceTopicsInfo = getSourceTopicsInfo(storeName); if (sourceTopicsInfo == null) { return null; } return getStreamsMetadataForKey(storeName, key, new DefaultStreamPartitioner<>(keySerializer, clusterMetadata, sourceTopicsInfo.topicWithMostPartitions), sourceTopicsInfo); } StreamsMetadataState(final TopologyBuilder builder, final HostInfo thisHost); synchronized Collection<StreamsMetadata> getAllMetadata(); synchronized Collection<StreamsMetadata> getAllMetadataForStore(final String storeName); synchronized StreamsMetadata getMetadataWithKey(final String storeName,
                                                               final K key,
                                                               final Serializer<K> keySerializer); synchronized StreamsMetadata getMetadataWithKey(final String storeName,
                                                               final K key,
                                                               final StreamPartitioner<? super K, ?> partitioner); synchronized void onChange(final Map<HostInfo, Set<TopicPartition>> currentState, final Cluster clusterMetadata); static final HostInfo UNKNOWN_HOST; }### Answer:
@Test public void shouldReturnNullOnGetWithKeyWhenStoreDoesntExist() throws Exception { final StreamsMetadata actual = discovery.getMetadataWithKey("not-a-store", "key", Serdes.String().serializer()); assertNull(actual); }
@Test(expected = NullPointerException.class) public void shouldThrowWhenKeyIsNull() throws Exception { discovery.getMetadataWithKey("table-three", null, Serdes.String().serializer()); }
@Test(expected = NullPointerException.class) public void shouldThrowWhenSerializerIsNull() throws Exception { discovery.getMetadataWithKey("table-three", "key", (Serializer) null); }
@Test(expected = NullPointerException.class) public void shouldThrowIfStoreNameIsNull() throws Exception { discovery.getMetadataWithKey(null, "key", Serdes.String().serializer()); }
@SuppressWarnings("unchecked") @Test(expected = NullPointerException.class) public void shouldThrowIfStreamPartitionerIsNull() throws Exception { discovery.getMetadataWithKey(null, "key", (StreamPartitioner) null); }
@Test public void shouldGetMyMetadataForGlobalStoreWithKey() throws Exception { final StreamsMetadata metadata = discovery.getMetadataWithKey(globalTable, "key", Serdes.String().serializer()); assertEquals(hostOne, metadata.hostInfo()); }
@Test public void shouldGetMyMetadataForGlobalStoreWithKeyAndPartitioner() throws Exception { final StreamsMetadata metadata = discovery.getMetadataWithKey(globalTable, "key", partitioner); assertEquals(hostOne, metadata.hostInfo()); }