[Bug] Topic cannot be accessed for "Failed to load topic within timeout" unless unloading the namespace

[BewareMyPower]

Search before reporting

• I searched in the issues and found nothing similar.

Read release policy

• I understand that unsupported versions don't get bug fixes. I will attempt to reproduce the issue on a supported version of Pulsar client and Pulsar broker.

User environment

StreamNative Pulsar 4.0.5.2, which is based on 6b5fdbf

Issue Description

This issue happens when we restarted brokers for an upgrade. There were two partitions (10 and 23 of a partitioned topic) owned by a broker stuck at initializing the PersistentTopic. Here are the info from the heap dump:

topic:
  - name: "<topic>-partition-23"
    messageDeduplication:
      snapshotCounter: 21
      managedCursor:
        entriesReadCount: 42
        lastMarkDeleteEntry:
          newPosition: "860608:298"
        pendingReadOps: 0
        readPosition: "861217:3"
      managedLedger:
        config:
          readEntryTimeoutSeconds: 0
        executor:
          runner: "BookKeeperClientWorker-OrderedExecutor-0-0"
          tasksFailed: 1
        lastConfirmedEntry: "861217:2"
        ledgers:
          - key: 860608
            value: {"entries": 338}
          - key: 861217
            value: {"entries": 3}
          - key: 861315
            value: {"entries": 0}
  - name: "<topic>-partition-10"
    messageDeduplication:
      snapshotCounter: 0
      managedCursor:
        entriesReadCount: 84
        lastMarkDeleteEntry:
          newPosition: "860650:231"
        pendingReadOps: 1
        readPosition: "861214:0"
      managedLedger:
        currentLedger:
          ledgerId: 861295
        lastConfirmedEntry: "861214:1"
        ledgers:
          - key: 860650
            value: {"entries": 314}
          - key: 861214
            value: {"entries": 2}
          - key: 861295
            value: {"entries": 0}
        state: LedgerOpened

The key is the MessageDeduplication field:

• status is Initialized
• managedCursor is not null
• snapshotCounter is non-zero

From

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

