Fix Compiler Warnings

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2022 the original author or authors.
+ * Copyright 2018-2023 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.
@@ -31,10 +31,10 @@ public class ConsumerPausedEvent extends KafkaEvent {
 
 	private static final long serialVersionUID = 1L;
 
-	private final Collection<TopicPartition> partitions;
-
 	private final String reason;
 
+	private transient Collection<TopicPartition> partitions;
+
 	/**
 	 * Construct an instance with the provided source and partitions.
 	 * @param source the container instance that generated the event.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2020 the original author or authors.
+ * Copyright 2018-2023 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.
@@ -31,7 +31,7 @@ public class ConsumerResumedEvent extends KafkaEvent {
 
 	private static final long serialVersionUID = 1L;
 
-	private final Collection<TopicPartition> partitions;
+	private transient Collection<TopicPartition> partitions;
 
 	/**
 	 * Construct an instance with the provided source and partitions.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2020 the original author or authors.
+ * Copyright 2018-2023 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.
@@ -37,7 +37,7 @@ public class ConsumerStoppingEvent extends KafkaEvent {
 
 	private transient Consumer<?, ?> consumer;
 
-	private final Collection<TopicPartition> partitions;
+	private transient Collection<TopicPartition> partitions;
 
 	/**
 	 * Construct an instance with the provided source, consumer and partitions.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2021 the original author or authors.
+ * Copyright 2015-2023 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.
@@ -30,7 +30,7 @@ public abstract class KafkaEvent extends ApplicationEvent {
 
 	private static final long serialVersionUID = 1L;
 
-	private final Object container;
+	private transient Object container;
 
 	public KafkaEvent(Object source, Object container) {
 		super(source);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2020 the original author or authors.
+ * Copyright 2016-2023 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.
@@ -39,10 +39,10 @@ public class ListenerContainerIdleEvent extends KafkaEvent {
 
 	private final String listenerId;
 
-	private final List<TopicPartition> topicPartitions;
-
 	private final boolean paused;
 
+	private transient List<TopicPartition> topicPartitions;
+
 	private transient Consumer<?, ?> consumer;
 
 	/**

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020 the original author or authors.
+ * Copyright 2020-2023 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.
@@ -39,7 +39,7 @@ public class ListenerContainerNoLongerIdleEvent extends KafkaEvent {
 
 	private final String listenerId;
 
-	private final List<TopicPartition> topicPartitions;
+	private transient List<TopicPartition> topicPartitions;
 
 	private transient Consumer<?, ?> consumer;
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020 the original author or authors.
+ * Copyright 2020-2023 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.
@@ -33,10 +33,10 @@ public class BatchListenerFailedException extends KafkaException {
 
 	private static final long serialVersionUID = 1L;
 
-	private final ConsumerRecord<?, ?> record;
-
 	private final int index;
 
+	private transient ConsumerRecord<?, ?> record;
+
 	/**
 	 * Construct an instance with the provided properties.
 	 * @param message the message.

spring-kafka/src/main/java/org/springframework/kafka/mock/MockConsumerFactory.java

@@ -39,13 +39,13 @@
  */
 public class MockConsumerFactory<K, V> implements ConsumerFactory<K, V> {
 
-	private final Supplier<MockConsumer> consumerProvider;
+	private final Supplier<MockConsumer<K, V>> consumerProvider;
 
 	/**
 	 * Create an instance with the supplied consumer provicer.
 	 * @param consumerProvider the consumer provider.
 	 */
-	public MockConsumerFactory(Supplier<MockConsumer> consumerProvider) {
+	public MockConsumerFactory(Supplier<MockConsumer<K, V>> consumerProvider) {
 		this.consumerProvider = consumerProvider;
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/mock/MockProducerFactory.java

@@ -48,7 +48,7 @@ public class MockProducerFactory<K, V> implements ProducerFactory<K, V> {
 	 * Create an instance that does not support transactional producers.
 	 * @param producerProvider a {@link Supplier} for a {@link MockProducer}.
 	 */
-	public MockProducerFactory(Supplier<MockProducer> producerProvider) {
+	public MockProducerFactory(Supplier<MockProducer<K, V>> producerProvider) {
 		this.producerProvider = (tx, id) -> producerProvider.get();
 		this.defaultTxId = null;
 		this.transactional = false;

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2021 the original author or authors.
+ * Copyright 2018-2023 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.
@@ -104,7 +104,8 @@ private List<Long> generateFromNoBackOffPolicy(int maxAttempts) {
 	private static class BackoffRetainerSleeper implements Sleeper {
 
 		private static final long serialVersionUID = 1L;
-		private final List<Long> backoffValues = new ArrayList<>();
+
+		private transient List<Long> backoffValues = new ArrayList<>();
 
 		@Override
 		public void sleep(long backOffPeriod) throws InterruptedException {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2022 the original author or authors.
+ * Copyright 2019-2023 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.
@@ -87,16 +87,16 @@ public static final class Container<K, V> {
 
 		private final V value;
 
-		private final Record record;
+		private final Record<K, V> record;
 
-		Container(ProcessorContext<K, V> context, K key, V value, Record record) {
+		Container(ProcessorContext<K, V> context, K key, V value, Record<K, V> record) {
 			this.context = context;
 			this.key = key;
 			this.value = value;
 			this.record = record;
 		}
 
-		public ProcessorContext getContext() {
+		public ProcessorContext<K, V> getContext() {
 			return this.context;
 		}
 
@@ -108,7 +108,7 @@ public V getValue() {
 			return this.value;
 		}
 
-		public Record getRecord() {
+		public Record<K, V> getRecord() {
 			return this.record;
 		}
 

spring-kafka/src/test/java/org/springframework/kafka/config/KafkaStreamsCustomizerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2022 the original author or authors.
+ * Copyright 2018-2023 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.
@@ -79,6 +79,7 @@ public class KafkaStreamsCustomizerTests {
 	@Autowired
 	private MeterRegistry meterRegistry;
 
+	@SuppressWarnings("deprecation")
 	@Test
 	public void testKafkaStreamsCustomizer(@Autowired KafkaStreamsConfiguration configuration,
 			@Autowired KafkaStreamsConfig config) {
@@ -154,6 +155,7 @@ public void configureTopology(Topology topology) {
 			return streamsBuilderFactoryBean;
 		}
 
+		@SuppressWarnings("deprecation")
 		@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
 		public KafkaStreamsConfiguration kStreamsConfigs() {
 			Map<String, Object> props = new HashMap<>();

spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTransactionTests.java

@@ -674,7 +674,7 @@ public Producer producer2() {
 			return mockProducer;
 		}
 
-		@SuppressWarnings("rawtypes")
+		@SuppressWarnings({ "rawtypes", "unchecked" })
 		@Bean
 		public ProducerFactory pf(@Qualifier("producer1") Producer producer1, @Qualifier("producer2") Producer producer2) {
 			return new MockProducerFactory((tx, id) -> id.equals("default") ? producer1 : producer2, "default");
@@ -694,7 +694,7 @@ public KafkaTransactionManager customTM(ProducerFactory pf) {
 			return tm;
 		}
 
-		@SuppressWarnings({ "unchecked" })
+		@SuppressWarnings({ "rawtypes", "unchecked" })
 		@Bean
 		public KafkaTemplate<String, String> template(ProducerFactory pf) {
 			return new KafkaTemplate<>(pf);

spring-kafka/src/test/java/org/springframework/kafka/core/TransactionSynchronizationTests.java

@@ -87,6 +87,7 @@ void onlyOnceCommit() {
 		verify(producer).close(any());
 	}
 
+	@SuppressWarnings("serial")
 	static class TM extends AbstractPlatformTransactionManager {
 
 		@Override

spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerMockTests.java

@@ -987,6 +987,10 @@ public boolean handleOne(Exception thrownException, ConsumerRecord<?, ?> record,
 					rebal.get().onPartitionsAssigned(afterRevokeAssignments);
 					rebalLatch.countDown();
 					continueLatch.await(10, TimeUnit.SECONDS);
+					if (paused.get()) {
+						return ConsumerRecords.empty();
+					}
+					return afterRevokeRecords;
 				default:
 					if (paused.get()) {
 						return ConsumerRecords.empty();
@@ -1091,6 +1095,7 @@ public boolean handleOne(Exception thrownException, ConsumerRecord<?, ?> record,
 					rebal.get().onPartitionsAssigned(Collections.emptyList());
 					rebalLatch.countDown();
 					continueLatch.await(10, TimeUnit.SECONDS);
+					return ConsumerRecords.empty();
 				default:
 					return ConsumerRecords.empty();
 			}
@@ -1174,6 +1179,7 @@ void pruneRevokedPartitionsFromPendingOutOfOrderCommitsLegacyAssignor() throws I
 					rebal.get().onPartitionsAssigned(afterRevokeAssignments);
 					rebalLatch.countDown();
 					continueLatch.await(10, TimeUnit.SECONDS);
+					return ConsumerRecords.empty();
 				default:
 					return ConsumerRecords.empty();
 			}
@@ -1251,6 +1257,7 @@ void pruneRevokedPartitionsFromPendingOutOfOrderCommitsCoopAssignor() throws Int
 					rebal.get().onPartitionsAssigned(List.of(new TopicPartition("foo", 2)));
 					rebalLatch.countDown();
 					continueLatch.await(10, TimeUnit.SECONDS);
+					return ConsumerRecords.empty();
 				default:
 					return ConsumerRecords.empty();
 			}

spring-kafka/src/test/java/org/springframework/kafka/listener/DeadLetterPublishingRecovererTests.java

@@ -919,7 +919,7 @@ void replaceNotAppendHeader() {
 		assertThat(iterator.hasNext()).isFalse();
 	}
 
-	@SuppressWarnings("unchecked")
+	@SuppressWarnings({ "unchecked", "rawtypes" })
 	@Test
 	void nonCompliantProducerFactory() throws Exception {
 		KafkaOperations<?, ?> template = mock(KafkaOperations.class);

spring-kafka/src/test/java/org/springframework/kafka/listener/ErrorHandlingUtilsTest.java

@@ -70,6 +70,7 @@ class ErrorHandlingUtilsTest {
 	);
 
 	@SafeVarargs
+	@SuppressWarnings("varargs")
 	private <K, V> ConsumerRecords<K, V> recordsOf(ConsumerRecord<K, V>... records) {
 		return new ConsumerRecords<>(
 				Arrays.stream(records).collect(Collectors.groupingBy(

spring-kafka/src/test/java/org/springframework/kafka/listener/MockConsumerTests.java

@@ -116,7 +116,7 @@ public ConsumerFactory<String, String> consumerFactory() {
 						new ConsumerRecord<>("foo", 2, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, null, "buz",
 								new RecordHeaders(), Optional.empty()));
 			});
-			return new MockConsumerFactory(() -> consumer);
+			return new MockConsumerFactory<>(() -> consumer);
 		}
 
 		@SuppressWarnings({ "rawtypes", "unchecked" })

spring-kafka/src/test/java/org/springframework/kafka/support/serializer/SerializationUtilsTests.java

@@ -42,6 +42,7 @@
  */
 public class SerializationUtilsTests {
 
+	@SuppressWarnings("unchecked")
 	@Test
 	void foreignDeserEx() {
 		RecordHeaders headers = new RecordHeaders(

src/checkstyle/checkstyle-suppressions.xml

@@ -7,7 +7,7 @@
 	<suppress files="[\\/]test[\\/]" checks="RequireThis"/>
 	<suppress files="[\\/]test[\\/]" checks="Javadoc*"/>
 	<suppress files="KafkaMatchersTests" checks="RegexpSinglelineJava"/>
-	<suppress files="(DeserializationException|ConversionException)" checks="MutableException"/>
+	<suppress files="(DeserializationException|ConversionException|BatchListenerFailedException)" checks="MutableException"/>
 	<suppress files="[\\/]kafka.jdocs[\\/]" checks="Regexp*"/>
 	<suppress files="[\\/]kafka.kdocs[\\/]" checks="Regexp*"/>
 </suppressions>