@Test public void skipMessagesBeforeLowestOffset() throws IOException { final BiFunction<Integer, LocalDateTime, ExpiringConsumer<String>> writerBuilder = mock(BiFunction.class); final Checkpointer mockCheckpointer = mock(Checkpointer.class); final PartitionedWriter<String> writer = new PartitionedWriter<>(writerBuilder, new FixedOffsetComputer("ignored", 42), "ignored", DataAccessEventProtos.FsEvent.newBuilder(), mockCheckpointer); writer.write(Instant.EPOCH, buildOffset(1, 10), "Ignored"); writer.write(Instant.EPOCH, buildOffset(1, 11), "Also ignored"); verifyZeroInteractions(writerBuilder); verifyZeroInteractions(mockCheckpointer); }
public void write(Instant when, Offset offset, MESSAGE_KIND msg) throws IOException { final LocalDateTime dayStartTime = LocalDateTime.ofInstant(when.truncatedTo(ChronoUnit.DAYS), UTC_ZONE); final int partitionId = offset.getPartition(); final AbstractMap.SimpleEntry<Integer, LocalDateTime> dayAndPartition = new AbstractMap.SimpleEntry<>( partitionId, dayStartTime); if (!latestMessageTimeForPartitionAndDay.containsKey(dayAndPartition) || when.isAfter(latestMessageTimeForPartitionAndDay.get(dayAndPartition))) { latestMessageTimeForPartitionAndDay.put(dayAndPartition, when); } synchronized (perPartitionDayWriters) { if (shouldSkipOffset(offset.getOffset(), partitionId)) return; final ExpiringConsumer<MESSAGE_KIND> consumer = getWriter(dayStartTime, partitionId); consumer.write(when.toEpochMilli(), msg, offset); } }
PartitionedWriter implements Closeable { public void write(Instant when, Offset offset, MESSAGE_KIND msg) throws IOException { final LocalDateTime dayStartTime = LocalDateTime.ofInstant(when.truncatedTo(ChronoUnit.DAYS), UTC_ZONE); final int partitionId = offset.getPartition(); final AbstractMap.SimpleEntry<Integer, LocalDateTime> dayAndPartition = new AbstractMap.SimpleEntry<>( partitionId, dayStartTime); if (!latestMessageTimeForPartitionAndDay.containsKey(dayAndPartition) || when.isAfter(latestMessageTimeForPartitionAndDay.get(dayAndPartition))) { latestMessageTimeForPartitionAndDay.put(dayAndPartition, when); } synchronized (perPartitionDayWriters) { if (shouldSkipOffset(offset.getOffset(), partitionId)) return; final ExpiringConsumer<MESSAGE_KIND> consumer = getWriter(dayStartTime, partitionId); consumer.write(when.toEpochMilli(), msg, offset); } } }
PartitionedWriter implements Closeable { public void write(Instant when, Offset offset, MESSAGE_KIND msg) throws IOException { final LocalDateTime dayStartTime = LocalDateTime.ofInstant(when.truncatedTo(ChronoUnit.DAYS), UTC_ZONE); final int partitionId = offset.getPartition(); final AbstractMap.SimpleEntry<Integer, LocalDateTime> dayAndPartition = new AbstractMap.SimpleEntry<>( partitionId, dayStartTime); if (!latestMessageTimeForPartitionAndDay.containsKey(dayAndPartition) || when.isAfter(latestMessageTimeForPartitionAndDay.get(dayAndPartition))) { latestMessageTimeForPartitionAndDay.put(dayAndPartition, when); } synchronized (perPartitionDayWriters) { if (shouldSkipOffset(offset.getOffset(), partitionId)) return; final ExpiringConsumer<MESSAGE_KIND> consumer = getWriter(dayStartTime, partitionId); consumer.write(when.toEpochMilli(), msg, offset); } } PartitionedWriter(BiFunction<Integer, LocalDateTime, ExpiringConsumer<MESSAGE_KIND>> writerBuilder,
                             OffsetComputer offsetComputer, String eventName, Message.Builder emptyMessageBuilder,
                             Checkpointer checkpointer); }
PartitionedWriter implements Closeable { public void write(Instant when, Offset offset, MESSAGE_KIND msg) throws IOException { final LocalDateTime dayStartTime = LocalDateTime.ofInstant(when.truncatedTo(ChronoUnit.DAYS), UTC_ZONE); final int partitionId = offset.getPartition(); final AbstractMap.SimpleEntry<Integer, LocalDateTime> dayAndPartition = new AbstractMap.SimpleEntry<>( partitionId, dayStartTime); if (!latestMessageTimeForPartitionAndDay.containsKey(dayAndPartition) || when.isAfter(latestMessageTimeForPartitionAndDay.get(dayAndPartition))) { latestMessageTimeForPartitionAndDay.put(dayAndPartition, when); } synchronized (perPartitionDayWriters) { if (shouldSkipOffset(offset.getOffset(), partitionId)) return; final ExpiringConsumer<MESSAGE_KIND> consumer = getWriter(dayStartTime, partitionId); consumer.write(when.toEpochMilli(), msg, offset); } } PartitionedWriter(BiFunction<Integer, LocalDateTime, ExpiringConsumer<MESSAGE_KIND>> writerBuilder,
                             OffsetComputer offsetComputer, String eventName, Message.Builder emptyMessageBuilder,
                             Checkpointer checkpointer); void dropPartition(int partitionId); void write(Instant when, Offset offset, MESSAGE_KIND msg); @Override void close(); Map<Integer, Long> getStartingOffsets(Collection<Integer> partitionsId); void heartbeat(int partition, Offset offset); }
PartitionedWriter implements Closeable { public void write(Instant when, Offset offset, MESSAGE_KIND msg) throws IOException { final LocalDateTime dayStartTime = LocalDateTime.ofInstant(when.truncatedTo(ChronoUnit.DAYS), UTC_ZONE); final int partitionId = offset.getPartition(); final AbstractMap.SimpleEntry<Integer, LocalDateTime> dayAndPartition = new AbstractMap.SimpleEntry<>( partitionId, dayStartTime); if (!latestMessageTimeForPartitionAndDay.containsKey(dayAndPartition) || when.isAfter(latestMessageTimeForPartitionAndDay.get(dayAndPartition))) { latestMessageTimeForPartitionAndDay.put(dayAndPartition, when); } synchronized (perPartitionDayWriters) { if (shouldSkipOffset(offset.getOffset(), partitionId)) return; final ExpiringConsumer<MESSAGE_KIND> consumer = getWriter(dayStartTime, partitionId); consumer.write(when.toEpochMilli(), msg, offset); } } PartitionedWriter(BiFunction<Integer, LocalDateTime, ExpiringConsumer<MESSAGE_KIND>> writerBuilder,
                             OffsetComputer offsetComputer, String eventName, Message.Builder emptyMessageBuilder,
                             Checkpointer checkpointer); void dropPartition(int partitionId); void write(Instant when, Offset offset, MESSAGE_KIND msg); @Override void close(); Map<Integer, Long> getStartingOffsets(Collection<Integer> partitionsId); void heartbeat(int partition, Offset offset);  }