[Bug] SnapshotSegmentAbortedTxnProcessorImpl contains thread safety issues

[lhotari]

Search before asking

• I searched in the issues and found nothing similar.

Version

master

Minimal reproduce step

There are several thread safety issues in SnapshotSegmentAbortedTxnProcessorImpl class.
examples:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java

Lines 374 to 376 in 529e1ab

	if (indexes.size() != 0) {
	persistentWorker.sequenceID.set(indexes.get(indexes.lastKey()).sequenceID + 1);
	}

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java

Line 383 in 529e1ab

unsealedTxnIds.forEach(txnID -> aborts.put(txnID, txnID));

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java

Line 744 in 529e1ab

indexes.put(abortedMarkerPersistentPosition, index);

What did you expect to see?

Multi-threading code should be thread safe.

What did you see instead?

Non thread-safe code.

Anything else?

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!

[lhotari]

related to #21304 and #22119

[nikam14]

@lhotari can you review above mentioned PR.