Fix KafkaItemReader ClassCastException during ExecutionContext deserialization

Signed-off-by: Hyunwoo Jung <[email protected]>

Author: noojung

spring-batch-infrastructure/src/main/java/org/springframework/batch/infrastructure/item/kafka/KafkaItemReader.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2023 the original author or authors.
+ * Copyright 2019-2025 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.
@@ -49,6 +49,7 @@
  *
  * @author Mathieu Ouellet
  * @author Mahmoud Ben Hassine
+ * @author Hyunwoo Jung
  * @since 4.2
  */
 public class KafkaItemReader<K, V> extends AbstractItemStreamItemReader<V> {
@@ -57,6 +58,8 @@ public class KafkaItemReader<K, V> extends AbstractItemStreamItemReader<V> {
 
 	private static final long DEFAULT_POLL_TIMEOUT = 30L;
 
+	private final String topicName;
+
 	private final List<TopicPartition> topicPartitions;
 
 	private @Nullable Map<TopicPartition, Long> partitionOffsets;
@@ -111,6 +114,7 @@ public KafkaItemReader(Properties consumerProperties, String topicName, List<Int
 				ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG + " property must be provided");
 		this.consumerProperties = consumerProperties;
 		Assert.hasLength(topicName, "Topic name must not be null or empty");
+		this.topicName = topicName;
 		Assert.isTrue(!partitions.isEmpty(), "At least one partition must be provided");
 		this.topicPartitions = new ArrayList<>();
 		for (Integer partition : partitions) {
@@ -175,10 +179,10 @@ public void open(ExecutionContext executionContext) {
 			}
 		}
 		if (this.saveState && executionContext.containsKey(TOPIC_PARTITION_OFFSETS)) {
-			Map<TopicPartition, Long> offsets = (Map<TopicPartition, Long>) executionContext
-				.get(TOPIC_PARTITION_OFFSETS);
-			for (Map.Entry<TopicPartition, Long> entry : offsets.entrySet()) {
-				this.partitionOffsets.put(entry.getKey(), entry.getValue() == 0 ? 0 : entry.getValue() + 1);
+			Map<String, Long> offsets = (Map<String, Long>) executionContext.get(TOPIC_PARTITION_OFFSETS);
+			for (Map.Entry<String, Long> entry : offsets.entrySet()) {
+				this.partitionOffsets.put(new TopicPartition(this.topicName, Integer.parseInt(entry.getKey())),
+						entry.getValue() == 0 ? 0 : entry.getValue() + 1);
 			}
 		}
 		this.kafkaConsumer.assign(this.topicPartitions);
@@ -205,7 +209,11 @@ public void open(ExecutionContext executionContext) {
 	@Override
 	public void update(ExecutionContext executionContext) {
 		if (this.saveState) {
-			executionContext.put(TOPIC_PARTITION_OFFSETS, new HashMap<>(this.partitionOffsets));
+			Map<String, Long> offsets = new HashMap<>();
+			for (Map.Entry<TopicPartition, Long> entry : this.partitionOffsets.entrySet()) {
+				offsets.put(String.valueOf(entry.getKey().partition()), entry.getValue());
+			}
+			executionContext.put(TOPIC_PARTITION_OFFSETS, offsets);
 		}
 		this.kafkaConsumer.commitSync();
 	}

spring-batch-infrastructure/src/test/java/org/springframework/batch/infrastructure/item/kafka/KafkaItemReaderIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2023 the original author or authors.
+ * Copyright 2019-2025 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.
@@ -60,6 +60,7 @@
  * @author Mahmoud Ben Hassine
  * @author François Martin
  * @author Patrick Baumgartner
+ * @author Hyunwoo Jung
  */
 @Testcontainers(disabledWithoutDocker = true)
 @ExtendWith(SpringExtension.class)
@@ -267,8 +268,8 @@ void testReadFromSinglePartitionAfterRestart() throws ExecutionException, Interr
 			future.get();
 		}
 		ExecutionContext executionContext = new ExecutionContext();
-		Map<TopicPartition, Long> offsets = new HashMap<>();
-		offsets.put(new TopicPartition("topic3", 0), 1L);
+		Map<String, Long> offsets = new HashMap<>();
+		offsets.put("0", 1L);
 		executionContext.put("topic.partition.offsets", offsets);
 
 		// topic3-0: val0, val1, val2, val3, val4
@@ -308,9 +309,9 @@ void testReadFromMultiplePartitionsAfterRestart() throws ExecutionException, Int
 		}
 
 		ExecutionContext executionContext = new ExecutionContext();
-		Map<TopicPartition, Long> offsets = new HashMap<>();
-		offsets.put(new TopicPartition("topic4", 0), 1L);
-		offsets.put(new TopicPartition("topic4", 1), 2L);
+		Map<String, Long> offsets = new HashMap<>();
+		offsets.put("0", 1L);
+		offsets.put("1", 2L);
 		executionContext.put("topic.partition.offsets", offsets);
 
 		// topic4-0: val0, val2, val4, val6