turns the simplest approach works best

Author: janeklb

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

@@ -2239,15 +2239,9 @@ protected void doInTransactionWithoutResult(TransactionStatus status) {
 		}
 
 		private List<ConsumerRecord<K, V>> createRecordList(final ConsumerRecords<K, V> records) {
-			Iterator<ConsumerRecord<K, V>> iterator = records.iterator();
-			@SuppressWarnings("unchecked") ConsumerRecord<K, V>[] recordsArray =
-					(ConsumerRecord<K, V>[]) Array.newInstance(ConsumerRecord.class, records.count());
-			int index = 0;
-			while (iterator.hasNext()) {
-				recordsArray[index] = iterator.next();
-				index += 1;
-			}
-			return Arrays.asList(recordsArray);
+			List<ConsumerRecord<K, V>> recordList = new ArrayList<>(records.count());
+			records.forEach(recordList::add);
+			return recordList;
 		}
 
 		/**

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RecordFilterStrategy.java

@@ -16,6 +16,7 @@
 
 package org.springframework.kafka.listener.adapter;
 
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 
@@ -51,10 +52,9 @@ public interface RecordFilterStrategy<K, V> {
 	 * @return the filtered records.
 	 * @since 2.8
 	 */
-	@SuppressWarnings("unchecked")
 	default List<ConsumerRecord<K, V>> filterBatch(List<ConsumerRecord<K, V>> records) {
-		var recordsArray = records.stream().filter(record -> !this.filter(record)).toArray(ConsumerRecord[]::new);
-		return Arrays.asList(recordsArray);
+		records.removeIf(this::filter);
+		return records;
 	}
 
 	/**