If users are waiting, check your queues :)

I was at an on site meeting last week discussing with a client one of my favorite topics: "common mistakes" that Nuxeo developers make. And I realized there is one I have not mentioned in previous posts. (Actually there are many, but I want to keep my readers on the edge so I will just reveal one 🙂)

The mistake: queueing everything in the default Kafka topic in Nuxeo. This is not a mistake per se, but it can be problematic when optimizing, and it shows up when users complain about things taking too long ("why does it take so long to get my report?"). It might not sound that bad, but in reality if everything goes in the default topic, it will simply take longer to process. The good news: it is really easy to fix.

So how do things end up in the default queue in the first place? There are many cases where we want to run code asynchronously in a dedicated thread and transaction. For example: submit a translation request to an AI service every time a user marks a document ready for translation. In this case, starting a Work can fit better than using a Bulk Action.

public class SubmitAITranslationRequestWork extends AbstractWork {

    public static final String AI_TRANSLATION_CATEGORY = "aiTranslationRequest";

    public SubmitAITranslationRequestWork(String id) {
        super(id);
        setCategory(AI_TRANSLATION_CATEGORY);
    }

    @Override
    public String getCategory() {
        return AI_TRANSLATION_CATEGORY;
    }

    @Override
    public void work() {
        // call AI here
    }
}

This works great. It can retry if the LLM model is not answering, and it will not roll back the main transaction if it fails. In production, Nuxeo's WorkManager implementation that persists work and is cluster aware is the StreamWorkManager, so work is consumed through Kafka topics. Each queue has its own Kafka topic, and the queue used is based on the work category. If a worker's category is not assigned to a dedicated queue, it goes to the default queue, which means the default Kafka topic.

Nuxeo has already made a lot of improvements here. Before LTS 2021, many workers in the platform and asynchronous listeners used to land in the default queue. Now they have been moved to a common shared queue. You can see the categories that go to the common queue here (this one it's worth bookmarking because you might want to optimize this one too 🙂).

<#if "${nuxeo.work.queue.common.enabled}" == "true">
<queue id="common">
  <name>Common Shared Queue for Nuxeo Works</name>
  <maxThreads>${nuxeo.work.queue.common.threads}</maxThreads>
  <category>aceStatusUpdatedListener</category>
  <category>binary_metadata_work</category>
  <category>blobManagerDeleteMarkedBlobsListener</category>
  <category>checkedInCommentListener</category>
  <category>cleanOpenTasksOnWorkflowDone</category>
  <category>ConversionWork</category>
  <category>CounterListener</category>
  <category>deleteRoutingTaskListener</category>
  <category>docRemovedCommentListener</category>
  <category>documenttemplate-type-binding</category>
  <category>findRetentionExpired</category>
  <category>fulltextExtractor</category>
  <category>notificationListener</category>
  <category>nuxeoDriveGroupUpdateListener</category>
  <category>nuxeoDriveVirtualEventLoggerListener</category>
  <category>opchainpclistener</category>
  <category>orphanVersionRemoverListener</category>
  <category>permissionNotificationListener</category>
  <category>removeDocumentRoutesForDeletedDocument</category>
  <category>removeTasksForDeletedDocumentRoute</category>
  <#if "${nuxeo.work.queue.raclupdate.enabled}" != "true">
  <category>security</category>
  </#if>
  <category>storedRenditionsCleanup</category>
  <category>taggedVersionListener</category>
  <category>triggerEsclationRules</category>
  <category>unicityListener</category>
  <category>UserProfileImporterWork</category>
  <category>wopiDiscoveryRefreshListener</category>
  <category>wopiLocksExpiration</category>
  <category>workflowInstancesCleanup</category>
  <#if "${nuxeo.work.queue.thumbnail.enabled}" != "true">
  <category>updateThumbListener</category>
  </#if>
</queue>

So if you want more control and want to avoid queueing everything in default (and waiting for consumers on that topic), put your Work in a dedicated queue.

<extension target="org.nuxeo.ecm.core.work.service" point="queues">
  <queue id="aiTranslationQueue">
    <name>AI Translation Queue</name>
    <maxThreads>4</maxThreads>
    <category>aiTranslationRequest</category>
  </queue>
</extension>

You can take this further too. You can control which nodes process a given queue, for example by disabling the queue on some nodes and leaving it enabled on worker nodes. You definitely have more control than letting everything go into the default topic and then waiting for a huge lag to clear so users get their translations.

Monitoring tip: watch consumer lag and queue sizes for your queues. You can query the Nuxeo stream endpoint to inspect stream positions and lag, and of course your preferred tools for Kafka metrics and dashboards work fine too.

← All insights