CoordinatorStreamSystemProducer { public CoordinatorStreamSystemProducer(Config config, MetricsRegistry registry) { SystemStream coordinatorSystemStream = CoordinatorStreamUtil.getCoordinatorSystemStream(config); SystemFactory systemFactory = CoordinatorStreamUtil.getCoordinatorSystemFactory(config); SystemAdmin systemAdmin = systemFactory.getAdmin(coordinatorSystemStream.getSystem(), config, this.getClass().getSimpleName()); SystemProducer systemProducer = systemFactory.getProducer(coordinatorSystemStream.getSystem(), config, registry, this.getClass().getSimpleName()); this.systemStream = coordinatorSystemStream; this.systemProducer = systemProducer; this.systemAdmin = systemAdmin; this.keySerde = new JsonSerde<>(); this.messageSerde = new JsonSerde<>(); } CoordinatorStreamSystemProducer(Config config, MetricsRegistry registry);  CoordinatorStreamSystemProducer(SystemStream systemStream, SystemProducer systemProducer, SystemAdmin systemAdmin); void register(String source); void start(); void stop(); void send(CoordinatorStreamMessage message); void writeConfig(String source, Config config);  }
@Test public void testCoordinatorStreamSystemProducer() { MockCoordinatorStreamSystemFactory.enableMockConsumerCache(); String source = "source"; SystemStream systemStream = new SystemStream("system", "stream"); MockCoordinatorSystemProducer systemProducer = new MockCoordinatorSystemProducer(source); MockSystemAdmin systemAdmin = new MockSystemAdmin(); CoordinatorStreamSystemProducer producer = new CoordinatorStreamSystemProducer(systemStream, systemProducer, systemAdmin); SetConfig setConfig1 = new SetConfig(source, "job.name", "my-job-name"); SetConfig setConfig2 = new SetConfig(source, "job.id", "1234"); Delete delete = new Delete(source, "job.name", SetConfig.TYPE); assertFalse(systemProducer.isRegistered()); producer.register(source); assertTrue(systemProducer.isRegistered()); assertFalse(systemProducer.isStarted()); producer.start(); assertTrue(systemProducer.isStarted()); producer.send(setConfig1); producer.send(setConfig2); producer.send(delete); assertFalse(systemProducer.isStopped()); producer.stop(); assertTrue(systemProducer.isStopped()); List<OutgoingMessageEnvelope> envelopes = systemProducer.getEnvelopes(); OutgoingMessageEnvelope envelope0 = envelopes.get(0); OutgoingMessageEnvelope envelope1 = envelopes.get(1); OutgoingMessageEnvelope envelope2 = envelopes.get(2); TypeReference<Object[]> keyRef = new TypeReference<Object[]>() { }; TypeReference<Map<String, Object>> msgRef = new TypeReference<Map<String, Object>>() { }; assertEquals(3, envelopes.size()); assertEquals(new CoordinatorStreamMessage(setConfig1), new CoordinatorStreamMessage(deserialize((byte[]) envelope0.getKey(), keyRef), deserialize((byte[]) envelope0.getMessage(), msgRef))); assertEquals(new CoordinatorStreamMessage(setConfig2), new CoordinatorStreamMessage(deserialize((byte[]) envelope1.getKey(), keyRef), deserialize((byte[]) envelope1.getMessage(), msgRef))); assertEquals(new CoordinatorStreamMessage(delete), new CoordinatorStreamMessage(deserialize((byte[]) envelope2.getKey(), keyRef), deserialize((byte[]) envelope2.getMessage(), msgRef))); }