@Test void should_storeDependencies() {

  // Given: configs
  Duration dependenciesRetentionPeriod = Duration.ofMinutes(1);
  Duration dependenciesWindowSize = Duration.ofMillis(100);
  // When: topology created
  Topology topology = new DependencyStorageTopology(
      dependencyTopic,
      dependenciesRetentionPeriod,
      dependenciesWindowSize,
      true).get();
  TopologyDescription description = topology.describe();
  // Then: topology with 1 thread
  assertThat(description.subtopologies()).hasSize(1);
  // Given: streams configuration
  TopologyTestDriver testDriver = new TopologyTestDriver(topology, props);
  // When: a trace is passed
  ConsumerRecordFactory<String, DependencyLink> factory =
      new ConsumerRecordFactory<>(dependencyTopic, new StringSerializer(),
          dependencyLinkSerde.serializer());
  DependencyLink dependencyLink = DependencyLink.newBuilder()
      .parent("svc_a").child("svc_b").callCount(1).errorCount(0)
      .build();
  String dependencyLinkId = "svc_a:svc_b";
  testDriver.pipeInput(factory.create(dependencyTopic, dependencyLinkId, dependencyLink, 10L));
  WindowStore<String, DependencyLink> links = testDriver.getWindowStore(DEPENDENCIES_STORE_NAME);
  // Then: dependency link created
  WindowStoreIterator<DependencyLink> firstLink = links.fetch(dependencyLinkId, 0L, 100L);
  assertThat(firstLink).hasNext();
  assertThat(firstLink.next().value).isEqualTo(dependencyLink);
  // When: new links appear
  testDriver.pipeInput(factory.create(dependencyTopic, dependencyLinkId, dependencyLink, 90L));
  // Then: dependency link increases
  WindowStoreIterator<DependencyLink> secondLink = links.fetch(dependencyLinkId, 0L, 100L);
  assertThat(secondLink).hasNext();
  assertThat(secondLink.next().value.callCount()).isEqualTo(2);
  // When: time moves forward
  testDriver.advanceWallClockTime(dependenciesRetentionPeriod.toMillis() + 91L);
  testDriver.pipeInput(factory.create(dependencyTopic, dependencyLinkId, dependencyLink));
  // Then: dependency link is removed and restarted
  KeyValueIterator<Windowed<String>, DependencyLink> thirdLink = links.all();
  assertThat(thirdLink).hasNext();
  assertThat(thirdLink.next().value.callCount()).isEqualTo(1);
  // Close resources
  testDriver.close();
  dependencyLinkSerde.close();
}