source: class InventoryServiceTest  {
  
    private List<KeyValue<Product, Integer>> inventory;
    private List<Order> orders;
    private List<OrderValidation> expected;
    private InventoryService inventoryService;
  
    @Test
    public void shouldProcessOrdersWithSufficientStockAndRejectOrdersWithInsufficientStock()
        throws Exception {
  
target: 
      //Given
      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);
  
      //When
      inventoryService.start(CLUSTER.bootstrapServers());
  
      //Then the final order for Jumpers should have been 'rejected' as it's out of stock
      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);
  
      //And the reservations should have been incremented twice, once for each validated order
      List<KeyValue<Product, Long>> inventoryChangelog = readInventoryStateStore(2);
      assertThat(inventoryChangelog).isEqualTo(asList(
          new KeyValue<>(UNDERPANTS.toString(), 3L),
          new KeyValue<>(JUMPERS.toString(), 1L)
      ));
    }
  
  }