Fresh results

Makefile

@@ -54,7 +54,7 @@ build: license-header
 	${MAVEN} clean install -DskipTests
 
 .PHONY: test
-test: build
+test:
 	${MAVEN} test verify
 
 .PHONY: zipkin-local

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

@@ -16,6 +16,7 @@
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.List;
 import java.util.Set;
 import org.apache.kafka.streams.KafkaStreams;
@@ -111,10 +112,12 @@ static class GetServiceNamesCall extends KafkaStreamsStoreCall<List<String>> {
 
     @Override public List<String> query() {
       List<String> serviceNames = new ArrayList<>();
-      serviceStore.all().forEachRemaining(keyValue -> {
-        // double check service names are unique
-        if (!serviceNames.contains(keyValue.value)) serviceNames.add(keyValue.value);
-      });
+      try (KeyValueIterator<String, String> all = serviceStore.all()) {
+        all.forEachRemaining(keyValue -> {
+          // double check service names are unique
+          if (!serviceNames.contains(keyValue.value)) serviceNames.add(keyValue.value);
+        });
+      }
       // comply with Zipkin API as service names are required to be ordered lexicographically
       Collections.sort(serviceNames);
       return serviceNames;
@@ -178,53 +181,74 @@ static class GetRemoteServiceNamesCall extends KafkaStreamsStoreCall<List<String
   static class GetTracesCall extends KafkaStreamsStoreCall<List<List<Span>>> {
     final ReadOnlyKeyValueStore<String, List<Span>> tracesStore;
     final ReadOnlyKeyValueStore<Long, Set<String>> traceIdsByTsStore;
-    final QueryRequest queryRequest;
+    final QueryRequest request;
 
     GetTracesCall(
         ReadOnlyKeyValueStore<String, List<Span>> tracesStore,
         ReadOnlyKeyValueStore<Long, Set<String>> traceIdsByTsStore,
-        QueryRequest queryRequest) {
+        QueryRequest request) {
       this.tracesStore = tracesStore;
       this.traceIdsByTsStore = traceIdsByTsStore;
-      this.queryRequest = queryRequest;
+      this.request = request;
     }
 
     @Override public List<List<Span>> query() {
       List<List<Span>> traces = new ArrayList<>();
       List<String> traceIds = new ArrayList<>();
       // milliseconds to microseconds
-      long from = (queryRequest.endTs() - queryRequest.lookback()) * 1000;
-      long to = queryRequest.endTs() * 1000;
-      // first index
-      KeyValueIterator<Long, Set<String>> spanIds = traceIdsByTsStore.range(from, to);
-      spanIds.forEachRemaining(keyValue -> {
-        for (String traceId : keyValue.value) {
-          if (!traceIds.contains(traceId) && traces.size() < queryRequest.limit()) {
-            List<Span> spans = tracesStore.get(traceId);
-            if (spans != null && queryRequest.test(spans)) { // apply filters
-              traceIds.add(traceId); // adding to check if we have already add it later
-              traces.add(spans);
+      long from = (request.endTs() - request.lookback()) * 1000;
+      long to = request.endTs() * 1000;
+      long checkpoint = to - (30 * 1000 * 1000); // 30 sec before upper bound
+      if (checkpoint <= from) { // do one run
+        try (KeyValueIterator<Long, Set<String>> spanIds = traceIdsByTsStore.range(from, to)) {
+          spanIds.forEachRemaining(keyValue -> {
+            for (String traceId : keyValue.value) {
+              if (!traceIds.contains(traceId)) {
+                List<Span> spans = tracesStore.get(traceId);
+                if (spans != null && request.test(spans)) { // apply filters
+                  traceIds.add(traceId); // adding to check if we have already add it later
+                  traces.add(spans);
+                }
+              }
             }
+          });
+        }
+      } else {
+        while (checkpoint > from && traces.size() < request.limit()) {
+          try (KeyValueIterator<Long, Set<String>> spanIds = traceIdsByTsStore.range(checkpoint, to)) {
+            spanIds.forEachRemaining(keyValue -> {
+              for (String traceId : keyValue.value) {
+                if (!traceIds.contains(traceId)) {
+                  List<Span> spans = tracesStore.get(traceId);
+                  if (spans != null && request.test(spans)) { // apply filters
+                    traceIds.add(traceId); // adding to check if we have already add it later
+                    traces.add(spans);
+                  }
+                }
+              }
+            });
           }
+          to = checkpoint;
+          checkpoint = checkpoint - (60 * 1000 * 1000); // 1 min before more
         }
-      });
-      LOG.debug("Traces found from query {}: {}", queryRequest, traces.size());
-      return traces;
+      }
+      traces.sort(Comparator.<List<Span>>comparingLong(o -> o.get(0).timestampAsLong()).reversed());
+      LOG.debug("Traces found from query {}: {}", request, traces.size());
+      return traces.subList(0,
+          request.limit() >= traces.size() ? traceIds.size() : request.limit());
     }
 
     @Override
     public Call<List<List<Span>>> clone() {
-      return new GetTracesCall(tracesStore, traceIdsByTsStore, queryRequest);
+      return new GetTracesCall(tracesStore, traceIdsByTsStore, request);
     }
   }
 
   static class GetTraceCall extends KafkaStreamsStoreCall<List<Span>> {
     final ReadOnlyKeyValueStore<String, List<Span>> traceStore;
     final String traceId;
 
-    GetTraceCall(
-        ReadOnlyKeyValueStore<String, List<Span>> traceStore,
-        String traceId) {
+    GetTraceCall(ReadOnlyKeyValueStore<String, List<Span>> traceStore, String traceId) {
       this.traceStore = traceStore;
       this.traceId = traceId;
     }

storage/src/test/java/zipkin2/storage/kafka/KafkaStorageIT.java

@@ -158,9 +158,14 @@ class KafkaStorageIT {
         .localEndpoint(Endpoint.newBuilder().serviceName("svc_b").build())
         .timestamp(TODAY * 1000).duration(2)
         .build();
+    Span other = Span.newBuilder().traceId("c").id("c").name("op_c").kind(Span.Kind.SERVER)
+        .localEndpoint(Endpoint.newBuilder().serviceName("svc_c").build())
+        .timestamp(TODAY * 1000 + 10).duration(8)
+        .build();
     List<Span> spans = Arrays.asList(parent, child);
     // When: been published
     tracesProducer.send(new ProducerRecord<>(storage.spansTopicName, parent.traceId(), spans));
+    tracesProducer.send(new ProducerRecord<>(storage.spansTopicName, other.traceId(), Collections.singletonList(other)));
     tracesProducer.flush();
     // Then: stored
     IntegrationTestUtils.waitUntilMinRecordsReceived(
@@ -169,14 +174,14 @@ class KafkaStorageIT {
     SpanStore spanStore = storage.spanStore();
     ServiceAndSpanNames serviceAndSpanNames = storage.serviceAndSpanNames();
     // Then: services names are searchable
-    await().atMost(30, TimeUnit.SECONDS)
+    await().atMost(10, TimeUnit.SECONDS)
         .until(() -> {
           List<List<Span>> traces = new ArrayList<>();
           try {
             traces =
                 spanStore.getTraces(QueryRequest.newBuilder()
                     .endTs(TODAY + 1)
-                    .lookback(Duration.ofMinutes(1).toMillis())
+                    .lookback(Duration.ofSeconds(30).toMillis())
                     .serviceName("svc_a")
                     .limit(10)
                     .build())
@@ -189,6 +194,25 @@ class KafkaStorageIT {
           return traces.size() == 1
               && traces.get(0).size() == 2; // Trace is found and has two spans
         });
+    await().atMost(10, TimeUnit.SECONDS)
+        .until(() -> {
+          List<List<Span>> traces = new ArrayList<>();
+          try {
+            traces =
+                spanStore.getTraces(QueryRequest.newBuilder()
+                    .endTs(TODAY + 1)
+                    .lookback(Duration.ofMinutes(1).toMillis())
+                    .limit(1)
+                    .build())
+                    .execute();
+          } catch (InvalidStateStoreException e) { // ignoring state issues
+            System.err.println(e.getMessage());
+          } catch (Exception e) {
+            e.printStackTrace();
+          }
+          return traces.size() == 1
+              && traces.get(0).size() == 1; // last trace is returned first
+        });
     await().atMost(5, TimeUnit.SECONDS)
         .until(() -> {
           List<String> services = new ArrayList<>();
@@ -199,7 +223,7 @@ class KafkaStorageIT {
           } catch (Exception e) {
             e.printStackTrace();
           }
-          return services.size() == 2;
+          return services.size() == 3;
         }); // There are two service names
     await().atMost(5, TimeUnit.SECONDS)
         .until(() -> {