RemoteReporter implements Reporter { void flush() { metrics.reporterQueueLength.update(commandQueue.size()); commandQueue.offer(new FlushCommand()); } private  RemoteReporter(Sender sender, int flushInterval, int maxQueueSize, int closeEnqueueTimeout,
      Metrics metrics); @Override void report(JaegerSpan span); @Override void close(); static final int DEFAULT_FLUSH_INTERVAL_MS; static final int DEFAULT_MAX_QUEUE_SIZE; }
@Test public void testFlushIsCalledOnSender() throws InterruptedException { CountDownLatch latch = new CountDownLatch(1); InMemorySender sender = new InMemorySender() { @Override public int flush() throws SenderException { latch.countDown(); return super.flush(); } }; RemoteReporter remoteReporter = new Builder() .withSender(sender) .withFlushInterval(flushInterval) .withMaxQueueSize(maxQueueSize) .withMetrics(metrics) .build(); tracer = new JaegerTracer.Builder("test-remote-reporter") .withReporter(remoteReporter) .withSampler(new ConstSampler(true)) .withMetrics(metrics) .build(); tracer.buildSpan("mySpan").start().finish(); remoteReporter.flush(); latch.await(); assertEquals("Should have called the custom sender flush", 0, latch.getCount()); assertEquals("mySpan", (sender.getReceived().get(0)).getOperationName()); }
@Test public void testFlushErrorsLoggedJustOnce() throws InterruptedException { Object logMonitor = new Object(); AtomicReference<String> logMsg = new AtomicReference<>(null); mockLogger(e -> { synchronized (logMonitor) { logMsg.set(e.getFormattedMessage()); logMonitor.notifyAll(); } }); class FailingSender extends InMemorySender { private final AtomicInteger flushCounter = new AtomicInteger(0); @Override public int flush() throws SenderException { int i = super.flush(); switch (flushCounter.getAndIncrement()) { case 1: case 2: case 3: throw new SenderException("test1", super.flush()); default: return i; } } private String awaitMessage(AtomicReference<String> ref) throws InterruptedException { synchronized (logMonitor) { while (ref.get() == null) { logMonitor.wait(); } return ref.getAndSet(null); } } } FailingSender sender = new FailingSender(); RemoteReporter remoteReporter = new Builder() .withSender(sender) .withFlushInterval(Integer.MAX_VALUE) .withMaxQueueSize(maxQueueSize) .withMetrics(metrics) .build(); tracer = new JaegerTracer.Builder("test-remote-reporter") .withReporter(remoteReporter) .withSampler(new ConstSampler(true)) .withMetrics(metrics) .build(); tracer.buildSpan("mySpan").start().finish(); remoteReporter.flush(); tracer.buildSpan("mySpan").start().finish(); remoteReporter.flush(); assertEquals("FlushCommand execution failed! Repeated errors of this command will not be logged.", sender.awaitMessage(logMsg)); remoteReporter.flush(); remoteReporter.flush(); remoteReporter.flush(); assertEquals("FlushCommand is working again!", sender.awaitMessage(logMsg)); }