Developing Event-driven Applications with Apache Kafka and Red Hat AMQ Streams

Procedure 6.4. Instructions

1. Navigate to the troubleshooting-test directory in your workspace and examine the application code.

   • The inventory.VehicleInventory class generates a VehicleRegistered event when a user sends a POST request to the /vehicle endpoint. Note that the VehicleInventory class does not have references to Kafka.

   • The inventory.VehicleResource mediates between the Kafka client code and the VehicleInventory business logic. This class emits the events returned by the VehicleInventory class.

   • The movement.tracker.VehicleMovementsTracker class defines a Kafka Streams topology to process VehicleRegistered and VehicleMoved events. The topology saves registered vehicles in the vehicles-store store, and writes the results of processing movements to the vehicle-metrics-store store.

   • The test directory contains the test files. You will implement the tests in this directory.

2. Unit test the VehicleInventory#register method. The register method accepts an instance of Vehicle as a parameter, and registers the vehicle.

   Acceptance criteria:

   • The method should return a VehicleRegistered event if the vehicle type is not empty.

   • Otherwise, the method should throw a InvalidVehicleException exception.

   1. Edit the TestVehicleInventory class. Create a test case to verify that the register method throws an exception if the given vehicle type is an empty string.

      // TODO: implement exception test case
      @Test
      public void testRegisterThrowsExceptionIfTypeIsEmpty() {
          // Given
          Vehicle vehicle = new Vehicle(2, "", "Test Car");
      
          // When
          Exception exception = assertThrows(InvalidVehicleException.class, () -> {
              vehicleManager.register(vehicle);
          });
      
          // Then
          assertThat(exception.getMessage(), containsString("Invalid vehicle"));
      }

   2. Run the tests by executing the ./mvnw test command. One test should pass.

      (.venv) [user@host troubleshooting-test]$ ./mvnw test
      ...output omitted...
      [INFO] Results:
      [INFO]
      [INFO] Tests run: 1, Failures: 0, Errors: 0, Skipped: 0
      [INFO]
      [INFO] ------------------------------------------------------------------------
      [INFO] BUILD SUCCESS
      [INFO] ------------------------------------------------------------------------
      ...output omitted...

   3. In the same file, implement a test case to verify that the method emits a VehicleRegistered if the vehicle type is not empty.

      // TODO: implement VehicleRegistered test case
      @Test
      public void testRegisterReturnsVehicleRegistered()
              throws InvalidVehicleException {
          // Given
          Vehicle vehicle = new Vehicle(2, "car", "Test Car");
      
          // When
          VehicleRegistered event = vehicleManager.register(vehicle);
      
          // Then
          assertEquals(2, event.id);
          assertEquals("car", event.type);
          assertEquals("Test Car", event.model);
      }

   4. Run the tests with the ./mvnw test command. Two tests should pass.

