[#1036] Assert that a parked change is handed out, not which thread hands i...

Valery Kharseko

15 hours ago d0082246b9d667fb620c5d4b9c7d8e5ad01e1d81

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

@@ -3199,10 +3199,22 @@
   * handed out: the changes parked behind it stayed parked, and the ServerState of this
   * domain stayed behind them until some other change was replayed here.
   * <p>
   * What tells the two apart is which thread replays the parked change. It is handed to
   * whichever thread cleared the change it was waiting for, so it is replayed by the very
   * thread which has just committed the change whose ack threw - a change nobody handed out
   * is replayed by no one at all, and the wait below is what says so.
   * What tells the two apart is that the parked change is replayed at all. The child is
   * seen parked before the parent is let go, so the replay queue is not where it can come
   * from anymore: only {@code getNextUpdate()} hands it out, and with the parent held until
   * then, the thread which committed the parent is, as a rule, the one left to call it - a
   * change nobody handed out is replayed by no one at all, and the wait below is what says so.
   * Which thread replays it is deliberately not asserted: {@code getNextUpdate()} hands a
   * parked change to whichever thread calls it first once the changes before it are gone,
   * and the thread which parked it calls it on its own way out, so a parker which is slow
   * to get there takes the child back itself when the parent commits in between.
   * <p>
   * The pin this gives issue #922 holds on that first arm alone: nothing here orders the
   * parker's own {@code getNextUpdate()} call against the parent being let go, and a parker
   * delayed past the parent's release, commit and ack takes the child back on its own way
   * out instead - measured with a mutant, 500 ms after {@code checkDependencies()} parks the
   * child. A revert of the fix this test is for goes uncaught on that arm: the wait above
   * sees the child parked either way, and only the parent's road runs the code #922 is about.
   */
  @Test
  public void theChangesParkedBehindAChangeWhoseAckFailedAreReplayed() throws Exception
@@ -3244,25 +3256,30 @@
        OperationType.ADD, "PreParse",
        op -> parentCsn.equals(OperationContext.getCSN(op))
            || childCsn.equals(OperationContext.getCSN(op)));
    final long initialDependent = getMonitorAttrValue(baseDN, "dependent-changes-size");
    try
    {
      domain.processUpdate(new AddMsgWhoseAckThrows(parentCsn, parent.getName(), parentUUID,
          baseUUID, parent.getObjectClassAttribute(), parent.getAllAttributes()));
      final Thread replayingParent = parked.awaitParked(60, SECONDS);
      parked.awaitParked(60, SECONDS);

      domain.processUpdate(new AddMsg(childCsn, child.getName(), childUUID, parentUUID,
          child.getObjectClassAttribute(), child.getAllAttributes(), null));
      /*
       * Seen parked before the parent is let go. Released on the spot, the parent could be
       * applied and committed before a replay thread has taken the child off the queue at
       * all, and the child would then be replayed from the queue with nothing to wait for -
       * a pass which says nothing about the hand-out this test is about.
       */
      assertMonitorAttrValueEventually(baseDN, "dependent-changes-size", initialDependent + 1,
          "the child must be parked behind the parent while the parent is held");

      /*
       * The parent is applied and its ack throws where it is published. The replay carries
       * on all the same, and the child is the change it hands itself next.
       * on all the same, and the child is the change it hands out next.
       */
      parked.release();
      final Thread replayingChild = parked.awaitParked(60, SECONDS);
      Assertions.assertThat(replayingChild)
          .as("the change which was parked must be replayed by the thread which cleared what"
              + " it was waiting for, rather than be left waiting")
          .isSameAs(replayingParent);
      parked.awaitParked(60, SECONDS);
      parked.release();

      assertNotNull(getEntry(child.getName(), 30000, true),
@@ -3279,19 +3296,23 @@
   * replay it was handed to is unwound.
   * <p>
   * A change which was parked behind another one is handed out by {@code getNextUpdate()}
   * to the thread which cleared what it was waiting for, and that thread owns it from then
   * on. The give-back on the way out of an unwound replay asks which change this thread
   * owns, so the hand-out has to be recorded where that question is answered, not only on
   * the change: left out, the change would stay owned by a thread which is not replaying
   * it anymore, and every later delivery of it would be refused as a duplicate - the wedge
   * of this issue, on the dependency road.
   * to whichever thread calls it first once the changes before it are gone, and that
   * thread owns it from then on. The give-back on the way out of an unwound replay asks
   * which change this thread owns, so the hand-out has to be recorded where that question
   * is answered, not only on the change: left out, the change would stay owned by a thread
   * which is not replaying it anymore, and every later delivery of it would be refused as
   * a duplicate - the wedge of this issue, on the dependency road.
   * <p>
   * The parent is held at the pre-parse plugin point while the child is delivered, so the
   * child is parked behind a change in flight, and the thread which is thrown out of the
   * child is read at the same plugin point: it must be the one which committed the parent,
   * which is what says the child was handed out rather than taken off the queue. The child
   * is thrown out of once, inside the replay, and then unwound past its ack, on the road
   * every catch of the replay has already run on.
   * The parent is held at the pre-parse plugin point while the child is delivered, and it
   * is let go only once the child is seen parked behind it: that is what says the child was
   * handed out rather than taken off the queue, since a parked change leaves by
   * {@code getNextUpdate()} and by no other road. The thread which is thrown out of the
   * child is read at the same plugin point, for the assertion that the Error did not end
   * it; which thread it is says nothing about the hand-out and is not asserted - the parker
   * calls {@code getNextUpdate()} on its own way out, so it takes the child back itself
   * when the parent commits before it gets there. The child is thrown out of once, inside
   * the replay, and then unwound past its ack, on the road every catch of the replay has
   * already run on.
   */
  @Test
  public void aChangeHandedOutAsADependencyIsGivenBackWhenItsReplayIsUnwound() throws Exception
@@ -3324,6 +3345,7 @@
    final CSN parentCsn = gen.newCSN();
    final CSN childCsn = gen.newCSN();
    final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed");
    final long initialDependent = getMonitorAttrValue(baseDN, "dependent-changes-size");
    final ParkedReplay parked = ShortCircuitPlugin.parkReplayedOperations(
        OperationType.ADD, "PreParse", op -> parentCsn.equals(OperationContext.getCSN(op)));
    /*
@@ -3340,7 +3362,13 @@
          {
            return false;
          }
          replayingChild.set(Thread.currentThread());
          /*
           * The first replay of the child is the one which meets the Error, and it is the
           * one kept: this is evaluated ahead of the budget of the throw, so it runs on the
           * by-hand redelivery below too, which must not overwrite the thread the
           * assertion of issue #923 is about.
           */
          replayingChild.compareAndSet(null, Thread.currentThread());
          return true;
        },
        () -> new LinkageError("the replay of the change which was handed out meets an Error"),
@@ -3349,12 +3377,19 @@
    {
      domain.processUpdate(new AddMsg(parentCsn, parent.getName(), parentUUID, baseUUID,
          parent.getObjectClassAttribute(), parent.getAllAttributes(), null));
      final Thread replayingParent = parked.awaitParked(60, SECONDS);
      parked.awaitParked(60, SECONDS);

      domain.processUpdate(new AddMsgWhoseReplayIsUnwoundAfterItsAck(childCsn, child.getName(),
          childUUID, parentUUID, child.getObjectClassAttribute(), child.getAllAttributes()));
      /*
       * Seen parked before the parent is let go: a child taken off the queue once the
       * parent has committed is replayed with nothing to wait for, and the give-back it
       * would then exercise is the one of the ordinary road rather than of the hand-out.
       */
      assertMonitorAttrValueEventually(baseDN, "dependent-changes-size", initialDependent + 1,
          "the child must be parked behind the parent while the parent is held");

      // The parent is applied, and the child is the change its thread hands itself next.
      // The parent is applied, and the child is the change handed out next.
      parked.release();

      TestTimer timer = new TestTimer.Builder()
@@ -3370,10 +3405,6 @@
              "the change which was handed out must have been thrown out of");
        }
      });
      Assertions.assertThat(replayingChild.get())
          .as("the change which was parked must be replayed by the thread which cleared what"
              + " it was waiting for: that is the hand-out this test is about")
          .isSameAs(replayingParent);

      /*
       * The child was thrown out of and its replay was then unwound, so it is not in the
@@ -3406,7 +3437,7 @@
              + " from the one which was unwound");
      assertEquals(getMonitorAttrValue(baseDN, "replayed-updates-failed"), initialFailures,
          "a change which was delivered again must not be counted as one this replica gave up on");
      assertTrue(replayingParent.isAlive(),
      assertTrue(replayingChild.get().isAlive(),
          "an Error which unwinds a replay must not end the thread which met it (issue #923)");
    }
    finally
@@ -3591,7 +3622,8 @@
   * A change which waits for another one is parked and stays owned by the replay thread
   * which parked it, while that thread goes on to the changes which follow: it is handed
   * out again by {@code getNextUpdate()}, which every replay loop of this domain runs once
   * it is done, so it is replayed by whichever thread clears the change it was waiting for.
   * it is done, so it is replayed by whichever thread calls it first once the change it was
   * waiting for is gone - the thread which cleared it, as a rule.
   * A replay which is unwound leaves the thread which parked it without that road - it
   * takes the next delivery off the shared queue instead, and never comes back to the
   * change it parked - and every redelivery of a change a replay thread owns is refused as
@@ -3727,8 +3759,8 @@
                  + " it restarts the session for the changes it gave back");

          /*
           * The barrier is lifted, which lets the ServerState past it and hands the parked
           * change to the thread which cleared it.
           * The barrier is lifted, which lets the ServerState past it and has the parked
           * change handed out.
           */
          giveUpOn(domain, failing, waitedOn, failingMods, waitedOnUUID);
          checkEntryHasAttributeValue(waitedOn, "description", parkedDescription, 30,
@@ -3814,13 +3846,14 @@
   * <p>
   * A parked change stays owned by the replay thread which parked it while that thread
   * goes back to the pool and takes the changes which follow: {@code getNextUpdate()} is
   * what hands it out again, to whichever replay thread clears the change it was waiting
   * for. Changing the number of replay threads stops the whole pool and creates another
   * one, so a thread which parked a change and went back to the queue is joined while it
   * is idle, and it would end still recorded as the owner of that change - a thread which
   * does not exist anymore, while every redelivery of a change a replay thread owns is
   * refused as a duplicate. On a domain which then goes quiet that change is where this
   * replica's ServerState, and every change behind it from every master, stops.
   * what hands it out again, to whichever thread calls it first once the change it was
   * waiting for is gone - the thread which cleared it, as a rule. Changing the number of
   * replay threads stops the whole pool and creates another one, so a thread which parked
   * a change and went back to the queue is joined while it is idle, and it would end still
   * recorded as the owner of that change - a thread which does not exist anymore, while
   * every redelivery of a change a replay thread owns is refused as a duplicate. On a
   * domain which then goes quiet that change is where this replica's ServerState, and
   * every change behind it from every master, stops.
   */
  @Test
  public void aChangeParkedByAThreadThePoolStoppedIsDeliveredAgain() throws Exception