@Test public void testBackwardsOrderAck() throws Exception { RandomEventProducer eventProducer = new RandomEventProducer(); FlushlessEventProducerHandler<Long> handler = new FlushlessEventProducerHandler<>(eventProducer); int partition = 0; SourcePartition tp = new SourcePartition(TOPIC, partition); for (int i = 0; i < 1000; i++) { sendEvent(tp, handler, i); } for (int i = 999; i > 0; i--) { eventProducer.process(tp, i); Assert.assertEquals(handler.getAckCheckpoint(TOPIC, partition), Optional.empty(), "Safe checkpoint should be empty"); } eventProducer.process(tp, 0); Assert.assertEquals(handler.getAckCheckpoint(TOPIC, partition).get(), Long.valueOf(999), "Safe checkpoint should be 999"); }
public Optional<T> getAckCheckpoint(String source, int sourcePartition) { CallbackStatus status = _callbackStatusMap.get(new SourcePartition(source, sourcePartition)); return Optional.ofNullable(status).map(CallbackStatus::getAckCheckpoint); }
FlushlessEventProducerHandler { public Optional<T> getAckCheckpoint(String source, int sourcePartition) { CallbackStatus status = _callbackStatusMap.get(new SourcePartition(source, sourcePartition)); return Optional.ofNullable(status).map(CallbackStatus::getAckCheckpoint); } }
FlushlessEventProducerHandler { public Optional<T> getAckCheckpoint(String source, int sourcePartition) { CallbackStatus status = _callbackStatusMap.get(new SourcePartition(source, sourcePartition)); return Optional.ofNullable(status).map(CallbackStatus::getAckCheckpoint); } FlushlessEventProducerHandler(DatastreamEventProducer eventProducer); }
FlushlessEventProducerHandler { public Optional<T> getAckCheckpoint(String source, int sourcePartition) { CallbackStatus status = _callbackStatusMap.get(new SourcePartition(source, sourcePartition)); return Optional.ofNullable(status).map(CallbackStatus::getAckCheckpoint); } FlushlessEventProducerHandler(DatastreamEventProducer eventProducer); void clear(); void send(DatastreamProducerRecord record, String source, int sourcePartition, T sourceCheckpoint, SendCallback callback); Optional<T> getAckCheckpoint(String source, int sourcePartition); long getInFlightCount(String source, int sourcePartition); Map<SourcePartition, Long> getInFlightMessagesCounts(); Optional<T> getAckCheckpoint(T currentCheckpoint, Comparator<T> checkpointComparator); }
FlushlessEventProducerHandler { public Optional<T> getAckCheckpoint(String source, int sourcePartition) { CallbackStatus status = _callbackStatusMap.get(new SourcePartition(source, sourcePartition)); return Optional.ofNullable(status).map(CallbackStatus::getAckCheckpoint); } FlushlessEventProducerHandler(DatastreamEventProducer eventProducer); void clear(); void send(DatastreamProducerRecord record, String source, int sourcePartition, T sourceCheckpoint, SendCallback callback); Optional<T> getAckCheckpoint(String source, int sourcePartition); long getInFlightCount(String source, int sourcePartition); Map<SourcePartition, Long> getInFlightMessagesCounts(); Optional<T> getAckCheckpoint(T currentCheckpoint, Comparator<T> checkpointComparator);  }