Add checkstyle rule on empty-line-separater

* Update checkstyle.xml
* Fix checkstyle violations
* Fix test class checkstyle
* Add field also.

Author: JooHyukKim

spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaKraftBroker.java

@@ -204,7 +204,6 @@ public void afterPropertiesSet() {
 		}
 	}
 
-
 	private void start() {
 		if (this.cluster != null) {
 			return;

spring-kafka-test/src/main/java/org/springframework/kafka/test/assertj/KafkaConditions.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2024 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -90,7 +90,6 @@ public static <K, V> Condition<ConsumerRecord<K, V>> keyValue(K key, V value) {
 		return new ConsumerRecordPartitionCondition(partition);
 	}
 
-
 	public static class ConsumerRecordKeyCondition<K> extends Condition<ConsumerRecord<K, ?>> {
 
 		private final K key;

spring-kafka-test/src/main/java/org/springframework/kafka/test/condition/EmbeddedKafkaCondition.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2023 the original author or authors.
+ * Copyright 2019-2024 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -211,7 +211,6 @@ private Store getParentStore(ExtensionContext context) {
 		return parent.getStore(Namespace.create(getClass(), parent));
 	}
 
-
 	public static EmbeddedKafkaBroker getBroker() {
 		return BROKERS.get();
 	}

spring-kafka-test/src/test/java/org/springframework/kafka/test/context/EmbeddedKafkaContextCustomizerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2023 the original author or authors.
+ * Copyright 2017-2024 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -48,15 +48,13 @@ void beforeEachTest() {
 				AnnotationUtils.findAnnotation(SecondTestWithEmbeddedKafka.class, EmbeddedKafka.class);
 	}
 
-
 	@Test
 	void testHashCode() {
 		assertThat(new EmbeddedKafkaContextCustomizer(annotationFromFirstClass).hashCode()).isNotEqualTo(0);
 		assertThat(new EmbeddedKafkaContextCustomizer(annotationFromFirstClass).hashCode())
 				.isEqualTo(new EmbeddedKafkaContextCustomizer(annotationFromSecondClass).hashCode());
 	}
 
-
 	@Test
 	void testEquals() {
 		assertThat(new EmbeddedKafkaContextCustomizer(annotationFromFirstClass))
@@ -93,7 +91,6 @@ void testMulti() {
 				.matches("127.0.0.1:[0-9]+,127.0.0.1:[0-9]+");
 	}
 
-
 	@EmbeddedKafka(kraft = false)
 	private static final class TestWithEmbeddedKafka {
 

spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java

@@ -1190,7 +1190,6 @@ private MessageHandlerMethodFactory createDefaultMessageHandlerMethodFactory() {
 
 	private record BytesToStringConverter(Charset charset) implements Converter<byte[], String> {
 
-
 		@Override
 		public String convert(byte[] source) {
 			return new String(source, this.charset);

spring-kafka/src/main/java/org/springframework/kafka/annotation/RetryTopicConfigurationProvider.java

@@ -37,7 +37,6 @@
 import org.springframework.kafka.retrytopic.RetryTopicConfiguration;
 import org.springframework.lang.Nullable;
 
-
 /**
  *
  * Attempts to provide an instance of

spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointRegistry.java

@@ -361,7 +361,6 @@ else if (endpoint instanceof MethodKafkaListenerEndpoint<?, ?> mkle) {
 		return listenerContainer;
 	}
 
-
 	@Override
 	public void destroy() {
 		for (MessageListenerContainer listenerContainer : getListenerContainers()) {
@@ -419,7 +418,6 @@ public boolean isRunning() {
 		return this.running;
 	}
 
-
 	@Override
 	public void onApplicationEvent(ContextRefreshedEvent event) {
 		if (event.getApplicationContext().equals(this.applicationContext)) {
@@ -439,7 +437,6 @@ private void startIfNecessary(MessageListenerContainer listenerContainer) {
 		}
 	}
 
-
 	private static final class AggregatingCallback implements Runnable {
 
 		private final AtomicInteger count;

spring-kafka/src/main/java/org/springframework/kafka/config/MultiMethodKafkaListenerEndpoint.java

@@ -61,7 +61,6 @@ public MultiMethodKafkaListenerEndpoint(List<Method> methods, @Nullable Method d
 		setBean(bean);
 	}
 
-
 	/**
 	 * Get a method list.
 	 * @return the method list.

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

@@ -252,7 +252,6 @@ public void setValueDeserializerSupplier(Supplier<Deserializer<V>> valueDeserial
 		this.valueDeserializerSupplier = valueDeserializerSupplier(valueDeserializerSupplier);
 	}
 
-
 	/**
 	 * Set to false (default true) to prevent programmatically provided deserializers (via
 	 * constructor or setters) from being configured using the producer configuration,

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

@@ -68,8 +68,6 @@
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
-
-
 /**
  * The {@link ProducerFactory} implementation for a {@code singleton} shared {@link Producer} instance.
  * <p>
@@ -567,7 +565,6 @@ public ProducerFactory<K, V> copyWithConfigurationOverride(Map<String, Object> o
 		return newFactory;
 	}
 
-
 	/**
 	 * Ensures that the returned properties map contains a transaction id prefix.
 	 * The {@link org.springframework.kafka.core.DefaultKafkaProducerFactory}

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

@@ -308,7 +308,6 @@ public void setMessagingConverter(SmartMessageConverter messageConverter) {
 		((MessagingMessageConverter) this.messageConverter).setMessagingConverter(messageConverter);
 	}
 
-
 	@Override
 	public boolean isTransactional() {
 		return this.transactional;
@@ -590,7 +589,6 @@ public CompletableFuture<SendResult<K, V>> send(Message<?> message) {
 		return observeSend((ProducerRecord<K, V>) producerRecord);
 	}
 
-
 	@Override
 	public List<PartitionInfo> partitionsFor(String topic) {
 		Producer<K, V> producer = getTheProducer();

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2022 the original author or authors.
+ * Copyright 2020-2024 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -65,7 +65,6 @@ public MicrometerProducerListener(MeterRegistry meterRegistry, List<Tag> tags) {
 		this.tags = tags;
 	}
 
-
 	@Override
 	public synchronized void producerAdded(String id, Producer<K, V> producer) {
 		if (!this.metrics.containsKey(id)) {

spring-kafka/src/main/java/org/springframework/kafka/event/KafkaEvent.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2023 the original author or authors.
+ * Copyright 2015-2024 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -19,7 +19,6 @@
 import org.springframework.context.ApplicationEvent;
 import org.springframework.util.Assert;
 
-
 /**
  * Base class for events.
  *

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2019 the original author or authors.
+ * Copyright 2015-2024 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -40,7 +40,6 @@
 @FunctionalInterface
 public interface BatchMessageListener<K, V> extends GenericMessageListener<List<ConsumerRecord<K, V>>> {
 
-
 	/**
 	 * Listener receives the original {@link ConsumerRecords} object instead of a
 	 * list of {@link ConsumerRecord}.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021-2023 the original author or authors.
+ * Copyright 2021-2024 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -83,7 +83,6 @@ public void handleOtherException(Exception thrownException, Consumer<?, ?> consu
 		stopContainer(container, thrownException);
 	}
 
-
 	@Override
 	public void handleRemaining(Exception thrownException, List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer,
 			MessageListenerContainer container) {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021-2023 the original author or authors.
+ * Copyright 2021-2024 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -47,7 +47,6 @@ public void setAckAfterHandle(boolean ackAfterHandle) {
 		this.ackAfterHandle = ackAfterHandle;
 	}
 
-
 	@Override
 	public boolean handleOne(Exception thrownException, ConsumerRecord<?, ?> record, Consumer<?, ?> consumer,
 			MessageListenerContainer container) {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2023 the original author or authors.
+ * Copyright 2017-2024 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -34,7 +34,6 @@
  */
 public interface ConsumerAwareRebalanceListener extends ConsumerRebalanceListener {
 
-
 	/**
 	 * The same as {@link #onPartitionsRevoked(Collection)} with the additional consumer
 	 * parameter. It is invoked by the container before any pending offsets are committed.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021 the original author or authors.
+ * Copyright 2021-2024 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -32,7 +32,6 @@
 @FunctionalInterface
 public interface ConsumerAwareRecordRecoverer extends ConsumerRecordRecoverer {
 
-
 	@Override
 	default void accept(ConsumerRecord<?, ?> record, Exception exception) {
 		accept(record, null, exception);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2023 the original author or authors.
+ * Copyright 2018-2024 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -1364,6 +1364,7 @@ public ExceptionInfo exceptionCauseFqcn(String exceptionCauseFqcn) {
 					this.exceptionCauseFqcn = exceptionCauseFqcn;
 					return this;
 				}
+
 				/**
 				 * Sets the name of the header that will be used to store the keyExceptionMessage
 				 * of the original record.

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

@@ -137,7 +137,6 @@
 import io.micrometer.observation.Observation;
 import io.micrometer.observation.ObservationRegistry;
 
-
 /**
  * Single-threaded Message listener container using the Java {@link Consumer} supporting
  * auto-partition assignment or user-configured assignment.
@@ -593,7 +592,6 @@ public String toString() {
 				+ "]";
 	}
 
-
 	private final class ListenerConsumer implements SchedulingAwareRunnable, ConsumerSeekCallback {
 
 		private static final String COMMITTING = "Committing: ";

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2023 the original author or authors.
+ * Copyright 2016-2024 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -35,7 +35,6 @@
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.util.Assert;
 
-
 /**
  * A {@link org.springframework.kafka.listener.MessageListener MessageListener}
  * adapter that invokes a configurable {@link HandlerAdapter}; used when the factory is

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

@@ -47,7 +47,6 @@
 import org.springframework.util.Assert;
 import org.springframework.validation.Validator;
 
-
 /**
  * Delegates to an {@link InvocableHandlerMethod} based on the message payload type.
  * Matches a single, non-annotated parameter or one that is annotated with

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2023 the original author or authors.
+ * Copyright 2002-2024 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -28,7 +28,6 @@
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 
-
 /**
  * A {@link org.springframework.kafka.listener.MessageListener MessageListener}
  * adapter that invokes a configurable {@link HandlerAdapter}; used when the factory is

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DefaultDestinationTopicResolver.java

@@ -39,7 +39,6 @@
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
-
 /**
  *
  * Default implementation of the {@link DestinationTopicResolver} interface.

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DestinationTopic.java

@@ -148,6 +148,7 @@ public static class Properties {
 
 		@Nullable
 		private final Boolean autoStartDltHandler;
+
 		/**
 		 * Create an instance with the provided properties with the DLT container starting
 		 * automatically (if the container factory is so configured).

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfiguration.java

@@ -91,7 +91,9 @@ public Integer getConcurrency() {
 	static class TopicCreation {
 
 		private final boolean shouldCreateTopics;
+
 		private final int numPartitions;
+
 		private final short replicationFactor;
 
 		TopicCreation(@Nullable Boolean shouldCreate, @Nullable Integer numPartitions, @Nullable Short replicationFactor) {

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationBuilder.java

@@ -72,7 +72,6 @@ public class RetryTopicConfigurationBuilder {
 
 	private RetryTopicConfiguration.TopicCreation topicCreationConfiguration = new RetryTopicConfiguration.TopicCreation();
 
-
 	private ConcurrentKafkaListenerContainerFactory<?, ?> listenerContainerFactory;
 
 	@Nullable
@@ -401,7 +400,6 @@ public RetryTopicConfigurationBuilder customBackoff(SleepingBackOffPolicy<?> bac
 		return this;
 	}
 
-
 	/**
 	 * Configure a {@link FixedBackOffPolicy}.
 	 * @param interval the interval.

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurer.java

@@ -43,7 +43,6 @@
 import org.springframework.lang.NonNull;
 import org.springframework.lang.Nullable;
 
-
 /**
  *
  * <p>Configures main, retry and DLT topics based on a main endpoint and provided

spring-kafka/src/main/java/org/springframework/kafka/streams/KafkaStreamsMicrometerListener.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2022 the original author or authors.
+ * Copyright 2020-2024 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -64,7 +64,6 @@ public KafkaStreamsMicrometerListener(MeterRegistry meterRegistry, List<Tag> tag
 		this.tags = tags;
 	}
 
-
 	@Override
 	public synchronized void streamsAdded(String id, KafkaStreams kafkaStreams) {
 		if (!this.metrics.containsKey(id)) {