LocalityManager { public LocalityManager(MetadataStore metadataStore) { this.metadataStore = metadataStore; this.valueSerde = new CoordinatorStreamValueSerde(SetContainerHostMapping.TYPE); } LocalityManager(MetadataStore metadataStore); LocalityModel readLocality(); void writeContainerToHostMapping(String processorId, String hostName); void close();  }
@Test public void testLocalityManager() { LocalityManager localityManager = new LocalityManager(new NamespaceAwareCoordinatorStreamStore(coordinatorStreamStore, SetContainerHostMapping.TYPE)); localityManager.writeContainerToHostMapping("0", "localhost"); Map<String, Map<String, String>> localMap = readContainerLocality(localityManager); Map<String, Map<String, String>> expectedMap = new HashMap<String, Map<String, String>>() { { this.put("0", new HashMap<String, String>() { { this.put(SetContainerHostMapping.HOST_KEY, "localhost"); } }); } }; assertEquals(expectedMap, localMap); localityManager.close(); MockCoordinatorStreamSystemProducer producer = coordinatorStreamStoreTestUtil.getMockCoordinatorStreamSystemProducer(); MockCoordinatorStreamSystemConsumer consumer = coordinatorStreamStoreTestUtil.getMockCoordinatorStreamSystemConsumer(); assertTrue(producer.isStopped()); assertTrue(consumer.isStopped()); }