NamespaceAwareCoordinatorStreamStore implements MetadataStore { @Override public void put(String key, byte[] value) { String coordinatorMessageKeyAsJson = getCoordinatorMessageKey(key); metadataStore.put(coordinatorMessageKeyAsJson, value); } 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 testPutShouldDelegateTheInvocationToUnderlyingStore() { String value = RandomStringUtils.randomAlphabetic(5); String namespacedKey = CoordinatorStreamStore.serializeCoordinatorMessageKeyToJson(namespace, KEY1); Mockito.doNothing().when(coordinatorStreamStore).put(namespacedKey, value.getBytes(StandardCharsets.UTF_8)); namespaceAwareCoordinatorStreamStore.put(KEY1, value.getBytes(StandardCharsets.UTF_8)); Mockito.verify(coordinatorStreamStore).put(namespacedKey, value.getBytes(StandardCharsets.UTF_8)); }