Fix deprecated ConsumerRecords constructor calls and lossy conversion

  - Update all ConsumerRecords constructors to use the non-deprecated API by
    adding Map.of() as the second parameter
  - Fix lossy conversion in KafkaMessageListenerContainer by adding explicit
    cast to long when multiplying idleEventInterval by containerProperties.getIdleBeforeDataMultiplier()
    which is of type double.

Signed-off-by: Soby Chacko <[email protected]>

Author: sobychacko

spring-kafka-test/src/main/java/org/springframework/kafka/test/utils/KafkaTestUtils.java

@@ -389,7 +389,7 @@ public static <K, V> ConsumerRecords<K, V> getRecords(Consumer<K, V> consumer, D
 			}
 		}
 		while (count < minRecords && remaining > 0);
-		return new ConsumerRecords<>(records);
+		return new ConsumerRecords<>(records, Map.of());
 	}
 
 	/**

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java

@@ -772,7 +772,7 @@ public ConsumerRecords<K, V> receive(Collection<TopicPartitionOffset> requested,
 					consumerRecords.add(one);
 				}
 			});
-			return new ConsumerRecords<>(records);
+			return new ConsumerRecords<>(records, Map.of());
 		}
 	}
 

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

@@ -278,7 +278,7 @@ private <K, V> ConsumerRecords<K, V> seekOrRecover(Exception thrownException, @N
 			Map<TopicPartition, List<ConsumerRecord<K, V>>> remains = new HashMap<>();
 			remaining.forEach(rec -> remains.computeIfAbsent(new TopicPartition(rec.topic(), rec.partition()),
 					tp -> new ArrayList<>()).add((ConsumerRecord<K, V>) rec));
-			return new ConsumerRecords<>(remains);
+			return new ConsumerRecords<>(remains, Map.of());
 		}
 	}
 

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

@@ -1669,7 +1669,7 @@ private ConsumerRecords<K, V> doPoll() {
 				}
 				TopicPartition next = this.batchIterator.next();
 				List<ConsumerRecord<K, V>> subBatch = Objects.requireNonNull(this.lastBatch).records(next);
-				records = new ConsumerRecords<>(Collections.singletonMap(next, subBatch));
+				records = new ConsumerRecords<>(Collections.singletonMap(next, subBatch), Map.of());
 				if (!this.batchIterator.hasNext()) {
 					this.batchIterator = null;
 				}
@@ -1911,7 +1911,7 @@ private void checkIdle() {
 				long idleEventInterval2 = idleEventInterval;
 				long now = System.currentTimeMillis();
 				if (!this.receivedSome) {
-					idleEventInterval2 *= this.containerProperties.getIdleBeforeDataMultiplier();
+					idleEventInterval2 = (long) (idleEventInterval2 * this.containerProperties.getIdleBeforeDataMultiplier());
 				}
 				if (now > this.lastReceive + idleEventInterval2
 						&& now > this.lastAlertAt + idleEventInterval2) {
@@ -2649,7 +2649,7 @@ private boolean checkImmediatePause(Iterator<ConsumerRecord<K, V>> iterator) {
 							tp -> new ArrayList<ConsumerRecord<K, V>>()).add(next);
 				}
 				if (!remaining.isEmpty()) {
-					this.remainingRecords = new ConsumerRecords<>(remaining);
+					this.remainingRecords = new ConsumerRecords<>(remaining, Map.of());
 					return true;
 				}
 			}
@@ -2950,7 +2950,7 @@ private void invokeErrorHandlerBySingleRecord(FailedRecordTuple<K, V> failedReco
 											tp -> new ArrayList<>()).add(cRecord);
 				}
 				if (!records.isEmpty()) {
-					this.remainingRecords = new ConsumerRecords<>(records);
+					this.remainingRecords = new ConsumerRecords<>(records, Map.of());
 					this.pauseForPending = true;
 				}
 			}
@@ -2996,7 +2996,7 @@ private void invokeErrorHandler(final ConsumerRecord<K, V> cRecord,
 					}
 				}
 				if (!records.isEmpty()) {
-					this.remainingRecords = new ConsumerRecords<>(records);
+					this.remainingRecords = new ConsumerRecords<>(records, Map.of());
 					this.pauseForPending = true;
 				}
 			}
@@ -3614,7 +3614,7 @@ public void acknowledge(int index) {
 							tp -> new ArrayList<>()).add(record);
 				}
 				if (!offsetsToCommit.isEmpty()) {
-					processAcks(new ConsumerRecords<>(offsetsToCommit));
+					processAcks(new ConsumerRecords<>(offsetsToCommit, Map.of()));
 				}
 				this.partial = index;
 			}
@@ -3644,7 +3644,7 @@ public void nack(int index, Duration sleep) {
 					newRecords.computeIfAbsent(new TopicPartition(cRecord.topic(), cRecord.partition()),
 							tp -> new LinkedList<>()).add(cRecord);
 				}
-				processAcks(new ConsumerRecords<K, V>(newRecords));
+				processAcks(new ConsumerRecords<K, V>(newRecords, Map.of()));
 			}
 
 			@Override
@@ -3727,7 +3727,7 @@ private void removeRevocationsFromPending(Collection<TopicPartition> partitions)
 					if (!remainingParts.isEmpty()) {
 						Map<TopicPartition, List<ConsumerRecord<K, V>>> trimmed = new LinkedHashMap<>();
 						remainingParts.forEach(part -> trimmed.computeIfAbsent(part, tp -> remaining.records(tp)));
-						ListenerConsumer.this.remainingRecords = new ConsumerRecords<>(trimmed);
+						ListenerConsumer.this.remainingRecords = new ConsumerRecords<>(trimmed, Map.of());
 					}
 					else {
 						ListenerConsumer.this.remainingRecords = null;