@Test public void parallelCheckpoint() throws Exception { int spansPerEvent = 3; int eventsPerCheckpoint = processor.checkpointBatchSize / spansPerEvent; if (processor.checkpointBatchSize % spansPerEvent > 0) eventsPerCheckpoint++; int eventCount = 1000; final ConcurrentLinkedQueue<EventData> events = new ConcurrentLinkedQueue<>(); for (int i = 0; i < eventCount; i++) { events.add(jsonMessageWithThreeSpans(Integer.toHexString(i + 1), 1 + i)); } CountDownLatch latch = new CountDownLatch(events.size()); int threadCount = 10; ExecutorService exec = Executors.newFixedThreadPool(threadCount); for (int i = 0; i < threadCount; i++) { exec.execute( () -> { EventData event; while ((event = events.poll()) != null) { try { processor.onEvents(context, asList(event)); } catch (Exception e) { e.printStackTrace(); } latch.countDown(); } }); } exec.shutdown(); exec.awaitTermination(1, TimeUnit.SECONDS); assertThat(processor.countSinceCheckpoint).isZero(); assertThat(checkpointEvents).hasSize(eventCount / eventsPerCheckpoint); }
@Override public void onEvents(PartitionContext context, Iterable<EventData> messages) throws ExecutionException, InterruptedException { List<Span> buffer = new ArrayList<>(); for (EventData data : messages) { byte[] bytes = data.getBytes(); BytesDecoder<Span> decoder = decoderForListMessage(bytes); List<Span> nextSpans = decoder.decodeList(bytes); buffer.addAll(nextSpans); if (maybeCheckpoint(context, data, nextSpans.size())) { collector.accept(buffer, NOOP); buffer.clear(); } } if (!buffer.isEmpty()) { collector.accept(buffer, NOOP); } }
ZipkinEventProcessor implements IEventProcessor { @Override public void onEvents(PartitionContext context, Iterable<EventData> messages) throws ExecutionException, InterruptedException { List<Span> buffer = new ArrayList<>(); for (EventData data : messages) { byte[] bytes = data.getBytes(); BytesDecoder<Span> decoder = decoderForListMessage(bytes); List<Span> nextSpans = decoder.decodeList(bytes); buffer.addAll(nextSpans); if (maybeCheckpoint(context, data, nextSpans.size())) { collector.accept(buffer, NOOP); buffer.clear(); } } if (!buffer.isEmpty()) { collector.accept(buffer, NOOP); } } }
ZipkinEventProcessor implements IEventProcessor { @Override public void onEvents(PartitionContext context, Iterable<EventData> messages) throws ExecutionException, InterruptedException { List<Span> buffer = new ArrayList<>(); for (EventData data : messages) { byte[] bytes = data.getBytes(); BytesDecoder<Span> decoder = decoderForListMessage(bytes); List<Span> nextSpans = decoder.decodeList(bytes); buffer.addAll(nextSpans); if (maybeCheckpoint(context, data, nextSpans.size())) { collector.accept(buffer, NOOP); buffer.clear(); } } if (!buffer.isEmpty()) { collector.accept(buffer, NOOP); } } ZipkinEventProcessor(Collector collector, int checkpointBatchSize);   ZipkinEventProcessor(Logger logger, Collector collector, int checkpointBatchSize); }
ZipkinEventProcessor implements IEventProcessor { @Override public void onEvents(PartitionContext context, Iterable<EventData> messages) throws ExecutionException, InterruptedException { List<Span> buffer = new ArrayList<>(); for (EventData data : messages) { byte[] bytes = data.getBytes(); BytesDecoder<Span> decoder = decoderForListMessage(bytes); List<Span> nextSpans = decoder.decodeList(bytes); buffer.addAll(nextSpans); if (maybeCheckpoint(context, data, nextSpans.size())) { collector.accept(buffer, NOOP); buffer.clear(); } } if (!buffer.isEmpty()) { collector.accept(buffer, NOOP); } } ZipkinEventProcessor(Collector collector, int checkpointBatchSize);   ZipkinEventProcessor(Logger logger, Collector collector, int checkpointBatchSize); @Override void onOpen(PartitionContext context); @Override void onClose(PartitionContext context, CloseReason reason); @Override void onEvents(PartitionContext context, Iterable<EventData> messages); @Override void onError(PartitionContext context, Throwable error); }
ZipkinEventProcessor implements IEventProcessor { @Override public void onEvents(PartitionContext context, Iterable<EventData> messages) throws ExecutionException, InterruptedException { List<Span> buffer = new ArrayList<>(); for (EventData data : messages) { byte[] bytes = data.getBytes(); BytesDecoder<Span> decoder = decoderForListMessage(bytes); List<Span> nextSpans = decoder.decodeList(bytes); buffer.addAll(nextSpans); if (maybeCheckpoint(context, data, nextSpans.size())) { collector.accept(buffer, NOOP); buffer.clear(); } } if (!buffer.isEmpty()) { collector.accept(buffer, NOOP); } } ZipkinEventProcessor(Collector collector, int checkpointBatchSize);   ZipkinEventProcessor(Logger logger, Collector collector, int checkpointBatchSize); @Override void onOpen(PartitionContext context); @Override void onClose(PartitionContext context, CloseReason reason); @Override void onEvents(PartitionContext context, Iterable<EventData> messages); @Override void onError(PartitionContext context, Throwable error);  }