@Test void should_persistSpans_and_onlyQueryTraces_whenEnabled() {

  // 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,
      false).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);
  assertThat(serviceNames).isNull();
  KeyValueStore<String, Set<String>> spanNames =
      testDriver.getKeyValueStore(SPAN_NAMES_STORE_NAME);
  assertThat(spanNames).isNull();
  KeyValueStore<String, Set<String>> autocompleteTags =
      testDriver.getKeyValueStore(AUTOCOMPLETE_TAGS_STORE_NAME);
  assertThat(autocompleteTags).isNull();
  // Finally close resources
  testDriver.close();
  spansSerde.close();
}