@Test void testOnPartitionsAssignedUsesTokenOffsetsUponConsumerSeek() { long testOffsetForPartitionZero = 5L; long testOffsetForPartitionOne = 10L; long testOffsetForPartitionTwo = 15L; Map<TopicPartition, Long> testPositions = new HashMap<>(); testPositions.put(new TopicPartition(TEST_TOPIC, 0), testOffsetForPartitionZero); testPositions.put(new TopicPartition(TEST_TOPIC, 1), testOffsetForPartitionOne); testPositions.put(new TopicPartition(TEST_TOPIC, 2), testOffsetForPartitionTwo); KafkaTrackingToken testToken = KafkaTrackingToken.newInstance(testPositions); TopicPartition testPartitionZero = new TopicPartition(TEST_TOPIC, 0); TopicPartition testPartitionOne = new TopicPartition(TEST_TOPIC, 1); TopicPartition testPartitionTwo = new TopicPartition(TEST_TOPIC, 2); ArrayList<TopicPartition> testAssignedPartitions = new ArrayList<>(); testAssignedPartitions.add(testPartitionZero); testAssignedPartitions.add(testPartitionOne); testAssignedPartitions.add(testPartitionTwo); TrackingTokenConsumerRebalanceListener<?, ?> testSubject = new TrackingTokenConsumerRebalanceListener<>(consumer, () -> testToken); testSubject.onPartitionsAssigned(testAssignedPartitions); verify(consumer).seek(testPartitionZero, testOffsetForPartitionZero + 1); verify(consumer).seek(testPartitionOne, testOffsetForPartitionOne + 1); verify(consumer).seek(testPartitionTwo, testOffsetForPartitionTwo + 1); }
@Override public void onPartitionsAssigned(Collection<TopicPartition> assignedPartitions) { KafkaTrackingToken currentToken = tokenSupplier.get(); assignedPartitions.forEach(assignedPartition -> { Map<TopicPartition, Long> tokenPartitionPositions = currentToken.positions(); long offset = 0L; if (tokenPartitionPositions.containsKey(assignedPartition)) { offset = tokenPartitionPositions.get(assignedPartition) + 1; } logger.info("Seeking topic-partition [{}] with offset [{}]", assignedPartition, offset); consumer.seek(assignedPartition, offset); }); }
TrackingTokenConsumerRebalanceListener implements ConsumerRebalanceListener { @Override public void onPartitionsAssigned(Collection<TopicPartition> assignedPartitions) { KafkaTrackingToken currentToken = tokenSupplier.get(); assignedPartitions.forEach(assignedPartition -> { Map<TopicPartition, Long> tokenPartitionPositions = currentToken.positions(); long offset = 0L; if (tokenPartitionPositions.containsKey(assignedPartition)) { offset = tokenPartitionPositions.get(assignedPartition) + 1; } logger.info("Seeking topic-partition [{}] with offset [{}]", assignedPartition, offset); consumer.seek(assignedPartition, offset); }); } }
TrackingTokenConsumerRebalanceListener implements ConsumerRebalanceListener { @Override public void onPartitionsAssigned(Collection<TopicPartition> assignedPartitions) { KafkaTrackingToken currentToken = tokenSupplier.get(); assignedPartitions.forEach(assignedPartition -> { Map<TopicPartition, Long> tokenPartitionPositions = currentToken.positions(); long offset = 0L; if (tokenPartitionPositions.containsKey(assignedPartition)) { offset = tokenPartitionPositions.get(assignedPartition) + 1; } logger.info("Seeking topic-partition [{}] with offset [{}]", assignedPartition, offset); consumer.seek(assignedPartition, offset); }); } TrackingTokenConsumerRebalanceListener(Consumer<K, V> consumer, Supplier<KafkaTrackingToken> tokenSupplier); }
TrackingTokenConsumerRebalanceListener implements ConsumerRebalanceListener { @Override public void onPartitionsAssigned(Collection<TopicPartition> assignedPartitions) { KafkaTrackingToken currentToken = tokenSupplier.get(); assignedPartitions.forEach(assignedPartition -> { Map<TopicPartition, Long> tokenPartitionPositions = currentToken.positions(); long offset = 0L; if (tokenPartitionPositions.containsKey(assignedPartition)) { offset = tokenPartitionPositions.get(assignedPartition) + 1; } logger.info("Seeking topic-partition [{}] with offset [{}]", assignedPartition, offset); consumer.seek(assignedPartition, offset); }); } TrackingTokenConsumerRebalanceListener(Consumer<K, V> consumer, Supplier<KafkaTrackingToken> tokenSupplier); @Override void onPartitionsRevoked(Collection<TopicPartition> partitions); @Override void onPartitionsAssigned(Collection<TopicPartition> assignedPartitions); }
TrackingTokenConsumerRebalanceListener implements ConsumerRebalanceListener { @Override public void onPartitionsAssigned(Collection<TopicPartition> assignedPartitions) { KafkaTrackingToken currentToken = tokenSupplier.get(); assignedPartitions.forEach(assignedPartition -> { Map<TopicPartition, Long> tokenPartitionPositions = currentToken.positions(); long offset = 0L; if (tokenPartitionPositions.containsKey(assignedPartition)) { offset = tokenPartitionPositions.get(assignedPartition) + 1; } logger.info("Seeking topic-partition [{}] with offset [{}]", assignedPartition, offset); consumer.seek(assignedPartition, offset); }); } TrackingTokenConsumerRebalanceListener(Consumer<K, V> consumer, Supplier<KafkaTrackingToken> tokenSupplier); @Override void onPartitionsRevoked(Collection<TopicPartition> partitions); @Override void onPartitionsAssigned(Collection<TopicPartition> assignedPartitions);  }