CompletionTimeProcessingDeriver extends AbstractProcessor<CompletionTimeProcessing, CompletionTimeProcessing> { @Override public CompletionTimeProcessing processOneToOne(String tenantId, CompletionTimeProcessing completionTimeProcessing) throws RetryAttemptException{ if (completionTimeProcessing.getCompletionTime() != null) { return null; } Span rootSpan = completionTimeProcessing.getRootSpan(); Set<Span> trace = spanCache.getTrace(tenantId, rootSpan.getTraceId()); if (trace == null) { throw new RetryAttemptException("trace not present in the cache"); } Annotation lastAnnotation = extractLastAnnotation(trace); if (lastAnnotation == null) { throw new RetryAttemptException("No annotation present, cannot calculate completion time - trying retry"); } if (completionTimeProcessing.getLastTimestamp() == null || completionTimeProcessing.getLastTimestamp() < lastAnnotation.getTimestamp()) { completionTimeProcessing.setLastTimestamp(lastAnnotation.getTimestamp()); return completionTimeProcessing; } CompletionTime completionTime = CompletionTimeUtil.spanToCompletionTime(spanCache, rootSpan); if (completionTime == null) { log.warnf("NO URL, span = %s", rootSpan); return null; } if (completionTime.getUri() == null) { if (rootSpan.serverSpan() && spanCache.get(null, SpanUniqueIdGenerator.getClientId(rootSpan.getId())) == null) { throw new RetryAttemptException("URL is null, span id = " + rootSpan.getId()); } } completionTime.setProperties(extractProperties(trace)); completionTime.setDuration(getTraceDuration(rootSpan, lastAnnotation.getTimestamp())); completionTimeProcessing.setCompletionTime(completionTime); return completionTimeProcessing; } CompletionTimeProcessingDeriver(SpanCache spanCache); @Override CompletionTimeProcessing processOneToOne(String tenantId, CompletionTimeProcessing completionTimeProcessing); @Override long getRetryDelay(List<CompletionTimeProcessing> completionTimeProcessings, int retryCount); @Override long getDeliveryDelay(List<CompletionTimeProcessing> completionTimeProcessings);  }
@Test public void testSpanTraceCompletionTime() throws RetryAttemptException { SpanCache spanCacheMock = Mockito.mock(SpanCache.class); CompletionTimeProcessingDeriver completionTimeProcessingDeriver = new CompletionTimeProcessingDeriver(spanCacheMock); Annotation sr = new Annotation(); sr.setValue("sr"); sr.setTimestamp(0); Annotation ss = new Annotation(); ss.setTimestamp(1000); ss.setValue("ss"); BinaryAnnotation httpAddress = new BinaryAnnotation(); httpAddress.setKey(Constants.ZIPKIN_BIN_ANNOTATION_HTTP_URL); httpAddress.setValue("http: Span rootSpan = new Span(Arrays.asList(httpAddress), Arrays.asList(sr, ss)); rootSpan.setId("trace"); rootSpan.setTraceId("trace"); rootSpan.setTimestamp(0L); rootSpan.setDuration(ss.getTimestamp() - sr.getTimestamp()); Mockito.when(spanCacheMock.getTrace(null, "trace")).thenReturn(new HashSet<>(Arrays.asList(rootSpan))); CompletionTimeProcessing completionTimeProcessing = new CompletionTimeProcessing(rootSpan); completionTimeProcessing = completionTimeProcessingDeriver.processOneToOne(null, completionTimeProcessing); Assert.assertNull(completionTimeProcessing.getCompletionTime()); completionTimeProcessing = completionTimeProcessingDeriver.processOneToOne(null, completionTimeProcessing); Assert.assertEquals(rootSpan.getDuration().longValue(), completionTimeProcessing.getCompletionTime().getDuration()); completionTimeProcessing = completionTimeProcessingDeriver.processOneToOne(null, completionTimeProcessing); Assert.assertNull(completionTimeProcessing); sr = new Annotation(); sr.setTimestamp(1500); ss = new Annotation(); ss.setTimestamp(2000); Span descendant = new Span(null, Arrays.asList(sr, ss)); descendant.setId("descendant"); descendant.setTraceId("trace"); descendant.setTimestamp(1500L); Mockito.when(spanCacheMock.getTrace(null, "trace")) .thenReturn(new HashSet<>(Arrays.asList(rootSpan, descendant))); completionTimeProcessing = new CompletionTimeProcessing(rootSpan); completionTimeProcessing = completionTimeProcessingDeriver.processOneToOne(null, completionTimeProcessing); Assert.assertNull(completionTimeProcessing.getCompletionTime()); completionTimeProcessing = completionTimeProcessingDeriver.processOneToOne(null, completionTimeProcessing); Assert.assertEquals(2000, completionTimeProcessing.getCompletionTime().getDuration()); completionTimeProcessing = completionTimeProcessingDeriver.processOneToOne(null, completionTimeProcessing); Assert.assertNull(completionTimeProcessing); Mockito.verify(spanCacheMock, Mockito.times(4)).getTrace(null, "trace"); }