This is an automated email from the ASF dual-hosted git repository. showuon pushed a commit to branch trunk in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push: new 554fa57af85 KAFKA-16209 : fetchSnapshot might return null if topic is created before v2.8 (#15444) 554fa57af85 is described below commit 554fa57af85ec337a556f35cbc6d2970ff252dc4 Author: John Yu <54207775+chia...@users.noreply.github.com> AuthorDate: Wed Mar 6 09:00:58 2024 +0800 KAFKA-16209 : fetchSnapshot might return null if topic is created before v2.8 (#15444) Change the function with a better way to deal with the NULL pointer exception. Reviewers: Luke Chen <show...@gmail.com> --- core/src/test/scala/unit/kafka/log/ProducerStateManagerTest.scala | 6 ++++++ .../apache/kafka/storage/internals/log/ProducerStateManager.java | 2 +- 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/core/src/test/scala/unit/kafka/log/ProducerStateManagerTest.scala b/core/src/test/scala/unit/kafka/log/ProducerStateManagerTest.scala index 03aa847ded7..810e0b1e4ee 100644 --- a/core/src/test/scala/unit/kafka/log/ProducerStateManagerTest.scala +++ b/core/src/test/scala/unit/kafka/log/ProducerStateManagerTest.scala @@ -579,6 +579,12 @@ class ProducerStateManagerTest { assertTrue(logDir.list().head.nonEmpty, "Snapshot file is empty") } + @Test + def testFetchSnapshotEmptySnapShot(): Unit = { + val offset = 1 + assertEquals(Optional.empty(), stateManager.fetchSnapshot(offset)) + } + @Test def testRecoverFromSnapshotUnfinishedTransaction(): Unit = { val epoch = 0.toShort diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/ProducerStateManager.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/ProducerStateManager.java index 270aa0a42f9..da56ddaccc8 100644 --- a/storage/src/main/java/org/apache/kafka/storage/internals/log/ProducerStateManager.java +++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/ProducerStateManager.java @@ -605,7 +605,7 @@ public class ProducerStateManager { } public Optional<File> fetchSnapshot(long offset) { - return Optional.of(snapshots.get(offset)).map(x -> x.file()); + return Optional.ofNullable(snapshots.get(offset)).map(x -> x.file()); } private Optional<SnapshotFile> oldestSnapshotFile() {