source: class TraceStorageTopologyTest {
  
     String spansTopic;
     Properties props;
  
    @Test void should_persistSpans_and_searchQueryTraces_whenAllEnabled() {
  
target:     // Given: configs
      Duration traceTtl = Duration.ofMillis(5);
      Duration traceTtlCheckInterval = Duration.ofMinutes(1);
      List<String> autocompleteKeys = Collections.singletonList("environment");
      SpansSerde spansSerde = new SpansSerde();
      // When: topology provided
      Topology topology = new TraceStorageTopology(
          spansTopic,
          autocompleteKeys,
          traceTtl,
          traceTtlCheckInterval,
          0,
          true,
          true).get();
      TopologyDescription description = topology.describe();
      // Then: 1 thread prepared
      assertThat(description.subtopologies()).hasSize(1);
      // Given: streams config
      TopologyTestDriver testDriver = new TopologyTestDriver(topology, props);
      // When: a trace is passed
      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())
          .timestamp(10000L).duration(11L)
          .putTag("environment", "dev")
          .build();
      Span b = Span.newBuilder().traceId("a").id("b").name("op_b").kind(Span.Kind.SERVER)
          .localEndpoint(Endpoint.newBuilder().serviceName("svc_b").build())
          .timestamp(10000L).duration(10L)
          .build();
      Span c = Span.newBuilder().traceId("c").id("c").name("op_a").kind(Span.Kind.CLIENT)
          .localEndpoint(Endpoint.newBuilder().serviceName("svc_a").build())
          .timestamp(10000L).duration(11L)
          .putTag("environment", "dev")
          .build();
      List<Span> spans = Arrays.asList(a, b, c);
      testDriver.pipeInput(factory.create(spansTopic, a.traceId(), spans, 10L));
      // Then: trace stores are filled
      KeyValueStore<String, List<Span>> traces = testDriver.getKeyValueStore(TRACES_STORE_NAME);
      assertThat(traces.get(a.traceId())).containsExactlyElementsOf(spans);
      KeyValueStore<Long, Set<String>> spanIdsByTs =
          testDriver.getKeyValueStore(SPAN_IDS_BY_TS_STORE_NAME);
      KeyValueIterator<Long, Set<String>> ids = spanIdsByTs.all();
      assertThat(ids).hasNext();
      assertThat(ids.next().value).containsExactly(a.traceId());
      // Then: service name stores are filled
      KeyValueStore<String, String> serviceNames =
          testDriver.getKeyValueStore(SERVICE_NAMES_STORE_NAME);
      List<String> serviceNameList = new ArrayList<>();
      serviceNames.all().forEachRemaining(serviceName -> serviceNameList.add(serviceName.value));
      assertThat(serviceNameList).hasSize(2);
      assertThat(serviceNames.get("svc_a")).isEqualTo("svc_a");
      assertThat(serviceNames.get("svc_b")).isEqualTo("svc_b");
      KeyValueStore<String, Set<String>> spanNames =
          testDriver.getKeyValueStore(SPAN_NAMES_STORE_NAME);
      assertThat(spanNames.get("svc_a")).containsExactly("op_a");
      assertThat(spanNames.get("svc_b")).containsExactly("op_b");
      KeyValueStore<String, Set<String>> autocompleteTags =
          testDriver.getKeyValueStore(AUTOCOMPLETE_TAGS_STORE_NAME);
      assertThat(autocompleteTags.get("environment")).containsExactly("dev");
      // When: clock moves forward
      Span d = Span.newBuilder()
          .traceId("d")
          .id("d")
          .timestamp(
              MILLISECONDS.toMicros(traceTtlCheckInterval.toMillis()) + MILLISECONDS.toMicros(20))
          .build();
      testDriver.pipeInput(
          factory.create(spansTopic, d.traceId(), Collections.singletonList(d),
              traceTtlCheckInterval.plusMillis(1).toMillis()));
      // Then: Traces store is empty
      assertThat(traces.get(a.traceId())).isNull();
      // Finally close resources
      testDriver.close();
      spansSerde.close();
    }
  
  }