@Test
public void dropPartition() throws IOException {

    final BiFunction<Integer, LocalDateTime, ExpiringConsumer<String>> writerBuilder = mock(BiFunction.class);
    final OffsetComputer offsetComputer = new FixedOffsetComputer("0", 0);
    final PartitionedWriter<String> writer = new PartitionedWriter<>(writerBuilder, offsetComputer,
            "ignored", DataAccessEventProtos.FsEvent.newBuilder(), mock(Checkpointer.class));
    final ExpiringConsumer<String> firstMockConsumer = mock(ExpiringConsumer.class);
    final ExpiringConsumer<String> secondMockConsumer = mock(ExpiringConsumer.class);
    final Instant instant = Instant.EPOCH;
    when(writerBuilder.apply(any(), any(LocalDateTime.class))).thenReturn(firstMockConsumer).thenReturn(secondMockConsumer);
    writer.write(instant, buildOffset(1, 8), "Ignored");
    verify(firstMockConsumer, times(1)).write(anyLong(), anyString(), any(Offset.class));
    writer.dropPartition(1);
    writer.expireConsumers();
    writer.write(instant, buildOffset(1, 8), "Ignored");
    verify(secondMockConsumer, times(1)).write(anyLong(), anyString(), any(Offset.class));
    writer.expireConsumers();
    verify(secondMockConsumer, times(1)).isExpired();
    verifyNoMoreInteractions(firstMockConsumer);
    verifyNoMoreInteractions(secondMockConsumer);
}