Fix getHighestOffsetRecords

* 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

Author: realcbb

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

@@ -982,18 +982,14 @@ private void commitIfNecessary() {
 		}
 
 		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();
 		}