@Test public void partitionsAssignedNoWriter() { final Consumer<Long, String> consumer = mock(Consumer.class); final OffsetResetter offsetResetter = new OffsetResetter<>(consumer, mock(java.util.function.Consumer.class), Collections.emptyList()); final TopicPartition partition = new TopicPartition(TOPIC, 1); final List<TopicPartition> partitions = Collections.singletonList(partition); offsetResetter.onPartitionsAssigned(partitions); verify(consumer, times(1)).seekToBeginning(Collections.singleton(partition)); verifyNoMoreInteractions(consumer); }
@Override public void onPartitionsAssigned(Collection<TopicPartition> partitions) { Map<Integer, Collection<Long>> offsetsPerPartition = new HashMap<>(); for (PartitionedWriter<MESSAGE_KIND> writer : writers) { final List<Integer> partitionsId = partitions.stream() .map(TopicPartition::partition) .collect(Collectors.toList()); try { final Map<Integer, Long> startingOffsets = writer.getStartingOffsets(partitionsId); startingOffsets.forEach((part, offset) -> offsetsPerPartition.computeIfAbsent(part, ignored -> new HashSet<>()).add(offset)); } catch (IOException e) { LOGGER.warn("Couldn't get offset for partitions {}, will resume from earliest", partitionsId.stream().map(String::valueOf).collect(Collectors.joining(", ")), e); partitions.forEach(part -> offsetsPerPartition.computeIfAbsent(part.partition(), ignored -> new HashSet<>()) .add(UNKNOWN_OFFSET)); } } Map<Integer, Long> minOffsetPerPartition = offsetsPerPartition .entrySet() .stream() .collect(Collectors.toMap(Map.Entry::getKey, offsets -> offsets .getValue() .stream() .filter(offset -> offset != UNKNOWN_OFFSET) .mapToLong(l -> l) .min() .orElse(UNKNOWN_OFFSET)) ); for (TopicPartition topicPartition: partitions) { int partition = topicPartition.partition(); long startingOffset = minOffsetPerPartition.getOrDefault(partition, UNKNOWN_OFFSET); synchronized (consumer) { if (startingOffset == UNKNOWN_OFFSET) { consumer.seekToBeginning(Collections.singleton(topicPartition)); LOGGER.warn("Resuming consumption of partition {} from the beginning. " + "This should not happen unless this is the first time this app runs", partition); } else { consumer.seek(topicPartition, startingOffset); LOGGER.info("Resuming consumption of partition {} from offset {}", partition, startingOffset); } } } }
OffsetResetter implements ConsumerRebalanceListener { @Override public void onPartitionsAssigned(Collection<TopicPartition> partitions) { Map<Integer, Collection<Long>> offsetsPerPartition = new HashMap<>(); for (PartitionedWriter<MESSAGE_KIND> writer : writers) { final List<Integer> partitionsId = partitions.stream() .map(TopicPartition::partition) .collect(Collectors.toList()); try { final Map<Integer, Long> startingOffsets = writer.getStartingOffsets(partitionsId); startingOffsets.forEach((part, offset) -> offsetsPerPartition.computeIfAbsent(part, ignored -> new HashSet<>()).add(offset)); } catch (IOException e) { LOGGER.warn("Couldn't get offset for partitions {}, will resume from earliest", partitionsId.stream().map(String::valueOf).collect(Collectors.joining(", ")), e); partitions.forEach(part -> offsetsPerPartition.computeIfAbsent(part.partition(), ignored -> new HashSet<>()) .add(UNKNOWN_OFFSET)); } } Map<Integer, Long> minOffsetPerPartition = offsetsPerPartition .entrySet() .stream() .collect(Collectors.toMap(Map.Entry::getKey, offsets -> offsets .getValue() .stream() .filter(offset -> offset != UNKNOWN_OFFSET) .mapToLong(l -> l) .min() .orElse(UNKNOWN_OFFSET)) ); for (TopicPartition topicPartition: partitions) { int partition = topicPartition.partition(); long startingOffset = minOffsetPerPartition.getOrDefault(partition, UNKNOWN_OFFSET); synchronized (consumer) { if (startingOffset == UNKNOWN_OFFSET) { consumer.seekToBeginning(Collections.singleton(topicPartition)); LOGGER.warn("Resuming consumption of partition {} from the beginning. " + "This should not happen unless this is the first time this app runs", partition); } else { consumer.seek(topicPartition, startingOffset); LOGGER.info("Resuming consumption of partition {} from offset {}", partition, startingOffset); } } } } }
OffsetResetter implements ConsumerRebalanceListener { @Override public void onPartitionsAssigned(Collection<TopicPartition> partitions) { Map<Integer, Collection<Long>> offsetsPerPartition = new HashMap<>(); for (PartitionedWriter<MESSAGE_KIND> writer : writers) { final List<Integer> partitionsId = partitions.stream() .map(TopicPartition::partition) .collect(Collectors.toList()); try { final Map<Integer, Long> startingOffsets = writer.getStartingOffsets(partitionsId); startingOffsets.forEach((part, offset) -> offsetsPerPartition.computeIfAbsent(part, ignored -> new HashSet<>()).add(offset)); } catch (IOException e) { LOGGER.warn("Couldn't get offset for partitions {}, will resume from earliest", partitionsId.stream().map(String::valueOf).collect(Collectors.joining(", ")), e); partitions.forEach(part -> offsetsPerPartition.computeIfAbsent(part.partition(), ignored -> new HashSet<>()) .add(UNKNOWN_OFFSET)); } } Map<Integer, Long> minOffsetPerPartition = offsetsPerPartition .entrySet() .stream() .collect(Collectors.toMap(Map.Entry::getKey, offsets -> offsets .getValue() .stream() .filter(offset -> offset != UNKNOWN_OFFSET) .mapToLong(l -> l) .min() .orElse(UNKNOWN_OFFSET)) ); for (TopicPartition topicPartition: partitions) { int partition = topicPartition.partition(); long startingOffset = minOffsetPerPartition.getOrDefault(partition, UNKNOWN_OFFSET); synchronized (consumer) { if (startingOffset == UNKNOWN_OFFSET) { consumer.seekToBeginning(Collections.singleton(topicPartition)); LOGGER.warn("Resuming consumption of partition {} from the beginning. " + "This should not happen unless this is the first time this app runs", partition); } else { consumer.seek(topicPartition, startingOffset); LOGGER.info("Resuming consumption of partition {} from offset {}", partition, startingOffset); } } } } OffsetResetter(Consumer<K, V> consumer, java.util.function.Consumer<Integer> partitionRevokedConsumer,
                          Collection<PartitionedWriter<MESSAGE_KIND>> writers); }
