Bugs in snapsot and recover logic

[ShadowySpirits]

1. The failure to take snapshots resulted in fetching too many operations during recovery, leading to the OOM error.

2023-11-06 12:20:33 WARN  Failed to take snapshot
java.lang.IndexOutOfBoundsException: Range [0, 0 + 487) out of bounds for length 486
	at jdk.internal.util.Preconditions.outOfBounds(Preconditions.java:64) ~[?:?]
	at jdk.internal.util.Preconditions.outOfBoundsCheckFromIndexSize(Preconditions.java:82) ~[?:?]
	at jdk.internal.util.Preconditions.checkFromIndexSize(Preconditions.java:361) ~[?:?]
	at java.util.Objects.checkFromIndexSize(Objects.java:411) ~[?:?]
	at java.nio.CharBuffer.put(CharBuffer.java:1176) ~[?:?]
	at org.roaringbitmap.ArrayContainer.writeArray(ArrayContainer.java:1313) ~[RoaringBitmap-1.0.0.jar:?]
	at org.roaringbitmap.RoaringArray.serialize(RoaringArray.java:944) ~[RoaringBitmap-1.0.0.jar:?]
	at org.roaringbitmap.RoaringBitmap.serialize(RoaringBitmap.java:2930) ~[RoaringBitmap-1.0.0.jar:?]
	at com.automq.rocketmq.store.queue.DefaultLogicQueueStateMachine$AckCommitter.getAckBitmapBuffer(DefaultLogicQueueStateMachine.java:675) ~[rocketmq-store-5.1.3-automq-0-SNAPSHOT.jar:?]
	at com.automq.rocketmq.store.queue.DefaultLogicQueueStateMachine.lambda$takeSnapshot$14(DefaultLogicQueueStateMachine.java:533) ~[rocketmq-store-5.1.3-automq-0-SNAPSHOT.jar:?]
	at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197) ~[?:?]
	at java.util.concurrent.ConcurrentHashMap$ValueSpliterator.forEachRemaining(ConcurrentHashMap.java:3612) ~[?:?]
	at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509) ~[?:?]
	at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499) ~[?:?]
	at java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:921) ~[?:?]
	at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) ~[?:?]
	at java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:682) ~[?:?]
	at com.automq.rocketmq.store.queue.DefaultLogicQueueStateMachine.takeSnapshot(DefaultLogicQueueStateMachine.java:539) ~[rocketmq-store-5.1.3-automq-0-SNAPSHOT.jar:?]
	at com.automq.rocketmq.store.service.SnapshotService.takeSnapshot(SnapshotService.java:155) ~[rocketmq-store-5.1.3-automq-0-SNAPSHOT.jar:?]
	at com.automq.rocketmq.store.service.SnapshotService.run(SnapshotService.java:132) [rocketmq-store-5.1.3-automq-0-SNAPSHOT.jar:?]
	at java.lang.Thread.run(Thread.java:833) [?:?]
2023-11-06 12:27:07 WARN  Failed to take snapshot
java.lang.IndexOutOfBoundsException: Range [0, 0 + 3333) out of bounds for length 3332
	at jdk.internal.util.Preconditions.outOfBounds(Preconditions.java:64) ~[?:?]
	at jdk.internal.util.Preconditions.outOfBoundsCheckFromIndexSize(Preconditions.java:82) ~[?:?]
	at jdk.internal.util.Preconditions.checkFromIndexSize(Preconditions.java:361) ~[?:?]
	at java.util.Objects.checkFromIndexSize(Objects.java:411) ~[?:?]
	at java.nio.CharBuffer.put(CharBuffer.java:1176) ~[?:?]
	at org.roaringbitmap.ArrayContainer.writeArray(ArrayContainer.java:1313) ~[RoaringBitmap-1.0.0.jar:?]
	at org.roaringbitmap.RoaringArray.serialize(RoaringArray.java:944) ~[RoaringBitmap-1.0.0.jar:?]
	at org.roaringbitmap.RoaringBitmap.serialize(RoaringBitmap.java:2930) ~[RoaringBitmap-1.0.0.jar:?]
	at com.automq.rocketmq.store.queue.DefaultLogicQueueStateMachine$AckCommitter.getAckBitmapBuffer(DefaultLogicQueueStateMachine.java:675) ~[rocketmq-store-5.1.3-automq-0-SNAPSHOT.jar:?]
	at com.automq.rocketmq.store.queue.DefaultLogicQueueStateMachine.lambda$takeSnapshot$14(DefaultLogicQueueStateMachine.java:533) ~[rocketmq-store-5.1.3-automq-0-SNAPSHOT.jar:?]
	at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197) ~[?:?]
	at java.util.concurrent.ConcurrentHashMap$ValueSpliterator.forEachRemaining(ConcurrentHashMap.java:3612) ~[?:?]
	at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509) ~[?:?]
	at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499) ~[?:?]
	at java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:921) ~[?:?]
	at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) ~[?:?]
	at java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:682) ~[?:?]
	at com.automq.rocketmq.store.queue.DefaultLogicQueueStateMachine.takeSnapshot(DefaultLogicQueueStateMachine.java:539) ~[rocketmq-store-5.1.3-automq-0-SNAPSHOT.jar:?]
	at com.automq.rocketmq.store.service.SnapshotService.takeSnapshot(SnapshotService.java:155) ~[rocketmq-store-5.1.3-automq-0-SNAPSHOT.jar:?]
	at com.automq.rocketmq.store.service.SnapshotService.run(SnapshotService.java:132) [rocketmq-store-5.1.3-automq-0-SNAPSHOT.jar:?]
	at java.lang.Thread.run(Thread.java:833) [?:?]

2. Failed to replay operation

2023-11-03 12:20:12 ERROR [DefaultLogicQueueManaager]: Open logic queue failed: topic: 60 queue:7
java.lang.IllegalStateException: Unexpected value:36
at com.automq.rocketmq.store.util.SerializeUtil.decode0peration(SerializeUtil.java:170) ~[rocketmq-store-5.1.3-automq-0-SNAPSHOT.jar:?]
at com.automq.rocketmq.store.service.StreamOperationLogService.lambda$recover$2(StreamOperationLogService.java:85) ~[rocketmq-store-5.1.3-automq-0-SNAPSHOT.jar:?
at java.util.concurrent.CompletableFuture$UniAccept.tryFire(CompletableFuture.java:718) ~[?:?]
at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510) [?:?]
at java.util.concurrent.CompletableFuture.postFire(CompLetableFuture.java:614) [?:?]
at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:653) [?:?]
at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:482) [?:?]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1135) [?:?]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) [?:?]
at java.lang.Thread.run(Thread.java:831) [?:?]
2023-11-03 12:20:12 ERROR [DefaultLogicQueueManager)|: Create logic queue failed: topic: 60 queue: 7
com.automq.rocketmq.store.exception.StoreException:Unexpected value: 36
at com.automq.rocketmq.store.queue.DefaultLogicQueueManager.lambda$createAndOpen$2(DefaultLogicQueueManager.java:180) ~[rocketmq-store-5.1.3-automq-0-SNAISHOT.jar:?
at java.util.concurrent.CompletableFuture.uniExceptionally(CompletableFuture.java:990) [?:?]
at java.util.concurrent.CompletableFuture$UniExceptionally.tryFire(CompletableFuture.java:974) [?:?]
at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510) [?:?]
at java.util.concurrent.CompletableFuture.postFire(CompLetableFuture.java:614) [?:?]
at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:653) [?:?]
at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:482) [?:?]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1135) [?:?]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) [?:?]
at java.lang.Thread.run(Thread.java:831) [?:?]