Fix the topic in fenced state and can not recover.

[codelipenghui]

Here is the log when the issue happens. The producer continues to reconnect to the broker, but the fenced state of the topic is always true.

19:01:42.351 [pulsar-io-4-1] INFO  org.apache.pulsar.broker.service.ServerCnx - [/10.24.34.151:48052][persistent://public/default/test-8] Creating producer. producerId=8
19:01:42.352 [Thread-174681] INFO  org.apache.pulsar.broker.service.ServerCnx - [/10.24.34.151:48052] persistent://public/default/test-8 configured with schema false
19:01:42.352 [Thread-174681] WARN  org.apache.pulsar.broker.service.AbstractTopic - [persistent://public/default/test-8] Attempting to add producer to a fenced topic
19:01:42.352 [Thread-174681] ERROR org.apache.pulsar.broker.service.ServerCnx - [/10.24.34.151:48052] Failed to add producer to topic persistent://public/default/test-8: producerId=8, org.apache.pulsar.broker.service.BrokerServiceException$TopicFencedException: Topic is temporarily unavailable

After check the heap dump of the broker, the pendingWriteOps is 5, this is the reason why the topic can not recover from the fenced state.

The topic will change to unfenced only the pendingWriteOps is 0, details can find at

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

Line 463 in 794aa20

if (pending == 0 && isFenced && !isClosingOrDeleting) {

But after checking the ML state of the topic, it shows the pendingAddEntries is 0 which not equals to pendingWriteOps of the topic.

The root cause is we are polling add entry op from the pendingAddEntries in multiple threads, one is the the ZK callback thread when complete the ledger creating (

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

Line 1406 in 794aa20

clearPendingAddEntries(status);

,

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

Line 1669 in 794aa20

void clearPendingAddEntries(ManagedLedgerException e) {

),
another one is the ML worker thread when complete the add entry op (

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

Line 181 in 794aa20

OpAddEntry firstInQueue = ml.pendingAddEntries.poll();

)

After the ongoing add entry op complete, but the corresponding op might been polled by the clearPendingAddEntries method. So it will poll another one, but due to
not equals to the current op, the polled op will not get a chance to be failed, so that the pendingWriteOps will not change to 0.

I have attached the complete logs for the topic: downloaded-logs-20210820-125853.csv

The fix is to complete the add entry op with ManagedLedgerException if the polled op is not equals to the current op.

[315157973]

LGTM

[lhotari]

@codelipenghui Do you think that this PR could fix #5284 ?

[lhotari]

LGTM

[lhotari]

@codelipenghui Doesn't the problem impact 2.7.x too? Can we cherry-pick to 2.7.4 ?

[eolivelli]

LGTM

I wound if it is the case to add a unit test around this method, in order to prevent regressions in the future. I am not talking about adding something that reproduces the issue, but to add some entry in pendingAddEntries and trigger safeRun

[codelipenghui]

@lhotari I have added the label release/2.7.4.

[eolivelli]

still LGTM
thank you @lhotari for your second pass

[lhotari]

I have one more question about this solution. Would it be necessary to call the callback for the OpAddEntry by using the "failed" method?

[michaeljmarshall]

+1. @codelipenghui, can you explain why we call firstInQueue.failed and then release this.data and call this.recycle() without also calling this.failed? Note that the failed method has concurrency controls built in by using the callbackUpdater. Wouldn't we want to use those concurrency controls here as well?

    public void failed(ManagedLedgerException e) {
        AddEntryCallback cb = callbackUpdater.getAndSet(this, null);
        if (cb != null) {
            ReferenceCountUtil.release(data);
            cb.addFailed(e, ctx);
            ml.mbean.recordAddEntryError();
        }
    }

[codelipenghui]

@lhotari @michaeljmarshall As I mentioned in the PR description, there are 2 threads polling the add entry op from the queue, but here poll out the op from the queue but does not call the callback method of the op, this will lead to the pendingWriteOps of the topic can't change to 0 (before adding an entry, pendingWriteOps++. after entry add success for failed, pendingWriteOps--), so we should call the callback of the op here to make sure the pendingWriteOps can change to 0. The root cause of the issue is we missed some op callback calls here because the first op in the queue is not equaled to the current OpAddEntry(the previous behavior just throws an exception). So It's not about concurrency controls of the callbackUpdater.

[michaeljmarshall]

Thank you for your clarification. I think I understand why we're calling firstInQueue.failed. I don't understand why we're calling this.recycle without also calling addComplete or failed. If I understand correctly, by calling this.recycle, we're going to set callback = null; without calling its callback. If my understanding is incorrect, can you explain why we're recycling this without calling its callback? Thank you.

[lhotari]

there are 2 threads polling the add entry op from the queue

@codelipenghui The 2 threads polling seem to be part of the problem. Is that so?

One goal of the Managed Ledger design seems to be that there's a single thread per managed ledger executing operations. PR #11387 "Pin executor and scheduled executor threads for ManagedLedgerImpl" fixes some issues where this pinned executor solution wasn't used.

Just curious, would #11387 prevent the 2 threads polling the add entry op from the queue?

[codelipenghui]

Oh, I think I made a wrong change there, the OpAddEntry.failed() has released the data but looks like missed the object recycle, So we don't need to release it again.

So a thread takes the op from the queue, it should call the callback and release the data.

[codelipenghui]

@lhotari Yes, I think run the clearPendingAddEntries() in the pinned executor can also fix the problem

[lhotari]

Oh, I think I made a wrong change there, the OpAddEntry.failed() has released the data but looks like missed the object recycle, So we don't need to release it again.

So a thread takes the op from the queue, it should call the callback and release the data.

@codelipenghui Why isn't this.failed(...) called when this != firstInQueue ?

I would be expecting something like this:

        if (this != firstInQueue) {
            if (firstInQueue != null) {
                firstInQueue.failed(
                        new ManagedLedgerException("Unexpected add entry op when complete the add entry op."));
            }
            this.failed(new ManagedLedgerException("Unexpected add entry op when complete the add entry op."));
            return;
        }

Is there a reason to not call this.failed?

[lhotari]

@codelipenghui reading your previous answer, I would understand that clearPendingAddEntries is expected to call failed?

Since failed method has concurrency controls, I think it wouldn't cause any harm to call this.failed when this != firstInQueue. No?

[codelipenghui]

Yes, clearPendingAddEntries is expected to call failed.
And yes no harm to call this.failed, I think we only complete the op which comes from the queue is more clear here. All the add entry ops will add the queue, so someone poll the op from the queue, it should make sure the op.callback has been called so that we don't need to care about this.failed, we just need to complete all the ops in the queue.

[michaeljmarshall]

+1. @codelipenghui, can you explain why we call firstInQueue.failed and then release this.data and call this.recycle() without also calling this.failed? Note that the failed method has concurrency controls built in by using the callbackUpdater. Wouldn't we want to use those concurrency controls here as well?

    public void failed(ManagedLedgerException e) {
        AddEntryCallback cb = callbackUpdater.getAndSet(this, null);
        if (cb != null) {
            ReferenceCountUtil.release(data);
            cb.addFailed(e, ctx);
            ml.mbean.recordAddEntryError();
        }
    }

[hangc0276]

Nice catch!

[michaeljmarshall]

LGTM

[krishhh5]

the reason why Topic isTemporarily unavailable is the value “isFenced” is true。the code firstInQueue.failed(newManagedLedgerException) still set isFenced=true because of the addFauled method

public synchronized void addFailed(ManagedLedgerException exception, Object ctx) {
        if (exception instanceof ManagedLedgerFencedException) {
            // If the managed ledger has been fenced, we cannot continue using it. We need to close and reopen
            close();
        } else {
            // fence topic when failed to write a message to BK
            fence();

why not fix like this
firstInQueue.failed(newManagedLedgerFencedException)`
to set isFenced=false

[codelipenghui]

@krishhh5

The topic will be closed if we use new ManagedLedgerFencedException. All the producers and consumers will be disconnected. We don't want to close the topic to recover the topic from the wrong state. Instead, we can just make sure the managed ledger will not miss any callback for add entry operations.