Lines 292 to 298 in ce102da

	managedCursor = cursor;
	recoverSequenceIdsMap().thenRun(() -> {
	status = Status.Enabled;
	future.complete(null);
	log.info("[{}] Enabled deduplication", topic.getName());
	}).exceptionally(ex -> {
	status = Status.Failed;

we can get a conclusion that the future of recoverSequenceIdsMap never completed.

The difference between partition 10 and 23 is:

• Partition 10's cursor read position didn't reach the last confirmed entry, and there was a pending read operation. It seems that a read operation was somehow stuck forever.
• Partition 23's cursor read position exceeds the last confirmed entry, the future should have been completed. There is another suspicious point that snapshotCounter is 21 while entriesReadCount is 42.

The most possible reason is that exceptions were thrown in a callback like

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

Line 178 in ce102da

public void readEntriesComplete(List<Entry> entries, Object ctx) {

There was something wrong with our log collection, the key logs of SingleThreadExecutor#run have been lost. But we can check the taskFailed field is 1 from the heap dump.

Error messages

Reproducing the issue

Currently I cannot reproduce it.

Additional information

Currently Pulsar's topic caching mechanism is bad. There is a timeout configured by topicLoadTimeoutSeconds (default: 60). 1 minute is usually long enough in most cases. However, when the timeout happens, the topic cache won't be removed from BrokerService#topics until the pending PersistentTopic initialization completes. See

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

Lines 1849 to 1855 in ce102da

	persistentTopic
	.initialize()
	.thenCompose(__ -> persistentTopic.preCreateSubscriptionForCompactionIfNeeded())
	.thenCompose(__ -> persistentTopic.checkReplication())
	.thenCompose(v -> {
	// Also check dedup status
	return persistentTopic.checkDeduplicationStatus();

initialize():

1. Create a compaction service and set topicCompactionService with it.
2. Create subscriptions for all cursors in the managed ledger.
3. Get namespace policies for this topic from the metadata store.
4. Initialize some fields according to the namespace policies.
5. Register itself to the topic policies service.
6. Get topic policies and update some fields.
7. Remove orphan replication cursors.

preCreateSubscriptionForCompactionIfNeeded(): Create __compaction subscription, which might open the durable cursor.

checkReplication():

1. Get allowed clusters from the metadata store.
2. Start replicators if necessary.

checkDeduplicationStatus(): Perform the topic replay.

If there is a bug that makes PersistentTopic stuck at initialization, e.g. checkDeduplicationStatus(), the topic will be unavailable until manual intervention.

Are you willing to submit a PR?

• I'm willing to submit a PR!

[BewareMyPower]

The issue of partition 23 can be simulated by the following test:

The following test can simulate the case of partition 23 that readEntriesComplete threw an exception.

    @Test
    public void test() throws Exception {
        ManagedLedgerConfig config = new ManagedLedgerConfig();
        config.setMaximumRolloverTime(1, TimeUnit.SECONDS);
        final var ml = factory.open("test", config);
        final var cursor = ml.openCursor("dedup", InitialPosition.Earliest);
        for (int i = 0; i < 338; i++) {
            ml.addEntry(("msg-0-" + i).getBytes());
        }
        ml.rollCurrentLedgerIfFull();
        Thread.sleep(1000);
        for (int i = 0; i < 3; i++) {
            ml.addEntry(("msg-1-" + i).getBytes());
        }

        final var firstEntries = cursor.readEntries(299);
        cursor.markDelete(firstEntries.get(firstEntries.size() - 1).getPosition());
        firstEntries.forEach(Entry::release);

        ml.close();

        final var ml2 = factory.open("test", config);
        final var cursor2 = ml2.openCursor("dedup", InitialPosition.Earliest);
        cursor2.setAlwaysInactive();

        final var future = new CompletableFuture<List<Entry>>();
        cursor2.asyncReadEntries(100, new ReadEntriesCallback() {
            @Override
            public void readEntriesComplete(List<Entry> entries, Object ctx) {
                if (entries.size() > 1) {
                    throw new RuntimeException("fail");
                }
                future.complete(entries);
            }

            @Override
            public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
                future.completeExceptionally(exception);
            }
        }, null, PositionFactory.LATEST);
        try {
            future.get(1, TimeUnit.SECONDS);
            fail();
        } catch (TimeoutException e) {
            System.out.println("Expected timeout");
        }
    }

pulsar/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpReadEntry.java

Lines 179 to 184 in a40ac3c

	cursor.readOperationCompleted();
	} finally {
	cursor.ledger.getExecutor().execute(() -> {
	callback.readEntriesComplete(entries, ctx);
	recycle();

From the heap dump, line 179 must have been executed because pendingReadOps is 0. There could be a case that line 183 threw an exception.

[BewareMyPower]

The case of partition 10 is a bit different from partition 23.

  - name: "<topic>-partition-10"
    messageDeduplication:
      snapshotCounter: 0
      managedCursor:
        entriesReadCount: 84
        lastMarkDeleteEntry:
          newPosition: "860650:231"
        messagesConsumedCounter: -84
        pendingReadOps: 1
        readPosition: "861214:0"
      managedLedger:
        lastConfirmedEntry: "861214:1"
        ledgers:
          - key: 860650
            value: {"entries": 314}
          - key: 861214
            value: {"entries": 2}
          - key: 861295
            value: {"entries": 0}

From entriesReadCount: 84, we can see all entries have been read from BK. The number of entries can be calculated from lastMarkDeleteEntry and ledgers: 314 + 2 - 231 - 1 = 84, which also matches messagesConsumedCounter: -84.

Therefore, we can reach the 1st conclusion that all 84 entries have been read from BK.

However, readPosition is 861214:0.

Normally, there are 2 ledgers to read, while managed ledger can only read ledger by ledger. i.e. the readPosition will be updated 3 times:

1. 860650:314 when reading all 82 left entries of ledger 860650, the entry id is the last entry id + 1.
2. 861214:0 when detecting readPosition's entry id (314) is greater than the last entry id (313) of the current ledger (860650). See

   pulsar/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java

   Line 2212 in bc5b933

   opReadEntry.updateReadPosition(PositionFactory.create(nextLedgerId, 0));

3. 861214:2 when reading all 2 entries of ledger 861214

The single ManagedCursorImpl#asyncReadEntries method call will include two read operations (ManagedLedgerImpl#asyncReadEntries): one comes from ledger 860650, the other comes from ledger 861214.

Combined with the fact that pendingReadOps is 1, which should be reduced in

pulsar/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpReadEntry.java

Line 179 in bc5b933

cursor.readOperationCompleted();

pulsar/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java

Lines 3483 to 3484 in bc5b933

	void readOperationCompleted() {
	if (PENDING_READ_OPS_UPDATER.decrementAndGet(this) == 0) {

Combined with another fact that MessageDeduplication#snapshotCounter is 0, we can reach the 2nd conclusion: the asyncReadEntries call was stuck at the 2nd read so that the read callback was never triggered. This is the major difference from the case of partition 23, which should have triggered the read callback was triggered but an exception was thrown from the callback.

Since the stats (entriesReadCount) has been updated, this line must have been reached.

pulsar/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpReadEntry.java

Line 77 in bc5b933

cursor.updateReadStats(entriesCount, entriesSize);

Since the readPosition was not updated, this line must have not been reached:

pulsar/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpReadEntry.java

Line 96 in bc5b933

updateReadPosition(nexReadPosition);

The final result is that, an exception was thrown from the following code snippet:

        if (entriesCount != 0) { // entriesCount should be 2
            lastPosition = returnedEntries.get(entriesCount - 1).getPosition();
        } // lastPosition should be 861214:1

        List<Entry> filteredEntries = Collections.emptyList();
        if (entriesCount != 0) {
            filteredEntries = cursor.filterReadEntries(returnedEntries); // Exception was thrown HERE?
            entries.addAll(filteredEntries);
        }

        // if entries have been filtered out then try to skip reading of already deletedMessages in that range
        final Position nexReadPosition = entriesCount != filteredEntries.size()
                ? cursor.getNextAvailablePosition(lastPosition) : lastPosition.getNext();