InventoryService 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); @Override void stop(); static void main(String[] args); static final String INVENTORY_SERVICE_APP_ID; static final String RESERVED_STOCK_STORE_NAME; }
@Test public void shouldProcessOrdersWithSufficientStockAndRejectOrdersWithInsufficientStock() throws Exception { inventoryService = new InventoryService(); inventory = asList( new KeyValue<>(UNDERPANTS, 75), new KeyValue<>(JUMPERS, 1) ); sendInventory(inventory, Topics.WAREHOUSE_INVENTORY); orders = asList( new Order(id(0L), 1L, CREATED, UNDERPANTS, 3, 10.00d), new Order(id(1L), 2L, CREATED, JUMPERS, 1, 75.00d), new Order(id(2L), 2L, CREATED, JUMPERS, 1, 75.00d) ); sendOrders(orders); inventoryService.start(CLUSTER.bootstrapServers()); expected = asList( new OrderValidation(id(0L), OrderValidationType.INVENTORY_CHECK, OrderValidationResult.PASS), new OrderValidation(id(1L), OrderValidationType.INVENTORY_CHECK, OrderValidationResult.PASS), new OrderValidation(id(2L), OrderValidationType.INVENTORY_CHECK, OrderValidationResult.FAIL) ); assertThat(MicroserviceTestUtils .read(Topics.ORDER_VALIDATIONS, expected.size(), CLUSTER.bootstrapServers())) .isEqualTo(expected); List<KeyValue<Product, Long>> inventoryChangelog = readInventoryStateStore(2); assertThat(inventoryChangelog).isEqualTo(asList( new KeyValue<>(UNDERPANTS.toString(), 3L), new KeyValue<>(JUMPERS.toString(), 1L) )); }