OpcUaSubscriptionManager processingQueue is Blocking

[ixiaojingling]

Describe the bug
My project is out of memory, from dump log, we find class OpcUaSubscriptionManager processingQueue is blocking.

Expected behavior
When client receive disorderd PublishResponse sequenceNumber, it will not block processingQueue.

Logs and Packet Captures

milo log
09:29:55.143|W||15064|milo-shared-thread-pool-12721|o.e.m.o.s.c.s.OpcUaSubscriptionManager|[id=405] expected sequence=5957790, received sequence=5957791. Calling Republish service...
09:29:56.675|W||15064|milo-shared-thread-pool-12702|o.e.m.o.s.c.s.OpcUaSubscriptionManager|[id=405] expected sequence=5957791, received sequence=5957795. Calling Republish service...

dump log
wo find the processingQueue occupying over 1GB of memory，the first blocking member is :
publising time : DateTime{utcTime=134022005955438390, javaDate=Sat Sep 13 09:29:55 CST 2025}
subscriptionId: 405
sequenceNumber: 5957795
processingQueue pause: true
processingQueue pending: 0

The content of dump log is match milo log .
Is the milo log "expected sequence= xxx" print twice in a row, it may cause the problem ?

if (sequenceNumber > expectedSequenceNumber) {
            logger.warn("[id={}] expected sequence={}, received sequence={}. Calling Republish service...",
                subscriptionId, expectedSequenceNumber, sequenceNumber);

            processingQueue.pause();
            processingQueue.submitToHead(() -> onPublishComplete(response, pendingCount));

            republish(subscriptionId, expectedSequenceNumber, sequenceNumber).whenComplete((dataLost, ex) -> {
                if (ex != null) {
                    logger.debug("Republish failed: {}", ex.getMessage(), ex);

                    subscriptionListeners.forEach(l -> l.onNotificationDataLost(subscription));
                    subscription.getNotificationListeners().forEach(l -> l.onNotificationDataLost(subscription));
                } else {
                    // Republish succeeded, possibly with some data loss, resume processing.
                    if (dataLost) {
                        subscriptionListeners.forEach(l -> l.onNotificationDataLost(subscription));
                        subscription.getNotificationListeners().forEach(l -> l.onNotificationDataLost(subscription));
                    }
                }

                subscription.setLastSequenceNumber(sequenceNumber - 1);
                processingQueue.resume();
            });

            return;
        }

Additional context
milo version is v0.6.13.
my project is an client.

[kevinherron]

Seems like it could only happen if processingQueue.resume() is not getting called.

One possibility is that one of the onNotificationDataLost callbacks into user code blocks indefinitely. If you have a thread dump, or if this problem is reproducible, take a look to see if there's a thread blocked in user code originating from one of those callbacks.

Another is that republish() simply never completes, but since this has never been reported as an issue over all the years it seems a lot less likely.

[ixiaojingling]

In my projects, I implements UaSubscription.NotificationListener when create Subscription,but when print log 09:29:55.143, we not find
onNotificationDataLost log, the thid line log we find a watchdog error(It is my mistake). so the cause maybe republish() simply never completes
in log, I never found thread-name 'milo-shared-thread-pool-12721' and 'milo-shared-thread-pool-12702' printed again

log
09:29:55.143|W||15064|milo-shared-thread-pool-12721|o.e.m.o.s.c.s.OpcUaSubscriptionManager|[id=405] expected sequence=5957790, received sequence=5957791. Calling Republish service...
09:29:56.675|W||15064|milo-shared-thread-pool-12702|o.e.m.o.s.c.s.OpcUaSubscriptionManager|[id=405] expected sequence=5957791, received sequence=5957795. Calling Republish service...
09:30:51.232|E||15064|milo-shared-thread-pool-12734|o.e.m.o.s.c.Stack|Uncaught Throwable on shared stack ExecutorService thread!
java.lang.StringIndexOutOfBoundsException
at com.xkl.hy.hal.protocol.opcua.subscribe.SubNotifierListener.onSubscriptionWatchdogTimerElapsed

public class SubNotifierListener implements UaSubscription.NotificationListener {
    @Override
    public void onNotificationDataLost(UaSubscription subscription) {
        printLog(subscription, "onNotificationDataLost");
    }

    @Override
    public void onSubscriptionWatchdogTimerElapsed(UaSubscription subscription) {
        printLog(subscription, "onSubscriptionWatchdogTimerElapsed");
    }
}

[kevinherron]

If you are interested in actually fixing this, you'll have to do it in your own fork. I would start by adding some logging to the republish() methods to help track down how far it gets and if that's really happening.

Having the OpcUaSubscriptionManager logger set to DEBUG for a while would be useful too, though I understand it will create a lot of log messages.

[ixiaojingling]

I find the reason, In my projects, I implements UaSubscription.NotificationListener when create Subscription.
when method onNotificationDataLost() on call, there's a bug in my code, will throw exception, method republish() will failed, processingQueue is alwasy in paused.
When CompletableTableFuture method whenComplete() happen exception, it cannot be caught by Thread setUncaughtExceptionHandler, so there is not log to appear the cause of the problem
I suggest adding a method exceptionally() after method whenComplete() , What do you think?

there is code in class OpcUaSubscriptionManager method onPublishComplete

if (sequenceNumber > expectedSequenceNumber) {
            logger.warn("[id={}] expected sequence={}, received sequence={}. Calling Republish service...",
                subscriptionId, expectedSequenceNumber, sequenceNumber);

            processingQueue.pause();
            processingQueue.submitToHead(() -> onPublishComplete(response, pendingCount));

            republish(subscriptionId, expectedSequenceNumber, sequenceNumber).whenComplete((dataLost, ex) -> {
                if (ex != null) {
                    logger.debug("Republish failed: {}", ex.getMessage(), ex);

                    subscriptionListeners.forEach(l -> l.onNotificationDataLost(subscription));
                    subscription.getNotificationListeners().forEach(l -> l.onNotificationDataLost(subscription));
                } else {
                    // Republish succeeded, possibly with some data loss, resume processing.
                    if (dataLost) {
                        subscriptionListeners.forEach(l -> l.onNotificationDataLost(subscription));
                        subscription.getNotificationListeners().forEach(l -> l.onNotificationDataLost(subscription));
                    }
                }

                subscription.setLastSequenceNumber(sequenceNumber - 1);
                processingQueue.resume();
            });

            return;
        }

[kevinherron]

Yes, or doing listener notification inside a try/catch block.