deduplicationStatus failed for one of the partition of the topic

[adarshhm6]

Deduplication enabled at broker level using param brokerDeduplicationEnabled.

When checked for topic stats for 3 partitioned topic using
pulsar-admin topics stats persistent://{tenant}/{namespace}/topic-partition-{i}

2 partitions shows "deduplicationStatus" : "Enabled", while 1 partition shows "deduplicationStatus" : "Failed"

As a result backlogSize increased on one topic partition alone.
Logs got rolled over and I do not have have any failure traces.

Pulsar version is 2.10

Is it a known case where deDup fail for only one partition and no other topics under that broker gets impacted or faced any failures?

Please suggest.

[BewareMyPower]

Yes, you should check logs for why it's failed more info. When deduplication is enabled, it will perform a topic replay rather than simply setting the deduplication status to Enabled.

[BewareMyPower]

So highestSequencedPersisted will be pulsar-70-923 !?

No. "pulsar-70-923" is the producer name and 27455316 is the latest sequence id at the mark-delete position (146914057:-1). Normally, the workflow should be:

1. Loading the map { "pulsar-70-923": 27455316, "pulsar-70-186": 27606884, "pulsar-70-861" : 27473943 } as the initial value of highestSequencedPersisted
2. Reading entries from the mark-delete position 146914057:-1, for each entry:
   1. Parse the message metadata from the buffer
   2. Parse the producer name and sequence id
   3. Update producer_name => sequence_id in highestSequencedPersisted
3. Stop reading entries when there are no more entries.

As you can see, the readPosition is 147505614:682, while this ledger has 1045 entries. It means the step 2 is interrupted, typically by an exception throwing in the callback:

        managedCursor.asyncReadEntries(100, new ReadEntriesCallback() {
            @Override
            public void readEntriesComplete(List<Entry> entries, Object ctx) {
                /* ... */
                // NOTE: if an exception is thrown before the code below, the `future` will never complete
                if (managedCursor.hasMoreEntries()) {
                    pulsar.getExecutor().execute(() -> replayCursor(future));
                } else {
                    future.complete(lastPosition);
                }
            }

which could be fixed by #24511 that eliminates the possibility of such cases. You can also see waitingReadOp is false, which means it's not blocked at reading entries.

When tried to manually produce a dummy message after restarting brokers, we observed that it was still failing with Failed to load topic within timeout.

Yes, this error is confusing, see the motivation in #24500. In short, if a topic loading is blocked forever, accessing that topic will always fail with "Failed to load topic within timeout", including both admin and client operations. But we should try to avoid this case (topic loading is blocked forever) rather than clearing the cache, which might result in two PersistentTopic instances existing at the same time. That's why I closed that PR.

Finally to resolve it had to do it manually with removal of deDup and unloading the topic.

Yes, unloading the topic or its namespace could resolve this issue by clearing the cache and triggering the reloading of that topic

[adarshhm6]

Thank you for the response @BewareMyPower

As you can see, the readPosition is 147505614:682, while this ledger has 1045 entries. It means the step 2 is interrupted, typically by an exception throwing in the callback:

I believe you meant the ledger has 1060 entries because 1045 entries were present in ledger 146914057 and not in 47505614.

ledgerId is different for markDeletePosition and readPosition

And mark-delete position 146914057:-1 -->
This ledger has 1045 entries and -1 specifies no messages has been read and acknowledged yet.

And as you mentioned "readPosition" : "147505614:682" points to some exception in between and got interrupted as this ledger has 1060 entries.

You also mentioned since waitingReadOp is false, it is not blocked for reading entries, so why the entries were not read completely if it's not blocked. ?

Since you mentioned your PR - #24511 will fix these issues(deDuplication failures will be avoided completely and will it be successful?) and can be backported but your next statement was confusing and contradictory. Please suggest on that.

[BewareMyPower]

so why the entries were not read completely if it's not blocked. ?

Because it's a read -> process loop, the previous code handles the loop like:

    private void replayCursor(CompletableFuture<Position> future) {
        managedCursor.asyncReadEntries(100, new ReadEntriesCallback() {
            @Override
            public void readEntriesComplete(List<Entry> entries, Object ctx) {
                Position lastPosition = null;
                for (Entry entry : entries) {
                    /* ... */
                    entry.release();
                }

                // NOTE: if an exception was thrown before this line, then the future will never have a chance to complete
                if (managedCursor.hasMoreEntries()) {
                    // Read next batch of entries
                    pulsar.getExecutor().execute(() -> replayCursor(future));
                } else {
                    // Done replaying
                    future.complete(lastPosition);
                }
            }

            @Override
            public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
                future.completeExceptionally(exception);
            }
        }, null, PositionFactory.LATEST);

[adarshhm6]

@BewareMyPower Thank you for giving more insights.

So, earlier 100 entries will be read by default in a for loop and if any exception while reading in loop then next batch of entries will not be read.

And now with your fix #24511 , this interruption will be handled and entries will be read completely and deduplication will be enabled. Is my understanding correct?

[BewareMyPower]

Yes