Refactor - use method referencing in lambdas where possible

Author: prasanth08

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

@@ -551,9 +551,7 @@ public final void stop(boolean wait) {
 					}
 				}
 				else {
-					doStop(() -> {
-						publishContainerStoppedEvent();
-					});
+					doStop(this::publishContainerStoppedEvent);
 				}
 			}
 		}

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

@@ -233,10 +233,8 @@ private void configureChildContainer(int index, KafkaMessageListenerContainer<K,
 		container.setBatchInterceptor(getBatchInterceptor());
 		container.setInterceptBeforeTx(isInterceptBeforeTx());
 		container.setListenerInfo(getListenerInfo());
-		container.setEmergencyStop(() -> {
-			stopAbnormally(() -> {
-			});
-		});
+		container.setEmergencyStop(() -> stopAbnormally(() -> {
+		}));
 		AsyncTaskExecutor exec = container.getContainerProperties().getListenerTaskExecutor();
 		if (exec == null) {
 			if ((this.executors.size() > index)) {

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

@@ -118,7 +118,7 @@ public void handleOtherException(Exception thrownException, Consumer<?, ?> consu
 			MessageListenerContainer container, boolean batchListener) {
 
 		if (this.errorHandler != null) {
-			this.errorHandler.handle(thrownException, Collections.EMPTY_LIST, consumer, container);
+			this.errorHandler.handle(thrownException, Collections.emptyList(), consumer, container);
 		}
 		else {
 			this.batchErrorHandler.handle(thrownException, EMPTY_BATCH, consumer, container, () -> { });

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

@@ -3447,7 +3447,7 @@ public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
 					ListenerConsumer.this.assignedPartitions.removeAll(partitions);
 				}
 				ListenerConsumer.this.pausedForNack.removeAll(partitions);
-				partitions.forEach(tp -> ListenerConsumer.this.lastCommits.remove(tp));
+				partitions.forEach(ListenerConsumer.this.lastCommits::remove);
 				synchronized (ListenerConsumer.this) {
 					if (ListenerConsumer.this.offsetsInThisBatch != null) {
 						partitions.forEach(tp -> {

spring-kafka/src/main/java/org/springframework/kafka/support/KafkaUtils.java

@@ -44,7 +44,7 @@ public final class KafkaUtils {
 
 	private static final ThreadLocal<Boolean> LOG_METADATA_ONLY = new ThreadLocal<>();
 
-	private static Function<ProducerRecord<?, ?>, String> prFormatter = rec -> rec.toString();
+	private static Function<ProducerRecord<?, ?>, String> prFormatter = ProducerRecord::toString;
 
 	private static Function<ConsumerRecord<?, ?>, String> crFormatter =
 			rec -> rec.topic() + "-" + rec.partition() + "@" + rec.offset();

spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingDeserializer.java

@@ -214,7 +214,7 @@ Deserializer<? extends Object> trySerdes(String key) {
 
 	@Override
 	public void close() {
-		this.delegates.values().forEach(deser -> deser.close());
+		this.delegates.values().forEach(Deserializer::close);
 	}
 
 }

spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingSerializer.java

@@ -247,7 +247,7 @@ byte[] trySerdes(Object data) {
 
 	@Override
 	public void close() {
-		this.delegates.values().forEach(ser -> ser.close());
+		this.delegates.values().forEach(Serializer::close);
 	}
 
 }