@Test void should_doNothing_whenAggregationDisabled() { Duration traceTimeout = Duration.ofSeconds(1); Topology topology = new SpanAggregationTopology( spansTopic, traceTopic, dependencyTopic, traceTimeout, 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 (aggregationEnabled) { KStream<String, List<Span>> tracesStream = builder.stream(spansTopic, Consumed.with(Serdes.String(), spansSerde)) .groupByKey() .windowedBy(SessionWindows.with(traceTimeout).grace(Duration.ZERO)) .aggregate(ArrayList::new, aggregateSpans(), joinAggregates(), Materialized .<String, List<Span>>as( Stores.persistentSessionStore(TRACE_AGGREGATION_STORE, Duration.ofDays(1))) .withKeySerde(Serdes.String()) .withValueSerde(spansSerde) .withLoggingDisabled() .withCachingEnabled()) .suppress(untilWindowCloses(unbounded())) .toStream() .selectKey((windowed, spans) -> windowed.key()); tracesStream.to(traceTopic, Produced.with(Serdes.String(), spansSerde)); tracesStream.flatMapValues(spansToDependencyLinks()) .selectKey((key, value) -> linkKey(value)) .to(dependencyTopic, Produced.with(Serdes.String(), dependencyLinkSerde)); } return builder.build(); }
SpanAggregationTopology implements Supplier<Topology> { @Override public Topology get() { StreamsBuilder builder = new StreamsBuilder(); if (aggregationEnabled) { KStream<String, List<Span>> tracesStream = builder.stream(spansTopic, Consumed.with(Serdes.String(), spansSerde)) .groupByKey() .windowedBy(SessionWindows.with(traceTimeout).grace(Duration.ZERO)) .aggregate(ArrayList::new, aggregateSpans(), joinAggregates(), Materialized .<String, List<Span>>as( Stores.persistentSessionStore(TRACE_AGGREGATION_STORE, Duration.ofDays(1))) .withKeySerde(Serdes.String()) .withValueSerde(spansSerde) .withLoggingDisabled() .withCachingEnabled()) .suppress(untilWindowCloses(unbounded())) .toStream() .selectKey((windowed, spans) -> windowed.key()); tracesStream.to(traceTopic, Produced.with(Serdes.String(), spansSerde)); tracesStream.flatMapValues(spansToDependencyLinks()) .selectKey((key, value) -> linkKey(value)) .to(dependencyTopic, Produced.with(Serdes.String(), dependencyLinkSerde)); } return builder.build(); } }
SpanAggregationTopology implements Supplier<Topology> { @Override public Topology get() { StreamsBuilder builder = new StreamsBuilder(); if (aggregationEnabled) { KStream<String, List<Span>> tracesStream = builder.stream(spansTopic, Consumed.with(Serdes.String(), spansSerde)) .groupByKey() .windowedBy(SessionWindows.with(traceTimeout).grace(Duration.ZERO)) .aggregate(ArrayList::new, aggregateSpans(), joinAggregates(), Materialized .<String, List<Span>>as( Stores.persistentSessionStore(TRACE_AGGREGATION_STORE, Duration.ofDays(1))) .withKeySerde(Serdes.String()) .withValueSerde(spansSerde) .withLoggingDisabled() .withCachingEnabled()) .suppress(untilWindowCloses(unbounded())) .toStream() .selectKey((windowed, spans) -> windowed.key()); tracesStream.to(traceTopic, Produced.with(Serdes.String(), spansSerde)); tracesStream.flatMapValues(spansToDependencyLinks()) .selectKey((key, value) -> linkKey(value)) .to(dependencyTopic, Produced.with(Serdes.String(), dependencyLinkSerde)); } return builder.build(); } SpanAggregationTopology(
      String spansTopic,
      String traceTopic,
      String dependencyTopic,
      Duration traceTimeout,
      boolean aggregationEnabled
  ); }
SpanAggregationTopology implements Supplier<Topology> { @Override public Topology get() { StreamsBuilder builder = new StreamsBuilder(); if (aggregationEnabled) { KStream<String, List<Span>> tracesStream = builder.stream(spansTopic, Consumed.with(Serdes.String(), spansSerde)) .groupByKey() .windowedBy(SessionWindows.with(traceTimeout).grace(Duration.ZERO)) .aggregate(ArrayList::new, aggregateSpans(), joinAggregates(), Materialized .<String, List<Span>>as( Stores.persistentSessionStore(TRACE_AGGREGATION_STORE, Duration.ofDays(1))) .withKeySerde(Serdes.String()) .withValueSerde(spansSerde) .withLoggingDisabled() .withCachingEnabled()) .suppress(untilWindowCloses(unbounded())) .toStream() .selectKey((windowed, spans) -> windowed.key()); tracesStream.to(traceTopic, Produced.with(Serdes.String(), spansSerde)); tracesStream.flatMapValues(spansToDependencyLinks()) .selectKey((key, value) -> linkKey(value)) .to(dependencyTopic, Produced.with(Serdes.String(), dependencyLinkSerde)); } return builder.build(); } SpanAggregationTopology(
      String spansTopic,
      String traceTopic,
      String dependencyTopic,
      Duration traceTimeout,
      boolean aggregationEnabled
  ); @Override Topology get(); }
SpanAggregationTopology implements Supplier<Topology> { @Override public Topology get() { StreamsBuilder builder = new StreamsBuilder(); if (aggregationEnabled) { KStream<String, List<Span>> tracesStream = builder.stream(spansTopic, Consumed.with(Serdes.String(), spansSerde)) .groupByKey() .windowedBy(SessionWindows.with(traceTimeout).grace(Duration.ZERO)) .aggregate(ArrayList::new, aggregateSpans(), joinAggregates(), Materialized .<String, List<Span>>as( Stores.persistentSessionStore(TRACE_AGGREGATION_STORE, Duration.ofDays(1))) .withKeySerde(Serdes.String()) .withValueSerde(spansSerde) .withLoggingDisabled() .withCachingEnabled()) .suppress(untilWindowCloses(unbounded())) .toStream() .selectKey((windowed, spans) -> windowed.key()); tracesStream.to(traceTopic, Produced.with(Serdes.String(), spansSerde)); tracesStream.flatMapValues(spansToDependencyLinks()) .selectKey((key, value) -> linkKey(value)) .to(dependencyTopic, Produced.with(Serdes.String(), dependencyLinkSerde)); } return builder.build(); } SpanAggregationTopology(
      String spansTopic,
      String traceTopic,
      String dependencyTopic,
      Duration traceTimeout,
      boolean aggregationEnabled
  ); @Override Topology get();  }