@Test public void shouldNotThrowNPEWhenOnChangeNotCalled() throws Exception { new StreamsMetadataState(builder, hostOne).getAllMetadataForStore("store"); }
public synchronized Collection<StreamsMetadata> getAllMetadataForStore(final String storeName) { Objects.requireNonNull(storeName, "storeName cannot be null"); if (!isInitialized()) { return Collections.emptyList(); } if (globalStores.contains(storeName)) { return allMetadata; } final List<String> sourceTopics = builder.stateStoreNameToSourceTopics().get(storeName); if (sourceTopics == null) { return Collections.emptyList(); } final ArrayList<StreamsMetadata> results = new ArrayList<>(); for (StreamsMetadata metadata : allMetadata) { if (metadata.stateStoreNames().contains(storeName)) { results.add(metadata); } } return results; }
StreamsMetadataState { public synchronized Collection<StreamsMetadata> getAllMetadataForStore(final String storeName) { Objects.requireNonNull(storeName, "storeName cannot be null"); if (!isInitialized()) { return Collections.emptyList(); } if (globalStores.contains(storeName)) { return allMetadata; } final List<String> sourceTopics = builder.stateStoreNameToSourceTopics().get(storeName); if (sourceTopics == null) { return Collections.emptyList(); } final ArrayList<StreamsMetadata> results = new ArrayList<>(); for (StreamsMetadata metadata : allMetadata) { if (metadata.stateStoreNames().contains(storeName)) { results.add(metadata); } } return results; } }
StreamsMetadataState { public synchronized Collection<StreamsMetadata> getAllMetadataForStore(final String storeName) { Objects.requireNonNull(storeName, "storeName cannot be null"); if (!isInitialized()) { return Collections.emptyList(); } if (globalStores.contains(storeName)) { return allMetadata; } final List<String> sourceTopics = builder.stateStoreNameToSourceTopics().get(storeName); if (sourceTopics == null) { return Collections.emptyList(); } final ArrayList<StreamsMetadata> results = new ArrayList<>(); for (StreamsMetadata metadata : allMetadata) { if (metadata.stateStoreNames().contains(storeName)) { results.add(metadata); } } return results; } StreamsMetadataState(final TopologyBuilder builder, final HostInfo thisHost); }
StreamsMetadataState { public synchronized Collection<StreamsMetadata> getAllMetadataForStore(final String storeName) { Objects.requireNonNull(storeName, "storeName cannot be null"); if (!isInitialized()) { return Collections.emptyList(); } if (globalStores.contains(storeName)) { return allMetadata; } final List<String> sourceTopics = builder.stateStoreNameToSourceTopics().get(storeName); if (sourceTopics == null) { return Collections.emptyList(); } final ArrayList<StreamsMetadata> results = new ArrayList<>(); for (StreamsMetadata metadata : allMetadata) { if (metadata.stateStoreNames().contains(storeName)) { results.add(metadata); } } return results; } 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); }
StreamsMetadataState { public synchronized Collection<StreamsMetadata> getAllMetadataForStore(final String storeName) { Objects.requireNonNull(storeName, "storeName cannot be null"); if (!isInitialized()) { return Collections.emptyList(); } if (globalStores.contains(storeName)) { return allMetadata; } final List<String> sourceTopics = builder.stateStoreNameToSourceTopics().get(storeName); if (sourceTopics == null) { return Collections.emptyList(); } final ArrayList<StreamsMetadata> results = new ArrayList<>(); for (StreamsMetadata metadata : allMetadata) { if (metadata.stateStoreNames().contains(storeName)) { results.add(metadata); } } return results; } 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; }