EmailService implements Service { @Override public void start(String bootstrapServers) { streams = processStreams(bootstrapServers, "/tmp/kafka-streams"); streams.cleanUp(); streams.start(); log.info("Started Service " + APP_ID); } EmailService(Emailer emailer); @Override void start(String bootstrapServers); static void main(String[] args); @Override void stop();  }
@Test public void shouldSendEmailWithValidContents() throws Exception { String orderId = id(0L); Order order = new Order(orderId, 15L, CREATED, UNDERPANTS, 3, 5.00d); Customer customer = new Customer(15L, "Franz", "Kafka", "frans@thedarkside.net", "oppression street, prague, cze"); Payment payment = new Payment("Payment:1234", orderId, "CZK", 1000.00d); emailService = new EmailService(details -> { assertThat(details.customer).isEqualTo(customer); assertThat(details.payment).isEqualTo(payment); assertThat(details.order).isEqualTo(order); complete = true; }); send(Topics.CUSTOMERS, new KeyValue(customer.getId(), customer)); send(Topics.ORDERS, new KeyValue(order.getId(), order)); send(Topics.PAYMENTS, new KeyValue(payment.getId(), payment)); emailService.start(CLUSTER.bootstrapServers()); TestUtils.waitForCondition(() -> complete == true, "Email was never sent."); }