* Add method consumerWakeIfNecessary() to reuse consumer.wakeIfNecessary().

* Add method `getTxProducer()`.
* Optimize `determineListenerType()`.
* Replace iterator to for loop or `forEachRemaining()`.
* Change `OffsetMetadata` type to record.

Author: Wzy19930507

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

@@ -18,11 +18,11 @@
 
 import java.nio.ByteBuffer;
 import java.time.Duration;
-import java.util.AbstractMap.SimpleEntry;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
@@ -305,8 +305,7 @@ public boolean isContainerPaused() {
 
 	@Override
 	public boolean isPartitionPaused(TopicPartition topicPartition) {
-		return this.listenerConsumer != null && this.listenerConsumer
-				.isPartitionPaused(topicPartition);
+		return this.listenerConsumer != null && this.listenerConsumer.isPartitionPaused(topicPartition);
 	}
 
 	@Override
@@ -317,33 +316,28 @@ public boolean isInExpectedState() {
 	@Override
 	public void enforceRebalance() {
 		this.thisOrParentContainer.enforceRebalanceRequested.set(true);
-		KafkaMessageListenerContainer<K, V>.ListenerConsumer consumer = this.listenerConsumer;
-		if (consumer != null) {
-			consumer.wakeIfNecessary();
-		}
+		consumerWakeIfNecessary();
 	}
 
 	@Override
 	public void pause() {
 		super.pause();
-		KafkaMessageListenerContainer<K, V>.ListenerConsumer consumer = this.listenerConsumer;
-		if (consumer != null) {
-			consumer.wakeIfNecessary();
-		}
+		consumerWakeIfNecessary();
 	}
 
 	@Override
 	public void resume() {
 		super.resume();
-		KafkaMessageListenerContainer<K, V>.ListenerConsumer consumer = this.listenerConsumer;
-		if (consumer != null) {
-			consumer.wakeIfNecessary();
-		}
+		consumerWakeIfNecessary();
 	}
 
 	@Override
 	public void resumePartition(TopicPartition topicPartition) {
 		super.resumePartition(topicPartition);
+		consumerWakeIfNecessary();
+	}
+
+	private void consumerWakeIfNecessary() {
 		KafkaMessageListenerContainer<K, V>.ListenerConsumer consumer = this.listenerConsumer;
 		if (consumer != null) {
 			consumer.wakeIfNecessary();
@@ -422,15 +416,11 @@ private void checkAckMode(ContainerProperties containerProperties) {
 	}
 
 	private ListenerType determineListenerType(GenericMessageListener<?> listener) {
-		ListenerType listenerType = ListenerUtils.determineListenerType(listener);
-		if (listener instanceof DelegatingMessageListener) {
-			Object delegating = listener;
-			while (delegating instanceof DelegatingMessageListener<?> dml) {
-				delegating = dml.getDelegate();
-			}
-			listenerType = ListenerUtils.determineListenerType(delegating);
+		Object delegating = listener;
+		while (delegating instanceof DelegatingMessageListener<?> dml) {
+			delegating = dml.getDelegate();
 		}
-		return listenerType;
+		return ListenerUtils.determineListenerType(delegating);
 	}
 
 	@Override
@@ -1586,7 +1576,7 @@ private void fixTxOffsetsIfNeeded() {
 					this.lastCommits.forEach((tp, oamd) -> {
 						long position = this.consumer.position(tp);
 						Long saved = this.savedPositions.get(tp);
-						if (saved != null && saved.longValue() != position) {
+						if (saved != null && saved != position) {
 							this.logger.debug(() -> "Skipping TX offset correction - seek(s) have been performed; "
 									+ "saved: " + this.savedPositions + ", "
 									+ "committed: " + oamd + ", "
@@ -1609,9 +1599,7 @@ private void fixTxOffsetsIfNeeded() {
 						}
 						else {
 							this.transactionTemplate.executeWithoutResult(status -> {
-								doSendOffsets(((KafkaResourceHolder) TransactionSynchronizationManager
-										.getResource(this.kafkaTxManager.getProducerFactory()))
-										.getProducer(), toFix);
+								doSendOffsets(getTxProducer(), toFix);
 							});
 						}
 					}
@@ -2088,7 +2076,7 @@ private synchronized void ackInOrder(ConsumerRecord<K, V> cRecord) {
 					offs.remove(0);
 					ConsumerRecord<K, V> recordToAck = cRecord;
 					if (!deferred.isEmpty()) {
-						Collections.sort(deferred, (a, b) -> Long.compare(a.offset(), b.offset()));
+						deferred.sort(Comparator.comparingLong(ConsumerRecord::offset));
 						while (!ObjectUtils.isEmpty(deferred) && deferred.get(0).offset() == recordToAck.offset() + 1) {
 							recordToAck = deferred.remove(0);
 							offs.remove(0);
@@ -2195,9 +2183,7 @@ private void invokeBatchListenerInTx(final ConsumerRecords<K, V> records,
 					@Override
 					public void doInTransactionWithoutResult(TransactionStatus s) {
 						if (ListenerConsumer.this.kafkaTxManager != null) {
-							ListenerConsumer.this.producer = ((KafkaResourceHolder) TransactionSynchronizationManager
-									.getResource(ListenerConsumer.this.kafkaTxManager.getProducerFactory()))
-										.getProducer(); // NOSONAR nullable
+							ListenerConsumer.this.producer = getTxProducer();
 						}
 						RuntimeException aborted = doInvokeBatchListener(records, recordList);
 						if (aborted != null) {
@@ -2254,10 +2240,9 @@ protected void doInTransactionWithoutResult(TransactionStatus status) {
 		}
 
 		private List<ConsumerRecord<K, V>> createRecordList(final ConsumerRecords<K, V> records) {
-			Iterator<ConsumerRecord<K, V>> iterator = records.iterator();
 			List<ConsumerRecord<K, V>> list = new LinkedList<>();
-			while (iterator.hasNext()) {
-				list.add(iterator.next());
+			for (ConsumerRecord<K, V> record : records) {
+				list.add(record);
 			}
 			return list;
 		}
@@ -2324,9 +2309,7 @@ private void commitOffsetsIfNeededAfterHandlingError(final ConsumerRecords<K, V>
 					|| this.producer != null) {
 				if (this.remainingRecords != null) {
 					ConsumerRecord<K, V> firstUncommitted = this.remainingRecords.iterator().next();
-					Iterator<ConsumerRecord<K, V>> it = records.iterator();
-					while (it.hasNext()) {
-						ConsumerRecord<K, V> next = it.next();
+					for (ConsumerRecord<K, V> next : records) {
 						if (!next.equals(firstUncommitted)) {
 							this.acks.add(next);
 						}
@@ -2433,7 +2416,7 @@ private void invokeBatchOnMessageWithRecordsOrList(final ConsumerRecords<K, V> r
 			ConsumerRecords<K, V> records = recordsArg;
 			List<ConsumerRecord<K, V>> recordList = recordListArg;
 			if (this.listenerinfo != null) {
-				records.iterator().forEachRemaining(this::listenerInfo);
+				records.forEach(this::listenerInfo);
 			}
 			if (this.batchInterceptor != null) {
 				records = this.batchInterceptor.intercept(recordsArg, this.consumer);
@@ -2516,7 +2499,6 @@ private void invokeRecordListener(final ConsumerRecords<K, V> records) {
 		 * Invoke the listener with each record in a separate transaction.
 		 * @param records the records.
 		 */
-		@SuppressWarnings(RAWTYPES) // NOSONAR complexity
 		private void invokeRecordListenerInTx(final ConsumerRecords<K, V> records) {
 			Iterator<ConsumerRecord<K, V>> iterator = records.iterator();
 			while (iterator.hasNext()) {
@@ -2561,9 +2543,7 @@ private void invokeInTransaction(Iterator<ConsumerRecord<K, V>> iterator, final
 				@Override
 				public void doInTransactionWithoutResult(TransactionStatus s) {
 					if (ListenerConsumer.this.kafkaTxManager != null) {
-						ListenerConsumer.this.producer = ((KafkaResourceHolder) TransactionSynchronizationManager
-								.getResource(ListenerConsumer.this.kafkaTxManager.getProducerFactory()))
-										.getProducer(); // NOSONAR
+						ListenerConsumer.this.producer = getTxProducer();
 					}
 					RuntimeException aborted = doInvokeRecordListener(cRecord, iterator);
 					if (aborted != null) {
@@ -2579,9 +2559,7 @@ private void recordAfterRollback(Iterator<ConsumerRecord<K, V>> iterator, final
 
 			List<ConsumerRecord<K, V>> unprocessed = new ArrayList<>();
 			unprocessed.add(cRecord);
-			while (iterator.hasNext()) {
-				unprocessed.add(iterator.next());
-			}
+			iterator.forEachRemaining(unprocessed::add);
 			@SuppressWarnings(UNCHECKED)
 			AfterRollbackProcessor<K, V> afterRollbackProcessorToUse =
 					(AfterRollbackProcessor<K, V>) getAfterRollbackProcessor();
@@ -2639,11 +2617,10 @@ private void doInvokeWithRecords(final ConsumerRecords<K, V> records) {
 		private boolean checkImmediatePause(Iterator<ConsumerRecord<K, V>> iterator) {
 			if (isPauseRequested() && this.pauseImmediate) {
 				Map<TopicPartition, List<ConsumerRecord<K, V>>> remaining = new LinkedHashMap<>();
-				while (iterator.hasNext()) {
-					ConsumerRecord<K, V> next = iterator.next();
+				iterator.forEachRemaining(next -> {
 					remaining.computeIfAbsent(new TopicPartition(next.topic(), next.partition()),
-							tp -> new ArrayList<ConsumerRecord<K, V>>()).add(next);
-				}
+							tp -> new ArrayList<>()).add(next);
+				});
 				if (!remaining.isEmpty()) {
 					this.remainingRecords = new ConsumerRecords<>(remaining);
 					return true;
@@ -2712,9 +2689,7 @@ private void handleNack(final ConsumerRecords<K, V> records, final ConsumerRecor
 				processCommits();
 			}
 			List<ConsumerRecord<?, ?>> list = new ArrayList<>();
-			Iterator<ConsumerRecord<K, V>> iterator2 = records.iterator();
-			while (iterator2.hasNext()) {
-				ConsumerRecord<K, V> next = iterator2.next();
+			for (ConsumerRecord<K, V> next : records) {
 				if (!list.isEmpty() || recordsEqual(cRecord, next)) {
 					list.add(next);
 				}
@@ -2755,6 +2730,13 @@ private void pauseForNackSleep() {
 			this.nackSleepDurationMillis = -1;
 		}
 
+		@SuppressWarnings(RAWTYPES)
+		private Producer<?, ?> getTxProducer() {
+			return ((KafkaResourceHolder) TransactionSynchronizationManager
+					.getResource(ListenerConsumer.this.kafkaTxManager.getProducerFactory()))
+					.getProducer(); // NOSONAR
+		}
+
 		/**
 		 * Actually invoke the listener.
 		 * @param cRecord the record.
@@ -2911,9 +2893,7 @@ private void invokeErrorHandler(final ConsumerRecord<K, V> cRecord,
 				}
 				List<ConsumerRecord<?, ?>> records = new ArrayList<>();
 				records.add(cRecord);
-				while (iterator.hasNext()) {
-					records.add(iterator.next());
-				}
+				iterator.forEachRemaining(records::add);
 				this.commonErrorHandler.handleRemaining(rte, records, this.consumer,
 						KafkaMessageListenerContainer.this.thisOrParentContainer);
 			}
@@ -2929,12 +2909,9 @@ private void invokeErrorHandler(final ConsumerRecord<K, V> cRecord,
 				Map<TopicPartition, List<ConsumerRecord<K, V>>> records = new LinkedHashMap<>();
 				if (!handled) {
 					records.computeIfAbsent(new TopicPartition(cRecord.topic(), cRecord.partition()),
-							tp -> new ArrayList<ConsumerRecord<K, V>>()).add(cRecord);
-					while (iterator.hasNext()) {
-						ConsumerRecord<K, V> next = iterator.next();
-						records.computeIfAbsent(new TopicPartition(next.topic(), next.partition()),
-								tp -> new ArrayList<ConsumerRecord<K, V>>()).add(next);
-					}
+							tp -> new ArrayList<>()).add(cRecord);
+					iterator.forEachRemaining(next -> records.computeIfAbsent(
+							new TopicPartition(next.topic(), next.partition()), tp -> new ArrayList<>()).add(next));
 				}
 				if (!records.isEmpty()) {
 					this.remainingRecords = new ConsumerRecords<>(records);
@@ -3201,9 +3178,7 @@ private void initPartitionsIfNeeded() {
 			doInitialSeeks(partitions, beginnings, ends);
 			if (this.consumerSeekAwareListener != null) {
 				this.consumerSeekAwareListener.onPartitionsAssigned(this.definedPartitions.keySet().stream()
-							.map(tp -> new SimpleEntry<>(tp, this.consumer.position(tp)))
-							.collect(Collectors.toMap(entry -> entry.getKey(), entry -> entry.getValue())),
-						this.seekCallback);
+						.collect(Collectors.toMap(tp -> tp, this.consumer::position)), this.seekCallback);
 			}
 		}
 
@@ -3884,20 +3859,13 @@ private Long computeBackwardWhereTo(long offset, boolean toCurrent, TopicPartiti
 	}
 
 
-	private static final class OffsetMetadata {
-
-		final Long offset; // NOSONAR
-
-		final boolean relativeToCurrent; // NOSONAR
-
-		final SeekPosition seekPosition; // NOSONAR
-
-		OffsetMetadata(Long offset, boolean relativeToCurrent, SeekPosition seekPosition) {
-			this.offset = offset;
-			this.relativeToCurrent = relativeToCurrent;
-			this.seekPosition = seekPosition;
-		}
-
+	/**
+	 * Offset metadata record.
+	 * @param offset            current offset.
+	 * @param relativeToCurrent relative to current.
+	 * @param seekPosition      seek position strategy.
+	 */
+	private record OffsetMetadata(Long offset, boolean relativeToCurrent, SeekPosition seekPosition) {
 	}
 
 	private class StopCallback implements BiConsumer<Object, Throwable> {