You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by sh...@apache.org on 2024/03/06 01:01:04 UTC
(kafka) branch trunk updated: KAFKA-16209 : fetchSnapshot might return null if topic is created before v2.8 (#15444)
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 <54...@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 <sh...@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() {