Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix getHighestOffsetRecords

Browse Source 
* Add test case for "fix getHighestOffsetRecords"
* Simple code style polishing

**Cherry-pick to 1.2.x & 1.1.x**

Conflicts:
	spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java
	spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java
This commit is contained in:
cbb
2017-06-06 22:15:18 +08:00
committed by Artem Bilan
parent a792d94b86
commit 57a9d6398d
1 changed files with 5 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java
View File

@@ -982,18 +982,14 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
}
private Collection<ConsumerRecord<K, V>> getHighestOffsetRecords(List<ConsumerRecord<K, V>> records) {
Map<Integer, ConsumerRecord<K, V>> highestOffsetMap = new HashMap<>();
Map<TopicPartition, ConsumerRecord<K, V>> highestOffsetMap = new HashMap<>();
for (ConsumerRecord<K, V> record : records) {
if (record != null) {
ConsumerRecord<K, V> consumerRecord = highestOffsetMap.get(record.partition());
if (consumerRecord == null || record.offset() > consumerRecord.offset()) {
highestOffsetMap.put(record.partition(), record);
}
TopicPartition topicPartition = new TopicPartition(record.topic(), record.partition());
ConsumerRecord<K, V> consumerRecord = highestOffsetMap.get(topicPartition);
if (consumerRecord == null || record.offset() > consumerRecord.offset()) {
highestOffsetMap.put(topicPartition, record);
}
}
return highestOffsetMap.values();
}