OffsetResetter implements ConsumerRebalanceListener { @Override public void onPartitionsAssigned(Collection<TopicPartition> partitions) { Map<Integer, Collection<Long>> offsetsPerPartition = new HashMap<>(); for (PartitionedWriter<MESSAGE_KIND> writer : writers) { final List<Integer> partitionsId = partitions.stream() .map(TopicPartition::partition) .collect(Collectors.toList()); try { final Map<Integer, Long> startingOffsets = writer.getStartingOffsets(partitionsId); startingOffsets.forEach((part, offset) -> offsetsPerPartition.computeIfAbsent(part, ignored -> new HashSet<>()).add(offset)); } catch (IOException e) { LOGGER.warn("Couldn't get offset for partitions {}, will resume from earliest", partitionsId.stream().map(String::valueOf).collect(Collectors.joining(", ")), e); partitions.forEach(part -> offsetsPerPartition.computeIfAbsent(part.partition(), ignored -> new HashSet<>()) .add(UNKNOWN_OFFSET)); } } Map<Integer, Long> minOffsetPerPartition = offsetsPerPartition .entrySet() .stream() .collect(Collectors.toMap(Map.Entry::getKey, offsets -> offsets .getValue() .stream() .filter(offset -> offset != UNKNOWN_OFFSET) .mapToLong(l -> l) .min() .orElse(UNKNOWN_OFFSET)) ); for (TopicPartition topicPartition: partitions) { int partition = topicPartition.partition(); long startingOffset = minOffsetPerPartition.getOrDefault(partition, UNKNOWN_OFFSET); synchronized (consumer) { if (startingOffset == UNKNOWN_OFFSET) { consumer.seekToBeginning(Collections.singleton(topicPartition)); LOGGER.warn("Resuming consumption of partition {} from the beginning. " + "This should not happen unless this is the first time this app runs", partition); } else { consumer.seek(topicPartition, startingOffset); LOGGER.info("Resuming consumption of partition {} from offset {}", partition, startingOffset); } } } } 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 onPartitionsAssigned(Collection<TopicPartition> partitions) { Map<Integer, Collection<Long>> offsetsPerPartition = new HashMap<>(); for (PartitionedWriter<MESSAGE_KIND> writer : writers) { final List<Integer> partitionsId = partitions.stream() .map(TopicPartition::partition) .collect(Collectors.toList()); try { final Map<Integer, Long> startingOffsets = writer.getStartingOffsets(partitionsId); startingOffsets.forEach((part, offset) -> offsetsPerPartition.computeIfAbsent(part, ignored -> new HashSet<>()).add(offset)); } catch (IOException e) { LOGGER.warn("Couldn't get offset for partitions {}, will resume from earliest", partitionsId.stream().map(String::valueOf).collect(Collectors.joining(", ")), e); partitions.forEach(part -> offsetsPerPartition.computeIfAbsent(part.partition(), ignored -> new HashSet<>()) .add(UNKNOWN_OFFSET)); } } Map<Integer, Long> minOffsetPerPartition = offsetsPerPartition .entrySet() .stream() .collect(Collectors.toMap(Map.Entry::getKey, offsets -> offsets .getValue() .stream() .filter(offset -> offset != UNKNOWN_OFFSET) .mapToLong(l -> l) .min() .orElse(UNKNOWN_OFFSET)) ); for (TopicPartition topicPartition: partitions) { int partition = topicPartition.partition(); long startingOffset = minOffsetPerPartition.getOrDefault(partition, UNKNOWN_OFFSET); synchronized (consumer) { if (startingOffset == UNKNOWN_OFFSET) { consumer.seekToBeginning(Collections.singleton(topicPartition)); LOGGER.warn("Resuming consumption of partition {} from the beginning. " + "This should not happen unless this is the first time this app runs", partition); } else { consumer.seek(topicPartition, startingOffset); LOGGER.info("Resuming consumption of partition {} from offset {}", partition, startingOffset); } } } } 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);  }