Refactoring streams and storage

DESIGN.md

@@ -51,6 +51,14 @@ defines if a trace is still active or not. This is evaluated on the next span re
 regardless of incoming `traceId`. If session window is closed, a trace message is emitted to the 
 traces topic.
 
+![Session Windows](https://kafka.apache.org/20/images/streams-session-windows-02.png)
+
+Each color represents a trace. The longer `inactivity gap` (default: 1 minute) we have, the longer we wait 
+to close a window and the longer we wait to emit traces downstream for dependency link and additional
+aggregations; but also the more consistent the trace aggregation is.
+If we choose a smaller gap, then we emit traces faster with the risk of breaking traces into 
+smaller chunks, and potentially affecting counters downstream.
+
 **Dependencies**
 
 Once `traces` are emitted downstream as part of the initial processing, dependency links are evaluated

autoconfigure/src/main/java/zipkin2/autoconfigure/storage/kafka/ZipkinKafkaStorageProperties.java

@@ -28,7 +28,7 @@ public class ZipkinKafkaStorageProperties implements Serializable {
 
   private String bootstrapServers;
 
-  private Long traceTtlCheckInterval;
+  private Long traceFlushInterval;
   private Long traceTtl;
   private Long traceInactivityGap;
 
@@ -60,8 +60,8 @@ KafkaStorage.Builder toBuilder() {
     if (traceInactivityGap != null) {
       builder.traceInactivityGap(Duration.ofMillis(traceInactivityGap));
     }
-    if (traceTtlCheckInterval != null) {
-        builder.traceTtlCheckInterval(Duration.ofMillis(traceTtlCheckInterval));
+    if (traceFlushInterval != null) {
+        builder.traceFlushInterval(Duration.ofMillis(traceFlushInterval));
     }
     if (traceTtl != null) {
       builder.traceTtl(Duration.ofMillis(traceTtl));
@@ -110,12 +110,12 @@ public void setBootstrapServers(String bootstrapServers) {
     this.bootstrapServers = bootstrapServers;
   }
 
-  public Long getTraceTtlCheckInterval() {
-    return traceTtlCheckInterval;
+  public Long getTraceFlushInterval() {
+    return traceFlushInterval;
   }
 
-  public void setTraceTtlCheckInterval(Long traceTtlCheckInterval) {
-    this.traceTtlCheckInterval = traceTtlCheckInterval;
+  public void setTraceFlushInterval(Long traceFlushInterval) {
+    this.traceFlushInterval = traceFlushInterval;
   }
 
   public Long getTraceTtl() {

autoconfigure/src/main/resources/zipkin-server-kafka.yml

@@ -15,7 +15,7 @@ zipkin:
 
       span-consumer-enabled: true
 
-      trace-ttl-check-interval: ${KAFKA_STORAGE_TRACE_TTL_CHECK_INTERVAL:3600000}
+      trace-flush-interval: ${KAFKA_STORAGE_TRACE_FLUSH_INTERVAL:3600000}
       trace-ttl: ${KAFKA_STORAGE_TRACE_TTL:259200000}
       trace-inactivity-gap: ${KAFKA_STORAGE_TRACE_INACTIVITY_GAP:30000}
 

storage/src/main/java/zipkin2/storage/kafka/KafkaStorage.java

@@ -108,7 +108,7 @@ public class KafkaStorage extends StorageComponent {
     traceStoreTopology = new TraceStoreTopologySupplier(
         spansTopicName,
         autocompleteKeys,
-        builder.traceTtlCheckInterval,
+        builder.traceFlushInterval,
         builder.traceTtl,
         builder.minTracesStored).get();
     dependencyStoreTopology = new DependencyStoreTopologySupplier(
@@ -342,7 +342,7 @@ public static class Builder extends StorageComponent.Builder {
     List<String> autocompleteKeys = new ArrayList<>();
 
     Duration traceTtl = Duration.ofDays(3);
-    Duration traceTtlCheckInterval = Duration.ofHours(1);
+    Duration traceFlushInterval = Duration.ofHours(1);
     Duration traceInactivityGap = Duration.ofMinutes(1);
     Duration dependencyTtl = Duration.ofDays(7);
     Duration dependencyWindowSize = Duration.ofMinutes(1);
@@ -537,11 +537,11 @@ public Builder storeDirectory(String storeDirectory) {
     /**
      * Frequency to check retention policy.
      */
-    public Builder traceTtlCheckInterval(Duration traceTtlCheckInterval) {
-      if (traceTtlCheckInterval == null) {
-        throw new NullPointerException("traceTtlCheckInterval == null");
+    public Builder traceFlushInterval(Duration traceFlushInterval) {
+      if (traceFlushInterval == null) {
+        throw new NullPointerException("traceFlushInterval == null");
       }
-      this.traceTtlCheckInterval = traceTtlCheckInterval;
+      this.traceFlushInterval = traceFlushInterval;
       return this;
     }
 

storage/src/main/java/zipkin2/storage/kafka/streams/TraceStoreTopologySupplier.java

@@ -55,20 +55,20 @@ public class TraceStoreTopologySupplier implements Supplier<Topology> {
   // Kafka topics
   final String spansTopicName;
   // Limits
-  final List<String> autocompleteKeys;
-  final Duration tracesGcInterval;
+  final List<String> autoCompleteKeys;
+  final Duration tracesFlushInterval;
   final Duration tracesTtl;
   final long minTracesStored;
   // SerDes
   final SpansSerde spansSerde;
   final SpanIdsSerde spanIdsSerde;
   final NamesSerde namesSerde;
 
-  public TraceStoreTopologySupplier(String spansTopicName, List<String> autocompleteKeys,
-      Duration tracesGcInterval, Duration tracesTtl, long minTracesStored) {
+  public TraceStoreTopologySupplier(String spansTopicName, List<String> autoCompleteKeys,
+                                    Duration tracesFlushInterval, Duration tracesTtl, long minTracesStored) {
     this.spansTopicName = spansTopicName;
-    this.autocompleteKeys = autocompleteKeys;
-    this.tracesGcInterval = tracesGcInterval;
+    this.autoCompleteKeys = autoCompleteKeys;
+    this.tracesFlushInterval = tracesFlushInterval;
     this.tracesTtl = tracesTtl;
     this.minTracesStored = minTracesStored;
     spansSerde = new SpansSerde();
@@ -124,7 +124,7 @@ public TraceStoreTopologySupplier(String spansTopicName, List<String> autocomple
                 (KeyValueStore<Long, Set<String>>) context.getStateStore(SPAN_IDS_BY_TS_STORE_NAME);
             // Retention scheduling
             context.schedule(
-                tracesGcInterval,
+                tracesFlushInterval,
                 PunctuationType.STREAM_TIME,
                 timestamp -> {
                   if (tracesTtl.toMillis() > 0 &&
@@ -216,7 +216,7 @@ public void process(String traceId, List<Span> spans) {
               }
               if (!span.tags().isEmpty()) {
                 span.tags().forEach((key, value) -> {
-                  if (autocompleteKeys.contains(key)) {
+                  if (autoCompleteKeys.contains(key)) {
                     Set<String> values = autocompleteTagsStore.get(key);
                     if (values == null) values = new HashSet<>();
                     values.add(value);