[#1252] Go back to the changelog when the queue of a following handler drop...

Valery Kharseko

93 mins ago d27c892c81f6df051f737f54298cbdb530881d35

 opendj-server-legacy/src/main/java/org/opends/server/replication/server/MessageHandler.java

@@ -513,9 +513,16 @@
          {
            return null;
          }
          if (msgQueue.isEmpty())
          if (msgQueue.isEmpty() || !following)
          {
            // this handler is going back to the changelog, see stopFollowingWhenChangesAreMissing()
            /*
             * This handler is going back to the changelog: changes are missing, see
             * stopFollowingWhenChangesAreMissing(), or the queue dropped its oldest changes in
             * add() while this thread waited without the monitor - more changes arrived than
             * it holds. Its head is then the first change after the dropped ones: taken now, it
             * would move the state of the consumer past them, and the changelog would be read
             * from after them, so they would never be sent - see issue #1252.
             */
            continue;
          }
          UpdateMsg msg = msgQueue.removeFirst();

 opendj-server-legacy/src/test/java/org/opends/server/replication/server/MessageHandlerCatchUpTest.java

@@ -20,6 +20,7 @@
import static org.opends.server.TestCaseUtils.TEST_ROOT_DN_STRING;
import static org.opends.server.util.CollectionUtils.newArrayList;

import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.List;
import java.util.TreeSet;
@@ -516,6 +517,82 @@
  }

  /**
   * The case of #1252: the writer of a handler which follows its queue waits on it, empty, and
   * more changes arrive than the queue holds before the writer has the monitor back. The queue
   * drops its oldest changes and stops following; the writer must then go back to the changelog
   * for them rather than take the head of the queue, which is the first change after them.
   * <p>
   * Nothing holds the writer off the monitor on cue but the monitor itself, so the test takes it
   * while the changes arrive, as the threads which add changes kept it from the writer in the run
   * of #1252.
   */
  @Test
  public void changesTheQueueDroppedWhileTheWriterWaitedOnItAreSent() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    ReplicationServer replicationServer = null;
    final ExecutorService writer = Executors.newSingleThreadExecutor();
    try
    {
      replicationServer = newReplicationServer("messageHandlerWaitingDb");
      final List<UpdateMsg> changesOfA = changesOfA(baseDN, TimeThread.getTime() - 60000, 1, 15);
      final MessageHandler handler = newHandler(replicationServer, baseDN, 10, new ServerState());

      // the changelog is empty: the writer has nothing to catch up on, follows the queue and waits on it
      final Future<Thread> writerThread = writer.submit(Thread::currentThread);
      final Future<UpdateMsg> first = writer.submit(handler::getNextMessage);
      waitUntilItWaitsOnItsQueue(handler, writerThread.get());

      synchronized (queueOf(handler))
      {
        for (UpdateMsg change : changesOfA)
        {
          arrive(replicationServer, handler, change);
        }
        assertThat(handler.isFollowing())
            .as("the queue did not drop any change: the test shows nothing")
            .isFalse();
      }

      final List<UpdateMsg> sent = new ArrayList<>();
      sent.add(first.get(NEXT_MESSAGE_TIMEOUT_MS, TimeUnit.MILLISECONDS));
      for (Sent next : nextMessages(handler, changesOfA.size() - 1))
      {
        sent.add(next.msg);
      }
      assertThat(csnsOf(sent))
          .as("the consumer was not sent every change, in order")
          .containsExactlyElementsOf(csnsOf(changesOfA));
    }
    finally
    {
      writer.shutdownNow();
      removeQuietly(replicationServer);
    }
  }

  /** Waits until the writer waits on the queue of the handler, which it then follows. */
  private static void waitUntilItWaitsOnItsQueue(MessageHandler handler, Thread writer) throws Exception
  {
    final long deadline = System.currentTimeMillis() + NEXT_MESSAGE_TIMEOUT_MS;
    while (!handler.isFollowing() || writer.getState() != Thread.State.TIMED_WAITING)
    {
      assertThat(System.currentTimeMillis())
          .as("the writer did not come to wait on the queue it follows")
          .isLessThan(deadline);
      Thread.sleep(10);
    }
  }

  /** The queue of the handler, the monitor of which its writer waits on. */
  private static Object queueOf(MessageHandler handler) throws Exception
  {
    final Field queue = MessageHandler.class.getDeclaredField("msgQueue");
    queue.setAccessible(true);
    return queue.get(handler);
  }

  /**
   * A queue of {@link #size} changes which drops a change the consumer lacks, then takes a change
   * of B with an older CSN in the room the drop left - which only a queue bounded by its size in
   * bytes leaves, as it drops whole changes: one large change, dropped, leaves room for a small one.