CoordinatorStreamSystemConsumer { public CoordinatorStreamSystemConsumer(Config config, MetricsRegistry registry) { SystemStream coordinatorSystemStream = CoordinatorStreamUtil.getCoordinatorSystemStream(config); SystemFactory systemFactory = CoordinatorStreamUtil.getCoordinatorSystemFactory(config); SystemAdmin systemAdmin = systemFactory.getAdmin(coordinatorSystemStream.getSystem(), config, this.getClass().getSimpleName()); SystemConsumer systemConsumer = systemFactory.getConsumer(coordinatorSystemStream.getSystem(), config, registry, this.getClass().getSimpleName()); this.coordinatorSystemStreamPartition = new SystemStreamPartition(coordinatorSystemStream, new Partition(0)); this.systemConsumer = systemConsumer; this.systemAdmin = systemAdmin; this.configMap = new HashMap<>(); this.isBootstrapped = false; this.keySerde = new JsonSerde<>(); this.messageSerde = new JsonSerde<>(); } CoordinatorStreamSystemConsumer(Config config, MetricsRegistry registry);  CoordinatorStreamSystemConsumer(SystemStream coordinatorSystemStream, SystemConsumer systemConsumer, SystemAdmin systemAdmin); void register(); void start(); void stop(); void bootstrap(); Set<CoordinatorStreamMessage> getBootstrappedStream(String type); Config getConfig(); SystemStreamPartitionIterator getStartIterator(); Set<CoordinatorStreamMessage> getUnreadMessages(SystemStreamPartitionIterator iterator); Set<CoordinatorStreamMessage> getUnreadMessages(SystemStreamPartitionIterator iterator, String type); boolean hasNewMessages(SystemStreamPartitionIterator iterator);  }
@Test public void testCoordinatorStreamSystemConsumer() { Map<String, String> expectedConfig = new LinkedHashMap<String, String>(); expectedConfig.put("job.id", "1234"); SystemStream systemStream = new SystemStream("system", "stream"); MockSystemConsumer systemConsumer = new MockSystemConsumer(new SystemStreamPartition(systemStream, new Partition(0))); CoordinatorStreamSystemConsumer consumer = new CoordinatorStreamSystemConsumer(systemStream, systemConsumer, new SinglePartitionWithoutOffsetsSystemAdmin()); assertEquals(0, systemConsumer.getRegisterCount()); consumer.register(); assertEquals(1, systemConsumer.getRegisterCount()); assertFalse(systemConsumer.isStarted()); consumer.start(); assertTrue(systemConsumer.isStarted()); try { consumer.getConfig(); fail("Should have failed when retrieving config before bootstrapping."); } catch (SamzaException e) { } consumer.bootstrap(); assertEquals(expectedConfig, consumer.getConfig()); assertFalse(systemConsumer.isStopped()); consumer.stop(); assertTrue(systemConsumer.isStopped()); }