@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(); }
@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 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 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
  ); }
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(); }
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; }