CoordinatorStreamStore implements MetadataStore { @Override public byte[] get(String namespacedKey) { readMessagesFromCoordinatorStream(); return messagesReadFromCoordinatorStream.get(namespacedKey); } CoordinatorStreamStore(Config config, MetricsRegistry metricsRegistry); @VisibleForTesting protected  CoordinatorStreamStore(Config config, SystemProducer systemProducer, SystemConsumer systemConsumer, SystemAdmin systemAdmin); @Override void init(); @Override byte[] get(String namespacedKey); @Override void put(String namespacedKey, byte[] value); @Override void delete(String namespacedKey); @Override Map<String, byte[]> all(); @Override void close(); @Override void flush(); static String serializeCoordinatorMessageKeyToJson(String type, String key); static CoordinatorMessageKey deserializeCoordinatorMessageKeyFromJson(String coordinatorMsgKeyAsJson);  }
@Test public void testPutAll() { CoordinatorStreamStore spyCoordinatorStreamStore = Mockito.spy(coordinatorStreamStore); String key1 = getCoordinatorMessageKey("test-key1"); String key2 = getCoordinatorMessageKey("test-key2"); String key3 = getCoordinatorMessageKey("test-key3"); String key4 = getCoordinatorMessageKey("test-key4"); String key5 = getCoordinatorMessageKey("test-key5"); byte[] value1 = getValue("test-value1"); byte[] value2 = getValue("test-value2"); byte[] value3 = getValue("test-value3"); byte[] value4 = getValue("test-value4"); byte[] value5 = getValue("test-value5"); ImmutableMap<String, byte[]> map = ImmutableMap.of(key1, value1, key2, value2, key3, value3, key4, value4, key5, value5); spyCoordinatorStreamStore.putAll(map); Assert.assertEquals(value1, spyCoordinatorStreamStore.get(key1)); Assert.assertEquals(value2, spyCoordinatorStreamStore.get(key2)); Assert.assertEquals(value3, spyCoordinatorStreamStore.get(key3)); Assert.assertEquals(value4, spyCoordinatorStreamStore.get(key4)); Assert.assertEquals(value5, spyCoordinatorStreamStore.get(key5)); }