Log Uncommitted After Rebalance

When commits a retryable, some partitions may have been assigned to
another instance, in which case, those offsets can't be committed.
Log the offsets that could not be committed at WARN level.

Changed log level to improve troubleshooting

Changed log level to improve troubleshooting

Changed log level to improve troubleshooting

Fixed formatting

Author: rhryn

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

@@ -1550,7 +1550,15 @@ private void checkRebalanceCommits() {
 				Map<TopicPartition, OffsetAndMetadata> commits = this.commitsDuringRebalance.entrySet()
 						.stream()
 						.filter(entry -> this.assignedPartitions.contains(entry.getKey()))
-						.collect(Collectors.toMap(entry -> entry.getKey(), entry -> entry.getValue()));
+						.collect(Collectors.toMap(Entry::getKey, Entry::getValue));
+
+				Map<TopicPartition, OffsetAndMetadata> uncommitted = this.commitsDuringRebalance.entrySet()
+						.stream()
+						.filter(entry -> !this.assignedPartitions.contains(entry.getKey()))
+						.collect(Collectors.toMap(Entry::getKey, Entry::getValue));
+				this.logger.warn(() -> "These offsets could not be committed; partition(s) lost during rebalance: "
+								+ uncommitted);
+
 				this.commitsDuringRebalance.clear();
 				this.logger.debug(() -> "Commit list: " + commits);
 				commitSync(commits);