NamespaceAwareCoordinatorStreamStore implements MetadataStore { @Override public Map<String, byte[]> all() { Map<String, byte[]> bootstrappedMessages = readMessagesFromCoordinatorStore(); return Collections.unmodifiableMap(bootstrappedMessages); } NamespaceAwareCoordinatorStreamStore(MetadataStore metadataStore, String namespace); @Override void init(); @Override byte[] get(String key); @Override void put(String key, byte[] value); @Override void putAll(Map<String, byte[]> entries); @Override void delete(String key); @Override Map<String, byte[]> all(); @Override void flush(); @Override void close();  }
@Test public void testAllShouldDelegateToUnderlyingMetadaStore() { String value = RandomStringUtils.randomAlphabetic(5); String key2 = "key2"; String key3 = "key3"; byte[] valueAsBytes = value.getBytes(StandardCharsets.UTF_8); String namespacedKey1 = CoordinatorStreamStore.serializeCoordinatorMessageKeyToJson(namespace, KEY1); String namespacedKey2 = CoordinatorStreamStore.serializeCoordinatorMessageKeyToJson("namespace1", key2); String namespacedKey3 = CoordinatorStreamStore.serializeCoordinatorMessageKeyToJson("namespace1", key3); Mockito.when(coordinatorStreamStore.all()).thenReturn(ImmutableMap.of(namespacedKey1, valueAsBytes, namespacedKey2, new byte[0], namespacedKey3, new byte[0])); Assert.assertEquals(ImmutableMap.of(KEY1, valueAsBytes), namespaceAwareCoordinatorStreamStore.all()); Mockito.verify(coordinatorStreamStore).all(); }