fix: avoid snapshot-read close metadata NPE

core/src/main/scala/kafka/log/streamaspect/ElasticLog.scala

@@ -125,13 +125,16 @@ class ElasticLog(val metaStream: MetaStream,
     private val readAsyncThread = READ_ASYNC_EXECUTOR(math.abs(logIdent.hashCode % READ_ASYNC_EXECUTOR.length))
     var logStartOffset = _initStartOffset
 
-    // persist log meta when lazy stream real create
-    streamManager.setListener((_, event) => {
-        if (event == ElasticStreamMetaEvent.STREAM_DO_CREATE) {
-            logSegmentManager.asyncPersistLogMeta()
-            logSegmentManager.notifySegmentUpdate();
-        }
-    })
+    // Persist log metadata when a normal log lazily creates a stream. Snapshot-read logs
+    // intentionally do not own a MetaStream or LogSegmentManager.
+    if (!snapshotRead) {
+        streamManager.setListener((_, event) => {
+            if (event == ElasticStreamMetaEvent.STREAM_DO_CREATE) {
+                logSegmentManager.asyncPersistLogMeta()
+                logSegmentManager.notifySegmentUpdate();
+            }
+        })
+    }
 
     private def maybeHandleIOExceptionAsync[T](msg: => String)(fun: => CompletableFuture[T]): CompletableFuture[T] = {
         ElasticLog.maybeHandleIOExceptionAsync(logDirFailureChannel, _dir.getPath, msg) {
@@ -523,6 +526,23 @@ class ElasticLog(val metaStream: MetaStream,
      * ref. LocalLog#close
      */
     override private[log] def close(): Unit = {
+        if (snapshotRead) {
+            maybeHandleIOException(s"Error while closing snapshot-read $topicPartition in dir ${dir.getParent}") {
+                checkIfMemoryMappedBufferClosed()
+                try {
+                    segments.close()
+                } finally {
+                    closeStreams(false).handle[Void]((_, exception) => {
+                        if (exception != null) {
+                            warn(s"${logIdent}failed to close snapshot-read streams", FutureUtil.cause(exception))
+                        }
+                        null
+                    }).join()
+                }
+                info("snapshot-read log closed")
+            }
+            return
+        }
         metaStream.beforeClose()
         markFlushed(logEndOffset)
         partitionMeta.setCleanedShutdown(true)

core/src/test/scala/kafka/log/streamaspect/ElasticLogTest.scala

@@ -22,6 +22,7 @@ package kafka.log.streamaspect
 import com.automq.stream.DefaultRecordBatch
 import com.automq.stream.api.{AppendResult, CreateStreamOptions, FetchResult, OpenStreamOptions, RecordBatch, Stream, StreamClient}
 import com.automq.stream.s3.context.{AppendContext, FetchContext}
+import kafka.cluster.PartitionSnapshot
 import kafka.log._
 import kafka.log.streamaspect.reassignment._
 import kafka.server.KafkaConfig
@@ -1350,6 +1351,41 @@ class ElasticLogTest {
         assertTrue(fetchDataInfo.abortedTransactions.get.isEmpty)
     }
 
+    /** A snapshot-read log has no MetaStream or LogSegmentManager, but still closes its data streams. */
+    @Test
+    def testSnapshotReadCloseSkipsMetadataLifecycle(): Unit = {
+        val snapshotDir = Files.createTempDirectory("elastic-snapshot-read-").toFile
+        val snapshotPartition = new TopicPartition("snapshot-close", 0)
+        val snapshotBuilder = PartitionSnapshot.builder().leaderEpoch(1)
+        log.snapshot(snapshotBuilder)
+        val client = new CloseTrackingClient()
+        val snapshotLog = ElasticLog.apply(
+            client,
+            "",
+            snapshotDir,
+            LogTestUtils.createLogConfig(),
+            mockTime.scheduler,
+            time = mockTime,
+            topicPartition = snapshotPartition,
+            logDirFailureChannel = logDirFailureChannel,
+            maxTransactionTimeoutMs = 5 * 60 * 1000,
+            producerStateManagerConfig = producerStateManagerConfig,
+            topicId = Some(Uuid.randomUuid()),
+            leaderEpoch = 1,
+            openStreamChecker = OpenStreamChecker.NOOP,
+            snapshotRead = true)
+
+        try {
+            snapshotLog.snapshot(snapshotBuilder.build())
+            assertFalse(snapshotLog.segments.isEmpty)
+            snapshotLog.close()
+            assertFalse(client.createdStreams.isEmpty)
+            assertTrue(client.createdStreams.asScala.forall(_.closeCount.get() == 1))
+        } finally {
+            Utils.delete(snapshotDir)
+        }
+    }
+
     private def createElasticLogWithActiveSegment(dir: File = logDir,
         config: LogConfig,
         scheduler: Scheduler = mockTime.scheduler,