NamespaceAwareCoordinatorStreamStore implements MetadataStore { @Override public void putAll(Map<String, byte[]> entries) { Map<String, byte[]> mapWithCoordinatorMessageKeys = entries.entrySet().stream() .collect(Collectors.toMap(e -> getCoordinatorMessageKey(e.getKey()), e -> e.getValue())); metadataStore.putAll(mapWithCoordinatorMessageKeys); } 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 testPutAllShouldDelegateTheInvocationToUnderlyingStore() { byte[] value = RandomStringUtils.randomAlphabetic(5).getBytes(StandardCharsets.UTF_8); ImmutableMap<String, byte[]> map = ImmutableMap.of( "key1", value, "key2", value, "key3", value, "key4", value, "key5", value); ImmutableMap<String, byte[]> namespacedMap = ImmutableMap.of( CoordinatorStreamStore.serializeCoordinatorMessageKeyToJson(namespace, "key1"), value, CoordinatorStreamStore.serializeCoordinatorMessageKeyToJson(namespace, "key2"), value, CoordinatorStreamStore.serializeCoordinatorMessageKeyToJson(namespace, "key3"), value, CoordinatorStreamStore.serializeCoordinatorMessageKeyToJson(namespace, "key4"), value, CoordinatorStreamStore.serializeCoordinatorMessageKeyToJson(namespace, "key5"), value); Mockito.doNothing().when(coordinatorStreamStore).putAll(namespacedMap); namespaceAwareCoordinatorStreamStore.putAll(map); Mockito.verify(coordinatorStreamStore).putAll(namespacedMap); }