@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); }
@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 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 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); }
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); }
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);  }