[#695] Fix race in TraditionalWorkQueue.isIdle() (#688)

Valery Kharseko

11 hours ago 45900e1548fdf2cd81b31a6e328a0925e6f23c13

[#695] Fix race in TraditionalWorkQueue.isIdle() (#688)

 opendj-server-legacy/src/main/java/org/opends/server/extensions/TraditionalWorkQueue.java

@@ -13,6 +13,7 @@
 *
 * Copyright 2006-2010 Sun Microsystems, Inc.
 * Portions Copyright 2013-2016 ForgeRock AS.
 * Portions Copyright 2026 3A Systems, LLC
 */
package org.opends.server.extensions;

@@ -23,6 +24,7 @@
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;
@@ -65,6 +67,15 @@
  private AtomicLong opsSubmitted;

  /**
   * The number of operations that have been accepted by the work queue and are
   * not yet fully processed: this covers operations sitting in the queue as
   * well as operations currently handled by worker threads, including the
   * hand-off window in between, which is invisible to both the queue and the
   * worker thread activity flag.
   */
  private final AtomicInteger pendingOpsCount = new AtomicInteger();

  /**
   * The number of times that an attempt to submit a new request has been
   * rejected because the work queue is already at its maximum capacity.
   */
@@ -139,6 +150,7 @@
      killThreads = false;
      opsSubmitted = new AtomicLong(0);
      queueFullRejects = new AtomicLong(0);
      pendingOpsCount.set(0);

      // Register to be notified of any configuration changes.
      configuration.addTraditionalChangeListener(this);
@@ -302,6 +314,11 @@
  private void submitOperation(Operation operation,
      boolean blockEnqueuingWhenFull) throws DirectoryException
  {
    // Count the operation before enqueuing it so that isIdle() can never
    // observe an empty queue while the operation is in the process of being
    // submitted; the count is rolled back if the operation is rejected.
    pendingOpsCount.incrementAndGet();
    boolean submitted = false;
    queueReadLock.lock();
    try
    {
@@ -359,9 +376,14 @@
      }

      opsSubmitted.incrementAndGet();
      submitted = true;
    }
    finally
    {
      if (!submitted)
      {
        pendingOpsCount.decrementAndGet();
      }
      queueReadLock.unlock();
    }
  }
@@ -696,10 +718,12 @@
        if (pendingOperation != null)
        {
          pendingOperation.abort(cancelRequest);
          pendingOpsCount.decrementAndGet();
        }
        while ((pendingOperation = oldOpQueue.poll()) != null)
        {
          pendingOperation.abort(cancelRequest);
          pendingOpsCount.decrementAndGet();
        }
      }
      finally
@@ -714,28 +738,16 @@
  @Override
  public boolean isIdle()
  {
    queueReadLock.lock();
    try
    {
      if (!opQueue.isEmpty())
      {
        return false;
    return pendingOpsCount.get() == 0;
      }

      for (TraditionalWorkerThread t : workerThreads)
  /**
   * Notifies this work queue that a worker thread is done handling an
   * operation previously taken from the queue.
   */
  void operationDone()
      {
        if (t.isActive())
        {
          return false;
        }
      }

      return true;
    }
    finally
    {
      queueReadLock.unlock();
    }
    pendingOpsCount.decrementAndGet();
  }

  /**

 opendj-server-legacy/src/main/java/org/opends/server/extensions/TraditionalWorkerThread.java

@@ -13,7 +13,7 @@
 *
 * Copyright 2006-2010 Sun Microsystems, Inc.
 * Portions Copyright 2011-2016 ForgeRock AS.
 * Portions Copyright 2025 3A Systems, LLC.
 * Portions Copyright 2025-2026 3A Systems, LLC.
 */
package org.opends.server.extensions;

@@ -147,6 +147,8 @@
        }
        else
        {
          try
          {
          // The operation is not null, so process it.  Make sure that when
          // processing is complete.

@@ -173,6 +175,13 @@
            operation.getClientConnection().sendResponse(operation);
          }
        }
          finally
          {
            // Whatever the outcome, tell the queue this operation is no
            // longer pending so that isIdle() stays accurate.
            workQueue.operationDone();
          }
        }
      }
      catch (Throwable t)
      {

 opendj-server-legacy/src/test/java/org/opends/server/extensions/TraditionalWorkQueueTestCase.java

@@ -13,6 +13,7 @@
 *
 * Copyright 2006-2008 Sun Microsystems, Inc.
 * Portions Copyright 2013-2016 ForgeRock AS.
 * Portions Copyright 2026 3A Systems, LLC
 */
package org.opends.server.extensions;

@@ -125,7 +126,7 @@
   *
   * @throws  Exception  If an unexpected problem occurs.
   */
  @Test(groups = { "slow" })
  @Test
  public void testWaitUntilIdleNoOpsInProgress()
         throws Exception
  {
@@ -145,7 +146,7 @@
   *
   * @throws  Exception  If an unexpected problem occurs.
   */
  @Test(groups = { "slow" }, timeOut=10000)
  @Test(timeOut=10000)
  public void testWaitUntilIdleNoOpsInProgressNoTimeout()
         throws Exception
  {
@@ -165,7 +166,7 @@
   *
   * @throws  Exception  If an unexpected problem occurs.
   */
  @Test(groups = { "slow" })
  @Test
  public void testWaitUntilIdleSlowOpInProgress()
         throws Exception
  {
@@ -189,7 +190,7 @@
   *
   * @throws  Exception  If an unexpected problem occurs.
   */
  @Test(groups = { "slow" })
  @Test
  public void testWaitUntilTimeoutWithIdleSlowOpInProgress()
         throws Exception
  {