@Test
public void partitionsRevoked() {

    final Consumer<Long, String> consumer = mock(Consumer.class);
    final PartitionedWriter writer = mock(PartitionedWriter.class);
    final java.util.function.Consumer partitionsRevokedConsumer = mock(java.util.function.Consumer.class);
    final OffsetResetter offsetResetter = new OffsetResetter<>(consumer, partitionsRevokedConsumer,
            Collections.singleton(writer));
    offsetResetter.onPartitionsRevoked(Arrays.asList(new TopicPartition(TOPIC, 1),
            new TopicPartition(TOPIC, 2)));
    verify(writer, times(1)).close();
    verify(writer, times(1)).dropPartition(1);
    verify(writer, times(1)).dropPartition(2);
    verify(partitionsRevokedConsumer, times(1)).accept(1);
    verify(partitionsRevokedConsumer, times(1)).accept(2);
    verifyNoMoreInteractions(writer);
}