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();  }
@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(); }
@Test void should_aggregateSpans_and_mapDependencies() { Duration traceTimeout = Duration.ofSeconds(1); SpansSerde spansSerde = new SpansSerde(); DependencyLinkSerde dependencyLinkSerde = new DependencyLinkSerde(); Topology topology = new SpanAggregationTopology( spansTopic, traceTopic, dependencyTopic, traceTimeout, true).get(); TopologyDescription description = topology.describe(); assertThat(description.subtopologies()).hasSize(1); TopologyTestDriver testDriver = new TopologyTestDriver(topology, props); ConsumerRecordFactory<String, List<Span>> factory = new ConsumerRecordFactory<>(spansTopic, new StringSerializer(), spansSerde.serializer()); Span a = Span.newBuilder().traceId("a").id("a").name("op_a").kind(Span.Kind.CLIENT) .localEndpoint(Endpoint.newBuilder().serviceName("svc_a").build()) .build(); Span b = Span.newBuilder().traceId("a").id("b").name("op_b").kind(Span.Kind.SERVER) .localEndpoint(Endpoint.newBuilder().serviceName("svc_b").build()) .build(); testDriver.pipeInput( factory.create(spansTopic, a.traceId(), Collections.singletonList(a), 0L)); testDriver.pipeInput( factory.create(spansTopic, b.traceId(), Collections.singletonList(b), 0L)); Span c = Span.newBuilder().traceId("c").id("c").build(); testDriver.pipeInput(factory.create(spansTopic, c.traceId(), Collections.singletonList(c), traceTimeout.toMillis() + 1)); ProducerRecord<String, List<Span>> trace = testDriver.readOutput(traceTopic, new StringDeserializer(), spansSerde.deserializer()); assertThat(trace).isNotNull(); OutputVerifier.compareKeyValue(trace, a.traceId(), Arrays.asList(a, b)); ProducerRecord<String, DependencyLink> linkRecord = testDriver.readOutput(dependencyTopic, new StringDeserializer(), dependencyLinkSerde.deserializer()); assertThat(linkRecord).isNotNull(); DependencyLink link = DependencyLink.newBuilder() .parent("svc_a").child("svc_b").callCount(1).errorCount(0) .build(); OutputVerifier.compareKeyValue(linkRecord, "svc_a:svc_b", link); testDriver.close(); spansSerde.close(); dependencyLinkSerde.close(); }