3. Unit test the Kafka Streams topology specified by the VehicleMovementsTracker class.

   1. Edit the MovementTrackerTopologyTest class. In the setup method, create the TopologyTestDriver with the topology from the VehicleMovementTracker. The test driver runs the tests on this topology.

      // TODO: pass the VehicleMovementTracker topology to the test driver
      tracker = new VehicleMovementTracker();
      testDriver = new TopologyTestDriver(tracker.buildTopology());

   2. In the same setup method, create the input test topics, by specifying the topic names and the required serializers.

      // TODO: Create test input topics
      vehicleRegisteredSerde = new ObjectMapperSerde<>(VehicleRegistered.class);
      vehicleMovedSerde = new ObjectMapperSerde<>(VehicleMoved.class);
      vehicleRegisteredTopic = testDriver.createInputTopic(
          "vehicle-registered",
          new IntegerSerializer(),
          vehicleRegisteredSerde.serializer()
      );
      vehicleMovedTopic = testDriver.createInputTopic(
          "vehicle-moved",
          new IntegerSerializer(),
          vehicleMovedSerde.serializer()
      );

   3. In the same method, create the test stores for vehicles and metrics.

      // TODO: Create test stores
      vehiclesStore = testDriver.getKeyValueStore("vehicles-store");
      vehicleMetricsStore = testDriver.getKeyValueStore("vehicle-metrics-store");

   4. Implement a test to verify that the topology reads the VehicleRegistered event stream as a change log and writes the result into the vehicles store.

      // TODO: test VehicleRegistered events create vehicles in the store
      @Test
      public void testTopologySendsVehiclesToStore() throws JsonProcessingException {
          // Given
          VehicleRegistered event = new VehicleRegistered(12, "bike", "super bike");
      
          // When
          vehicleRegisteredTopic.pipeInput(event.id, event);
      
          // Then
          Vehicle vehicleInStore = vehiclesStore.get(12);
          assertEquals(vehicleInStore.id, 12);
      }

   5. Run the tests with the ./mvnw test command. Three tests should pass.

   6. Test that the result of joining and aggregating VehicleRegistered and VehicleMoved events is sent to the vehicle metrics store. Validate that the metrics gathered from the store contain the number of movements reported.

      // TODO: test VehicleMetrics counts VehicleMoved events for each vehicle
      @Test
      public void testVehicleMovementsCountEqualsEventsCount()
              throws JsonProcessingException {
          // Given
          Vehicle vehicle = new Vehicle(14, "car", "test");
          vehiclesStore.put(vehicle.id, vehicle);
          VehicleMoved event1 = new VehicleMoved(14, 0, 0, 5);
          VehicleMoved event2 = new VehicleMoved(14, 0, 0, 10);
      
          // When
          vehicleMovedTopic.pipeInput(event1.vehicleId, event1);
          vehicleMovedTopic.pipeInput(event2.vehicleId, event2);
      
          // Then
          VehicleMetrics metrics = vehicleMetricsStore.get(14);
          assertEquals(2, metrics.movementsReported);
      }

   7. Run the tests with the ./mvnw test command. Four tests should pass.

4. Create a new unit test for the VehicleMovementsTracker class. Test that the topology produces an enriched movements stream and sends the result to the vehicle-status topic. Verify that the test fails and add the necessary code in the VehicleMovementsTracker to make the test pass.

   1. Edit the MovementTrackerTopologyTest class. In the setup method, initialize the test output topic, called vehicle-status.

      // TODO: Create test output topic
      vehicleStatusSerde = new ObjectMapperSerde<>(VehicleStatus.class);
      vehicleStatusTopic = testDriver.createOutputTopic(
          "vehicle-status",
          new IntegerDeserializer(),
          vehicleStatusSerde.deserializer()
      );

      Note that, in contrast to test input topics, which require serializers, test output topics require deserializers.

   2. In the same file, implement a test to validate that, for each VehicleMoved event, the topology creates a new VehicleStatus value, which includes the movement and the vehicle details. Verify that the topology writes the new value to the vehicle-status topic.

      // TODO: test VehicleMoved and VehicleRegistered events are joined and sent to output topic
      @Test
      public void testVehicleMovementsStreamIsEnriched()
              throws JsonProcessingException {
          // Given
          VehicleRegistered vehicleRegistered = new VehicleRegistered(
              111, "car", "test car"
          );
          VehicleMoved vehicleMoved = new VehicleMoved(111, 0, 0, 5);
      
          // When
          vehicleRegisteredTopic.pipeInput(vehicleRegistered.id, vehicleRegistered);
          vehicleMovedTopic.pipeInput(vehicleMoved.vehicleId, vehicleMoved);
      
          // Then
          TestRecord<Integer, VehicleStatus> record = vehicleStatusTopic.readRecord();
          assertEquals(111, record.getKey());
          assertEquals(111, record.getValue().vehicle.id);
          assertEquals(5, record.getValue().elevation);
      }

   3. Run the tests with the ./mvnw test command. The last test should fail.

      (.venv) [user@host troubleshooting-test]$ ./mvnw test
      ...output omitted...
      [ERROR] Errors:
      [ERROR]   MovementTrackerTopologyTest.testVehicleMovementsStreamIsEnriched:126 » NoSuchElement
      [INFO]
      [ERROR] Tests run: 5, Failures: 0, Errors: 1, Skipped: 0
      [INFO]
      [INFO] ------------------------------------------------------------------------
      [INFO] BUILD FAILURE
      [INFO] ------------------------------------------------------------------------
      ...output omitted...

   4. Edit the VehicleMovementTracker class. Add the code to make the test pass.

      // TODO: materialize vehicleStatusStream to the "vehicle-status" topic
      vehicleStatusStream.to(
          "vehicle-status",
          Produced.with(intSerde, vehicleStatusSerde)
      );

   5. Run the tests with the ./mvnw test command. Five tests should pass.