public synchronized Collection<StreamsMetadata> getAllMetadataForStore(final String storeName) {
        Objects.requireNonNull(storeName, "storeName cannot be null");
        if (topologyMetadata.hasNamedTopologies()) {
            throw new IllegalArgumentException("Cannot invoke the getAllMetadataForStore(storeName) method when"
                                                   + "using named topologies, please use the overload that accepts"
                                                   + "a topologyName parameter to identify the correct store");
        }

        if (!isInitialized()) {
            return Collections.emptyList();
        }

        if (globalStores.contains(storeName)) {
            return allMetadata;
        }

        final Collection<String> sourceTopics = topologyMetadata.sourceTopicsForStore(storeName, null);
        if (sourceTopics.isEmpty()) {
            return Collections.emptyList();
        }

        final ArrayList<StreamsMetadata> results = new ArrayList<>();
        for (final StreamsMetadata metadata : allMetadata) {
            if (metadata.stateStoreNames().contains(storeName) || metadata.standbyStateStoreNames().contains(storeName)) {
                results.add(metadata);
            }
        }
        return results;
    }
@Test
    public void shouldGetInstancesForStoreName() {
        final StreamsMetadata one = new StreamsMetadataImpl(hostOne,
            mkSet(globalTable, "table-one", "table-two", "merged-table"),
            mkSet(topic1P0, topic2P1, topic4P0),
            mkSet("table-one", "table-two", "merged-table"),
            mkSet(topic2P0, topic1P1));
        final StreamsMetadata two = new StreamsMetadataImpl(hostTwo,
            mkSet(globalTable, "table-two", "table-one", "merged-table"),
            mkSet(topic2P0, topic1P1),
            mkSet("table-three"),
            mkSet(topic3P0));
        final Collection<StreamsMetadata> actual = metadataState.getAllMetadataForStore("table-one");
        final Map<HostInfo, StreamsMetadata> actualAsMap = actual.stream()
            .collect(Collectors.toMap(StreamsMetadata::hostInfo, Function.identity()));
        assertEquals(3, actual.size());
        assertTrue(actual.contains(one), "expected " + actual + " to contain " + one);
        assertTrue(actual.contains(two), "expected " + actual + " to contain " + two);
        assertTrue(actualAsMap.get(hostThree).standbyStateStoreNames().contains("table-one"),
            "expected " + hostThree + " to contain as standby");
    }