[fix][broker] Fix compaction subscription acknowledge Marker msg issue.

[Technoboy-]

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)

[mattisonchao]

I think it's a great catch!

Left some questions would you like to answer?

[HQebupt]

Nice catch! 👍

[BewareMyPower]

@Technoboy- Could you help cherry-pick this PR to branch-2.8? It's weird that the createNonPartitionedTopic call failed in my local env.

12:37:49.693 [AsyncHttpClient-43-1] WARN  org.apache.pulsar.client.admin.internal.BaseResource - [http://localhost:53870/admin/persistent/my-property/use/my-ns/testWriteMarker-caf36fc4-1e8c-45f8-a49f-674b1d145a19] Failed to perform http put request: javax.ws.rs.NotAllowedException: HTTP 405 Method Not Allowed

org.apache.pulsar.client.admin.PulsarAdminException$NotAllowedException: HTTP 405 Method Not Allowed

	at org.apache.pulsar.client.admin.internal.BaseResource.getApiException(BaseResource.java:232)
	at org.apache.pulsar.client.admin.internal.BaseResource$1.failed(BaseResource.java:130)

[BewareMyPower]

Move the release/2.8.4 label to #16918