[BUG] Ack wrong position

Describe the bug

Use the perf tool to work with the AoP, I found error logs below.

bin/runjava com.rabbitmq.perf.PerfTest -x 1 -y 2 -u "throughput-test-12" --id "test-11" \
  -f persistent -q 500 --rate 20000

2022-06-28T11:44:55,816+0800 [pulsar-ph-amqp-46-2] WARN  io.streamnative.pulsar.handlers.amqp.impl.PersistentExchange - Mark delete success for position: 8:3940 with error:
org.apache.bookkeeper.mledger.ManagedLedgerException: java.lang.IllegalArgumentException: Mark deleting an already mark-deleted position. Current mark-delete: 8:3960 -- attempted mark delete: 8:3940
Caused by: java.lang.IllegalArgumentException: Mark deleting an already mark-deleted position. Current mark-delete: 8:3960 -- attempted mark delete: 8:3940
	at org.apache.bookkeeper.mledger.impl.ManagedCursorImpl.setAcknowledgedPosition(ManagedCursorImpl.java:1693) ~[org.apache.pulsar-managed-ledger-2.11.0-SNAPSHOT.jar:2.11.0-SNAPSHOT]
	at org.apache.bookkeeper.mledger.impl.ManagedCursorImpl.asyncMarkDelete(ManagedCursorImpl.java:1833) ~[org.apache.pulsar-managed-ledger-2.11.0-SNAPSHOT.jar:2.11.0-SNAPSHOT]
	at org.apache.bookkeeper.mledger.impl.ManagedCursorImpl.asyncMarkDelete(ManagedCursorImpl.java:1760) ~[org.apache.pulsar-managed-ledger-2.11.0-SNAPSHOT.jar:2.11.0-SNAPSHOT]
	at io.streamnative.pulsar.handlers.amqp.impl.PersistentExchange.markDeleteAsync(PersistentExchange.java:164) ~[?:?]
	at io.streamnative.pulsar.handlers.amqp.impl.PersistentExchange.markDeleteAsync(PersistentExchange.java:149) ~[?:?]
	at io.streamnative.pulsar.handlers.amqp.impl.PersistentQueue.acknowledgeAsync(PersistentQueue.java:94) ~[?:?]
	at io.streamnative.pulsar.handlers.amqp.AmqpConsumer.lambda$messagesAck$5(AmqpConsumer.java:209) ~[?:?]
	at java.util.concurrent.ConcurrentHashMap.forEach(ConcurrentHashMap.java:1603) ~[?:?]
	at io.streamnative.pulsar.handlers.amqp.AmqpConsumer.lambda$messagesAck$6(AmqpConsumer.java:205) ~[?:?]
	at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:863) ~[?:?]
	at java.util.concurrent.CompletableFuture.uniWhenCompleteStage(CompletableFuture.java:887) ~[?:?]
	at java.util.concurrent.CompletableFuture.whenComplete(CompletableFuture.java:2325) ~[?:?]
	at io.streamnative.pulsar.handlers.amqp.AmqpConsumer.messagesAck(AmqpConsumer.java:199) ~[?:?]
	at io.streamnative.pulsar.handlers.amqp.AmqpConsumer.messagesAck(AmqpConsumer.java:220) ~[?:?]
	at io.streamnative.pulsar.handlers.amqp.AmqpChannel.lambda$messageAck$13(AmqpChannel.java:643) ~[?:?]
	at java.util.Collections$2.tryAdvance(Collections.java:4853) ~[?:?]
	at java.util.Collections$2.forEachRemaining(Collections.java:4861) ~[?:?]
	at java.util.stream.ReferencePipeline$Head.forEach(ReferencePipeline.java:762) ~[?:?]
	at io.streamnative.pulsar.handlers.amqp.AmqpChannel.messageAck(AmqpChannel.java:642) ~[?:?]
	at io.streamnative.pulsar.handlers.amqp.AmqpChannel.receiveBasicAck(AmqpChannel.java:635) ~[?:?]
	at org.apache.qpid.server.protocol.v0_8.transport.BasicAckBody.process(BasicAckBody.java:119) ~[?:?]
	at org.apache.qpid.server.protocol.v0_8.ServerDecoder.processMethod(ServerDecoder.java:194) ~[?:?]
	at org.apache.qpid.server.protocol.v0_8.AMQDecoder.processFrame(AMQDecoder.java:203) ~[?:?]
	at org.apache.qpid.server.protocol.v0_8.AMQDecoder.processInput(AMQDecoder.java:185) ~[?:?]
	at org.apache.qpid.server.protocol.v0_8.AMQDecoder.processAMQPFrames(AMQDecoder.java:138) ~[?:?]
	at org.apache.qpid.server.protocol.v0_8.AMQDecoder.decode(AMQDecoder.java:118) ~[?:?]
	at org.apache.qpid.server.protocol.v0_8.ServerDecoder.decodeBuffer(ServerDecoder.java:44) ~[?:?]
	at io.streamnative.pulsar.handlers.amqp.AmqpBrokerDecoder.decodeBuffer(AmqpBrokerDecoder.java:62) ~[?:?]
	at io.streamnative.pulsar.handlers.amqp.AmqpConnection.channelRead(AmqpConnection.java:151) ~[?:?]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379) ~[io.netty-netty-transport-4.1.77.Final.jar:4.1.77.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365) ~[io.netty-netty-transport-4.1.77.Final.jar:4.1.77.Final]
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357) ~[io.netty-netty-transport-4.1.77.Final.jar:4.1.77.Final]
	at io.netty.handler.timeout.IdleStateHandler.channelRead(IdleStateHandler.java:286) ~[io.netty-netty-handler-4.1.77.Final.jar:4.1.77.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379) ~[io.netty-netty-transport-4.1.77.Final.jar:4.1.77.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365) ~[io.netty-netty-transport-4.1.77.Final.jar:4.1.77.Final]
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357) ~[io.netty-netty-transport-4.1.77.Final.jar:4.1.77.Final]
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410) ~[io.netty-netty-transport-4.1.77.Final.jar:4.1.77.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379) ~[io.netty-netty-transport-4.1.77.Final.jar:4.1.77.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365) ~[io.netty-netty-transport-4.1.77.Final.jar:4.1.77.Final]
	at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919) ~[io.netty-netty-transport-4.1.77.Final.jar:4.1.77.Final]
	at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166) ~[io.netty-netty-transport-4.1.77.Final.jar:4.1.77.Final]
	at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:722) ~[io.netty-netty-transport-4.1.77.Final.jar:4.1.77.Final]
	at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:658) ~[io.netty-netty-transport-4.1.77.Final.jar:4.1.77.Final]
	at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:584) ~[io.netty-netty-transport-4.1.77.Final.jar:4.1.77.Final]
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:496) ~[io.netty-netty-transport-4.1.77.Final.jar:4.1.77.Final]
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:995) ~[io.netty-netty-common-4.1.77.Final.jar:4.1.77.Final]
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) ~[io.netty-netty-common-4.1.77.Final.jar:4.1.77.Final]
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[io.netty-netty-common-4.1.77.Final.jar:4.1.77.Final]
	at java.lang.Thread.run(Thread.java:833) [?:?]