[fix][broker] Fix NPE in MessageDeduplication.

[Technoboy-]

Motivation

When MessageDeduplication#purgeInactiveProducers:

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

Lines 455 to 477 in d09c6eb

	public synchronized void purgeInactiveProducers() {
	long minimumActiveTimestamp = System.currentTimeMillis() - TimeUnit.MINUTES
	.toMillis(pulsar.getConfiguration().getBrokerDeduplicationProducerInactivityTimeoutMinutes());
	Iterator<java.util.Map.Entry<String, Long>> mapIterator = inactiveProducers.entrySet().iterator();
	boolean hasInactive = false;
	while (mapIterator.hasNext()) {
	java.util.Map.Entry<String, Long> entry = mapIterator.next();
	String producerName = entry.getKey();
	long lastActiveTimestamp = entry.getValue();
	if (lastActiveTimestamp < minimumActiveTimestamp) {
	log.info("[{}] Purging dedup information for producer {}", topic.getName(), producerName);
	mapIterator.remove();
	highestSequencedPushed.remove(producerName);
	highestSequencedPersisted.remove(producerName);
	hasInactive = true;
	}
	}
	if (hasInactive) {
	takeSnapshot(getManagedCursor().getMarkDeletedPosition());
	}
	}

If MessageDeduplication status is not Enabled, the cursor will be null. and cause to be NPE at line 475.

See cursor initializes:

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

Lines 272 to 292 in a439811

	} else if ((status == Status.Disabled || status == Status.Initialized) && shouldBeEnabled) {
	// Enable deduping
	CompletableFuture<Void> future = new CompletableFuture<>();
	managedLedger.asyncOpenCursor(PersistentTopic.DEDUPLICATION_CURSOR_NAME, new OpenCursorCallback() {
	@Override
	public void openCursorComplete(ManagedCursor cursor, Object ctx) {
	// We don't want to retain cache for this cursor
	cursor.setAlwaysInactive();
	managedCursor = cursor;
	recoverSequenceIdsMap().thenRun(() -> {
	status = Status.Enabled;
	future.complete(null);
	log.info("[{}] Enabled deduplication", topic.getName());
	}).exceptionally(ex -> {
	status = Status.Failed;
	log.warn("[{}] Failed to enable deduplication: {}", topic.getName(), ex.getMessage());
	future.completeExceptionally(ex);
	return null;
	});
	}

StackTrace

May 25, 2022 @ 13:39:04.552	2022-05-25T05:39:04,546+0000 [pulsar-inactivity-monitor-22-1] ERROR org.apache.bookkeeper.common.util.SafeRunnable - Unexpected throwable caught 
May 25, 2022 @ 13:39:04.552		at org.apache.pulsar.broker.service.BrokerService.lambda$forEachTopic$78(BrokerService.java:1768) 
May 25, 2022 @ 13:39:04.552		at org.apache.pulsar.broker.service.persistent.MessageDeduplication.purgeInactiveProducers(MessageDeduplication.java:475) 
May 25, 2022 @ 13:39:04.552		at java.util.Optional.ifPresent(Optional.java:183) ~[?:?]
May 25, 2022 @ 13:39:04.552		at org.apache.pulsar.broker.service.BrokerService.checkMessageDeduplicationInfo(BrokerService.java:1735) 
May 25, 2022 @ 13:39:04.552		at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) [?:?]
May 25, 2022 @ 13:39:04.552		at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap.forEach(ConcurrentOpenHashMap.java:272) 
May 25, 2022 @ 13:39:04.552		at org.apache.bookkeeper.common.util.SafeRunnable.run(SafeRunnable.java:36) 
May 25, 2022 @ 13:39:04.552	java.lang.NullPointerException: null
May 25, 2022 @ 13:39:04.552		at org.apache.pulsar.broker.service.BrokerService.forEachTopic(BrokerService.java:1766) 
May 25, 2022 @ 13:39:04.552		at org.apache.bookkeeper.mledger.util.SafeRun$1.safeRun(SafeRun.java:32) 

Documentation

• doc-not-needed
  (Please explain why)

[github-actions]

@Technoboy-:Thanks for your contribution. For this PR, do we need to update docs?
(The PR template contains info about doc, which helps others know more about the changes. Can you provide doc-related info in this and future PR descriptions? Thanks)

[BewareMyPower]

I'm not sure if it could solve the NPE. Is there a race condition like the following code snippet I commented?

        if (hasInactive && isEnabled()) { // 1. isEnabled() returns true, managedCursor is not null
            // during 1 and 2, some other methods changed `managedCursor` to null
            takeSnapshot(getManagedCursor().getMarkDeletedPosition()); // 2. managedCursor might be null

I'd prefer the following way.

        final ManagedCursor cursor = managedCursor;
        if (hasInactive && cursor != null) {
            takeSnapshot(cursor.getMarkDeletedPosition(), cursor);
        }

    private void takeSnapshot(Position position) {
        takeSnapshot(position, getManagedCursor());
    }

    private void takeSnapshot(Position position, ManagedCursor cursor) {
        /* ... */
        cursor.asyncMarkDelete(position, snapshot, new MarkDeleteCallback() {

[Technoboy-]

I'm not sure if it could solve the NPE. Is there a race condition like the following code snippet I commented?

        if (hasInactive && isEnabled()) { // 1. isEnabled() returns true, managedCursor is not null
            // during 1 and 2, some other methods changed `managedCursor` to null
            takeSnapshot(getManagedCursor().getMarkDeletedPosition()); // 2. managedCursor might be null

I'd prefer the following way.

        final ManagedCursor cursor = managedCursor;
        if (hasInactive && cursor != null) {
            takeSnapshot(cursor.getMarkDeletedPosition(), cursor);
        }

    private void takeSnapshot(Position position) {
        takeSnapshot(position, getManagedCursor());
    }

    private void takeSnapshot(Position position, ManagedCursor cursor) {
        /* ... */
        cursor.asyncMarkDelete(position, snapshot, new MarkDeleteCallback() {

Method checkStatus and purgeInactiveProducers is synchonized. Seems no race condition.