ValidationsAggregatorService implements Service { @Override public void start(String bootstrapServers) { streams = aggregateOrderValidations(bootstrapServers, "/tmp/kafka-streams"); streams.cleanUp(); streams.start(); log.info("Started Service " + getClass().getSimpleName()); }  @Override void start(String bootstrapServers); @Override void stop(); static void main(String[] args);  }
@Test public void shouldAggregateRuleSuccesses() throws Exception { ordersService = new ValidationsAggregatorService(); orders = asList( new Order(id(0L), 0L, CREATED, UNDERPANTS, 3, 5.00d), new Order(id(1L), 0L, CREATED, JUMPERS, 1, 75.00d) ); sendOrders(orders); ruleResults = asList( new OrderValidation(id(0L), OrderValidationType.FRAUD_CHECK, OrderValidationResult.PASS), new OrderValidation(id(0L), OrderValidationType.ORDER_DETAILS_CHECK, OrderValidationResult.PASS), new OrderValidation(id(0L), OrderValidationType.INVENTORY_CHECK, OrderValidationResult.PASS), new OrderValidation(id(1L), OrderValidationType.FRAUD_CHECK, OrderValidationResult.PASS), new OrderValidation(id(1L), OrderValidationType.ORDER_DETAILS_CHECK, OrderValidationResult.FAIL), new OrderValidation(id(1L), OrderValidationType.INVENTORY_CHECK, OrderValidationResult.PASS) ); sendOrderValuations(ruleResults); ordersService.start(CLUSTER.bootstrapServers()); List<KeyValue<String, Order>> finalOrders = MicroserviceTestUtils .readKeyValues(Topics.ORDERS, 4, CLUSTER.bootstrapServers()); assertThat(finalOrders.size()).isEqualTo(4); assertThat(finalOrders.stream().map(kv -> kv.value).collect(Collectors.toList())).contains( new Order(id(0L), 0L, VALIDATED, UNDERPANTS, 3, 5.00d), new Order(id(1L), 0L, FAILED, JUMPERS, 1, 75.00d) ); }