fix(zerozone2): guard snapshot by lock

core/src/main/scala/kafka/cluster/Partition.scala

@@ -2361,17 +2361,23 @@ class Partition(val topicPartition: TopicPartition,
 
   def snapshot(snapshot: PartitionSnapshot): Unit = {
     inWriteLock(leaderIsrUpdateLock) {
-      if (enableTraceLog) {
-        trace(s"apply snapshot partition $topic-$partitionId $snapshot")
+      try {
+        if (enableTraceLog) {
+          trace(s"apply snapshot partition $topic-$partitionId $snapshot")
+        }
+        leaderEpoch = snapshot.leaderEpoch
+        val log = this.log.get.asInstanceOf[ElasticUnifiedLog]
+        log.snapshot(snapshot)
+        log.getLocalLog().appendAckThread.submit(() => {
+          // async it to avoid deadlock
+          tryCompleteDelayedRequests()
+          null
+        })
+      } catch {
+        case e: Throwable =>
+          error(s"apply snapshot fail, snapshot=$snapshot", e)
+          throw e
       }
-      leaderEpoch = snapshot.leaderEpoch
-      val log = this.log.get.asInstanceOf[ElasticUnifiedLog]
-      log.snapshot(snapshot)
-      log.getLocalLog().appendAckThread.submit(() => {
-        // async it to avoid deadlock
-        tryCompleteDelayedRequests()
-        null
-      });
     }
   }
   // AutoMQ injection end

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

@@ -277,29 +277,33 @@ class ElasticUnifiedLog(_logStartOffset: Long,
     }
 
     def snapshot(snapshot: PartitionSnapshot.Builder): Unit = {
-        snapshot.firstUnstableOffset(firstUnstableOffsetMetadata.orNull)
-        val localLog = getLocalLog()
-        localLog.snapshot(snapshot)
+        lock synchronized {
+            snapshot.firstUnstableOffset(firstUnstableOffsetMetadata.orNull)
+            val localLog = getLocalLog()
+            localLog.snapshot(snapshot)
+        }
     }
 
     def snapshot(snapshot: PartitionSnapshot): Unit = {
-        val localLog = getLocalLog()
-        localLog.snapshot(snapshot)
-        if (snapshot.firstUnstableOffset() == null) {
-            firstUnstableOffsetMetadata = None
-        } else {
-            var offset = snapshot.firstUnstableOffset()
-            val segmentBaseOffset = localLog.segments.floorSegment(offset.messageOffset).get().baseOffset()
-            offset = new LogOffsetMetadata(offset.messageOffset, segmentBaseOffset, offset.relativePositionInSegment)
-            firstUnstableOffsetMetadata = Some(offset)
-        }
-        if (snapshot.logMeta() != null) {
-            val opt = localLog.segments.firstSegmentBaseOffset()
-            opt.ifPresent(baseOffset => {
-                updateLogStartOffset(baseOffset)
-            })
+        lock synchronized {
+            val localLog = getLocalLog()
+            localLog.snapshot(snapshot)
+            if (snapshot.firstUnstableOffset() == null) {
+                firstUnstableOffsetMetadata = None
+            } else {
+                var offset = snapshot.firstUnstableOffset()
+                val segmentBaseOffset = localLog.segments.floorSegment(offset.messageOffset).get().baseOffset()
+                offset = new LogOffsetMetadata(offset.messageOffset, segmentBaseOffset, offset.relativePositionInSegment)
+                firstUnstableOffsetMetadata = Some(offset)
+            }
+            if (snapshot.logMeta() != null) {
+                val opt = localLog.segments.firstSegmentBaseOffset()
+                opt.ifPresent(baseOffset => {
+                    updateLogStartOffset(baseOffset)
+                })
+            }
+            highWatermarkMetadata = localLog.logEndOffsetMetadata
         }
-        highWatermarkMetadata = localLog.logEndOffsetMetadata
     }
 
 }