[fix][broker] Fix possible mark delete NPE when batch index ack is enabled

[BewareMyPower]

Motivation

MLPendingAckStoreTest#testMainProcess will fail by NPE if batch index ACK is enabled. The NPE happens when the position argument does not has a non-null ackSet.

Modifications

• Add an ackBatchPosition method to handle Optional objects correctly with elegant and clear code.
• Make BitSetRecycable#recycle() call safe to be called multiple times so that it can be called in ConcurrentSkipListMap's callback, which is not guaranteed to be called only once.
• Mark batchDeletedIndexes as @Nullable and check if it's not null instead of getConfig().isDeletionAtBatchIndexLevelEnabled() to eliminate all NPE warnings.

Verifying this change

Enable batch index ACK in MLPendingAckStoreTest and guarantee this test passes.

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository:

[dao-jun]

If we set the recycleHandler = null here, looks like it will downgrade to shortlive objects

[BewareMyPower]

Yes. What's the issue here?

[dao-jun]

I am just wandering that since it can be downgrade to shortlive objects, why not just new BitSetRecyclable() here directly, so we don't need to call recycle and don't need to set recycleHandle=null

[dao-jun]

By this way, we can keep the semantic align, WDYT?

[BewareMyPower]

I agree. I will use BitSet instead. Using a recyclable object is very error-prone in non-trivial cases. In this specific case, it's used as the value of a map and we must guarantee recycle() is called when the value is removed.

The benefit of recycler is a pre-mature optimization IMO.

[dao-jun]

+1, use bitset here directly will not lead to performance regression, and makes it more clearly. The key point is BitSet#words field, since we just wrap the given ackSet by BitSet, it is OK to use BitSet instead.