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.