OffsetResetter implements ConsumerRebalanceListener { @Override public void onPartitionsRevoked(Collection<TopicPartition> partitions) { for (PartitionedWriter<MESSAGE_KIND> writer : writers) { writer.close(); for (TopicPartition partition : partitions) { writer.dropPartition(partition.partition()); partitionRevokedConsumer.accept(partition.partition()); } } } OffsetResetter(Consumer<K, V> consumer, java.util.function.Consumer<Integer> partitionRevokedConsumer,
                          Collection<PartitionedWriter<MESSAGE_KIND>> writers); @Override void onPartitionsRevoked(Collection<TopicPartition> partitions); @Override void onPartitionsAssigned(Collection<TopicPartition> partitions);  }
@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); }
@Test public void partitionsRevokedNoTopic() { 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(Collections.emptyList()); verify(writer, times(1)).close(); verifyZeroInteractions(partitionsRevokedConsumer); }
@Test public void partitionsRevokedNoWriter() { final Consumer<Long, String> consumer = mock(Consumer.class); final OffsetResetter offsetResetter = new OffsetResetter<>(consumer, mock(java.util.function.Consumer.class), Collections.emptyList()); offsetResetter.onPartitionsRevoked(Collections.singleton(new TopicPartition(TOPIC, 1))); }