@Test public void checkpointsOnBatchSize() throws Exception { EventData event1 = jsonMessageWithThreeSpans("a", 1); EventData event2 = json2MessageWithThreeSpans("b", 2); EventData event3 = thriftMessageWithThreeSpans("c", 3); EventData event4 = json2MessageWithThreeSpans("d", 4); processor.onEvents(context, asList(event1, event2, event3)); assertThat(checkpointEvents).isEmpty(); assertThat(logger.messages.poll()).isNull(); processor.onEvents(context, asList(event4)); assertThat(checkpointEvents).containsExactly(event4); assertThat(logger.messages.poll()).isEqualTo("FINE: Partition 1 checkpointing at d,4"); }
@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);  }