diff --git a/core/src/main/scala/kafka/log/streamaspect/ElasticLog.scala b/core/src/main/scala/kafka/log/streamaspect/ElasticLog.scala index b37fa8db29..d25fa2043d 100644 --- a/core/src/main/scala/kafka/log/streamaspect/ElasticLog.scala +++ b/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) diff --git a/core/src/test/scala/kafka/log/streamaspect/ElasticLogTest.scala b/core/src/test/scala/kafka/log/streamaspect/ElasticLogTest.scala index c137301a2e..a1371573ab 100644 --- a/core/src/test/scala/kafka/log/streamaspect/ElasticLogTest.scala +++ b/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,