[Branch-2.8][Cherry-pick] Fix compaction subscription acknowledge Marker msg issue.

[Technoboy-]

Cherry-pick #16917 #16205

Motivation

06:03:58.778 [broker-topic-workers-OrderedScheduler-4-0] ERROR org.apache.bookkeeper.common.util.SafeRunnable - Unexpected throwable caught
java.lang.IllegalArgumentException: null
        at com.google.common.base.Preconditions.checkArgument(Preconditions.java:128)
        at org.apache.pulsar.broker.service.persistent.CompactorSubscription.acknowledgeMessage(CompactorSubscription.java:61) 
        at org.apache.pulsar.broker.service.AbstractBaseDispatcher.filterEntriesForConsumer(AbstractBaseDispatcher.java:154) 
        at org.apache.pulsar.broker.service.AbstractBaseDispatcher.filterEntriesForConsumer(AbstractBaseDispatcher.java:103)
        at org.apache.pulsar.broker.service.persistent.PersistentDispatcherSingleActiveConsumer.internalReadEntriesComplete(PersistentDispatcherSingleActiveConsumer.java:203) 
        at org.apache.pulsar.broker.service.persistent.PersistentDispatcherSingleActiveConsumer.lambda$readEntriesComplete$1(PersistentDispatcherSingleActiveConsumer.java:146) 
        at org.apache.bookkeeper.mledger.util.SafeRun$1.safeRun(SafeRun.java:32) 
        at org.apache.bookkeeper.common.util.SafeRunnable.run(SafeRunnable.java:36) 
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) [?:?]
        at java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?]
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304) [?:?]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?]
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) 
        at java.lang.Thread.run(Thread.java:829) [?:?]

If the topic enabled transaction or replicated cluster, the original topic will write some Marker msg. Then if enabled compaction, it will trigger CompactionSubscription to acknowledge failure due to inconsistent AckType:

line-178 will ack as Individual:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java

Lines 168 to 181 in c871d24

	} else if (msgMetadata == null || Markers.isServerOnlyMarker(msgMetadata)) {
	PositionImpl pos = (PositionImpl) entry.getPosition();
	// Message metadata was corrupted or the messages was a server-only marker
	if (Markers.isReplicatedSubscriptionSnapshotMarker(msgMetadata)) {
	processReplicatedSubscriptionSnapshot(pos, metadataAndPayload);
	}
	entries.set(i, null);
	entry.release();
	subscription.acknowledgeMessage(Collections.singletonList(pos), AckType.Individual,
	Collections.emptyMap());
	continue;
	} else if (msgMetadata.hasDeliverAtTime()

But CompactorSubscription only supports Cumulative :

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/CompactorSubscription.java

Lines 60 to 64 in c871d24

	public void acknowledgeMessage(List<Position> positions, AckType ackType, Map<String, Long> properties) {
	checkArgument(ackType == AckType.Cumulative);
	checkArgument(positions.size() == 1);
	checkArgument(properties.containsKey(Compactor.COMPACTED_TOPIC_LEDGER_PROPERTY));
	long compactedLedgerId = properties.get(Compactor.COMPACTED_TOPIC_LEDGER_PROPERTY);

Above is the root cause.

Then if occurs this error, it may cause the broker OOM, because some entries are not released.

Modification

• Check subscription type when acknowledge.

Verifying this change

• Add a new test to cover this change.

Documentation

• doc-not-needed
  (Please explain why)