FraudService implements Service { @Override public void start(String bootstrapServers) { streams = processStreams(bootstrapServers, "/tmp/kafka-streams"); streams.cleanUp(); streams.start(); log.info("Started Service " + getClass().getSimpleName()); }  @Override void start(String bootstrapServers); static void main(String[] args); @Override void stop();  }
@Test public void shouldValidateWhetherOrderAmountExceedsFraudLimitOverWindow() throws Exception { fraudService = new FraudService(); orders = asList( new Order(id(0L), 0L, CREATED, UNDERPANTS, 3, 5.00d), new Order(id(1L), 0L, CREATED, JUMPERS, 1, 75.00d), new Order(id(2L), 1L, CREATED, JUMPERS, 1, 75.00d), new Order(id(3L), 1L, CREATED, JUMPERS, 1, 75.00d), new Order(id(4L), 1L, CREATED, JUMPERS, 50, 75.00d), new Order(id(5L), 2L, CREATED, UNDERPANTS, 10, 100.00d), new Order(id(6L), 2L, CREATED, UNDERPANTS, 10, 100.00d), new Order(id(7L), 2L, CREATED, UNDERPANTS, 1, 5.00d) ); sendOrders(orders); fraudService.start(CLUSTER.bootstrapServers()); expected = asList( new OrderValidation(id(0L), FRAUD_CHECK, PASS), new OrderValidation(id(1L), FRAUD_CHECK, PASS), new OrderValidation(id(2L), FRAUD_CHECK, PASS), new OrderValidation(id(3L), FRAUD_CHECK, PASS), new OrderValidation(id(4L), FRAUD_CHECK, FAIL), new OrderValidation(id(5L), FRAUD_CHECK, PASS), new OrderValidation(id(6L), FRAUD_CHECK, FAIL), new OrderValidation(id(7L), FRAUD_CHECK, FAIL) ); List<OrderValidation> read = read(Topics.ORDER_VALIDATIONS, 8, CLUSTER.bootstrapServers()); assertThat(read).isEqualTo(expected); }