DependencyStorageTopology implements Supplier<Topology> { @Override public Topology get() { StreamsBuilder builder = new StreamsBuilder(); if (dependencyQueryEnabled) { builder.addStateStore( Stores.windowStoreBuilder( Stores.persistentWindowStore( DEPENDENCIES_STORE_NAME, dependencyTtl, dependencyWindowSize, false), Serdes.String(), dependencyLinkSerde ).withLoggingDisabled()); builder.stream(dependencyTopic, Consumed.with(Serdes.String(), dependencyLinkSerde)) .process(() -> new Processor<String, DependencyLink>() { ProcessorContext context; WindowStore<String, DependencyLink> dependenciesStore; @SuppressWarnings("unchecked") @Override public void init(ProcessorContext context) { this.context = context; dependenciesStore = (WindowStore<String, DependencyLink>) context.getStateStore( DEPENDENCIES_STORE_NAME); } @Override public void process(String linkKey, DependencyLink link) { Instant now = Instant.ofEpochMilli(context.timestamp()); Instant from = now.minus(dependencyWindowSize); try (WindowStoreIterator<DependencyLink> currentLinkWindow = dependenciesStore.fetch(linkKey, from, now)) { KeyValue<Long, DependencyLink> windowAndValue = null; if (currentLinkWindow.hasNext()) windowAndValue = currentLinkWindow.next(); if (currentLinkWindow.hasNext()) windowAndValue = currentLinkWindow.next(); if (windowAndValue != null) { DependencyLink currentLink = windowAndValue.value; DependencyLink aggregated = currentLink.toBuilder() .callCount(currentLink.callCount() + link.callCount()) .errorCount(currentLink.errorCount() + link.errorCount()) .build(); dependenciesStore.put(linkKey, aggregated, windowAndValue.key); } else { dependenciesStore.put(linkKey, link); } } } @Override public void close() { } }, DEPENDENCIES_STORE_NAME); } return builder.build(); } DependencyStorageTopology(
      String dependencyTopic,
      Duration dependencyTtl,
      Duration dependencyWindowSize,
      boolean dependencyQueryEnabled
  ); @Override Topology get(); static final String DEPENDENCIES_STORE_NAME; }
@Test void should_doNothing_whenDisabled() { Duration dependenciesRetentionPeriod = Duration.ofMinutes(1); Duration dependenciesWindowSize = Duration.ofMillis(100); Topology topology = new DependencyStorageTopology( dependencyTopic, dependenciesRetentionPeriod, dependenciesWindowSize, false).get(); TopologyDescription description = topology.describe(); assertThat(description.subtopologies()).hasSize(0); TopologyTestDriver testDriver = new TopologyTestDriver(topology, props); testDriver.close(); }
@Test void should_storeDependencies() { Duration dependenciesRetentionPeriod = Duration.ofMinutes(1); Duration dependenciesWindowSize = Duration.ofMillis(100); Topology topology = new DependencyStorageTopology( dependencyTopic, dependenciesRetentionPeriod, dependenciesWindowSize, true).get(); TopologyDescription description = topology.describe(); assertThat(description.subtopologies()).hasSize(1); TopologyTestDriver testDriver = new TopologyTestDriver(topology, props); 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); WindowStoreIterator<DependencyLink> firstLink = links.fetch(dependencyLinkId, 0L, 100L); assertThat(firstLink).hasNext(); assertThat(firstLink.next().value).isEqualTo(dependencyLink); testDriver.pipeInput(factory.create(dependencyTopic, dependencyLinkId, dependencyLink, 90L)); WindowStoreIterator<DependencyLink> secondLink = links.fetch(dependencyLinkId, 0L, 100L); assertThat(secondLink).hasNext(); assertThat(secondLink.next().value.callCount()).isEqualTo(2); testDriver.advanceWallClockTime(dependenciesRetentionPeriod.toMillis() + 91L); testDriver.pipeInput(factory.create(dependencyTopic, dependencyLinkId, dependencyLink)); KeyValueIterator<Windowed<String>, DependencyLink> thirdLink = links.all(); assertThat(thirdLink).hasNext(); assertThat(thirdLink.next().value.callCount()).isEqualTo(1); testDriver.close(); dependencyLinkSerde.close(); }