[#949] Report a ReplicaOfflineMsg the broker refused as not sent (#976)

Valery Kharseko

yesterday 13d57e063cbd3e884f787744a9eaedbfb98b5a51

[#949] Report a ReplicaOfflineMsg the broker refused as not sent (#976)

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

@@ -2221,8 +2221,9 @@
    else if (logger.isTraceEnabled())
    {
      logger.trace("Replica " + getServerId() + " of domain baseDN=" + getBaseDN()
          + " could not announce itself offline: a change which is still in flight holds"
          + " the message back, and " + pendingChanges.size() + " change(s) are pending");
          + " could not announce itself offline: the message was not published - a change which"
          + " is still in flight holds it back, or the broker had no session to write it to,"
          + " and " + pendingChanges.size() + " change(s) are pending");
    }
  }


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

@@ -132,8 +132,8 @@
   * Such a message is given up on rather than left queued, so that it is neither reported as
   * sent nor published later on the session which follows.
   *
   * @return the CSN of the message which was published, or {@code null} if it could not be
   *         published
   * @return the CSN of the message which was published, or {@code null} if it could not be:
   *         a change which is still in flight holds it back, or the broker refused it
   */
  public synchronized CSN putReplicaOfflineMsg()
  {
@@ -143,23 +143,36 @@
    pendingChange.setCommitted(true);

    pendingChanges.put(offlineCSN, pendingChange);
    pushCommittedChanges();
    // pushCommittedChanges() removes whatever it published, so the message is still listed
    // here if and only if a change before it held it back.
    final boolean heldBack = pendingChanges.remove(offlineCSN) != null;
    return heldBack ? null : offlineCSN;
    /*
     * The message is the last change of the queue, so a push which did not reach it, or which
     * the broker refused, reports another CSN or none.
     */
    final boolean published = offlineCSN.equals(pushCommittedChanges());
    // pushCommittedChanges() removes whatever it reached, so the message is still listed here
    // if and only if a change before it held it back - it is dropped rather than left queued.
    pendingChanges.remove(offlineCSN);
    return published ? offlineCSN : null;
  }

  /**
   * Push all committed local changes to the replicationServer service.
   *
   * @return the CSN of the last {@link ReplicaOfflineMsg} the replication service accepted, or
   *         {@code null} if none was pushed or the broker refused it. The announcement that a
   *         replica goes offline is the one message whose delivery the caller must know about:
   *         it is stored nowhere, so nothing publishes it again, while a change the broker
   *         refuses is republished from the historical information of its entry on the next
   *         session.
   */
  synchronized void pushCommittedChanges()
  synchronized CSN pushCommittedChanges()
  {
    CSN publishedOfflineCSN = null;

    // peek the oldest change
    Entry<CSN, PendingChange> firstEntry = pendingChanges.firstEntry();
    if (firstEntry == null)
    {
      return;
      return null;
    }

    PendingChange firstChange = firstEntry.getValue();
@@ -185,7 +198,10 @@
      }
      else if (msg instanceof ReplicaOfflineMsg)
      {
        domain.publish(msg);
        if (domain.publish(msg))
        {
          publishedOfflineCSN = msg.getCSN();
        }
      }

      // false warning: firstEntry will not be null if firstChange is not null
@@ -195,6 +211,7 @@
      firstEntry = pendingChanges.firstEntry();
      firstChange = firstEntry != null ? firstEntry.getValue() : null;
    }
    return publishedOfflineCSN;
  }

  /**

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

@@ -2436,7 +2436,12 @@
        }
      }
    }
    return true;
    /*
     * The loop also ends when the broker is stopped, and then nothing was written to any
     * session: a caller told the message was published would report as sent a message which
     * never reached the wire.
     */
    return done;
  }

  /**

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

@@ -3696,15 +3696,25 @@
   * {@link #processUpdate(UpdateMsg)} message.
   *
   * @param msg The UpdateMsg that should be published.
   * @return {@code true} if the message was written to the session of the
   *         Replication Service, {@code false} if the broker could not write
   *         it: it has no usable session, the changes which come before this
   *         one still have to be republished, or it was stopped in between.
   */
  public void publish(UpdateMsg msg)
  public boolean publish(UpdateMsg msg)
  {
    broker.publish(msg);
    final boolean published = broker.publish(msg, true);
    /*
     * The domain state is updated whether or not the message was written: it says which changes
     * this replica has done, and it is by finding it ahead of the state its replication server
     * reports that the next session knows which changes to republish.
     */
    if (msg.contributesToDomainState())
    {
      state.update(msg.getCSN());
    }
    numSentUpdates.incrementAndGet();
    return published;
  }

  /**

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

@@ -36,7 +36,7 @@
/**
 * Tests the bookkeeping a replica does on its own changes: they are published in the order of
 * their CSNs, and the announcement that the replica goes offline is only reported as sent when
 * it really was.
 * it really was published.
 * <p>
 * These tests need no server: the changes are built by a CSNGenerator, which reads the time
 * service, and the time service is up as soon as its class is loaded.
@@ -48,9 +48,9 @@
  private static final int SERVER_ID = 42;

  @Test
  public void replicaOfflineMsgIsSentWhenNoChangeIsPending() throws Exception
  public void replicaOfflineMsgTheBrokerPublishedIsReportedAsSent() throws Exception
  {
    final ReplicationDomain domain = mock(ReplicationDomain.class);
    final ReplicationDomain domain = domainWhichPublishes(true);
    final PendingChanges pendingChanges = newPendingChanges(domain);

    final CSN offlineCSN = pendingChanges.putReplicaOfflineMsg();
@@ -62,15 +62,31 @@
  }

  /**
   * The broker writes nothing when it has no usable session, when the changes which come before
   * this one still have to be republished by the recovery, or when it is stopped in between - and
   * what was not written must not be reported as sent: the shutdown of a collocated replication
   * server waits out the whole grace period of a message it was told about and which never
   * reached the wire.
   */
  @Test
  public void replicaOfflineMsgTheBrokerRefusedIsNotReportedAsSent() throws Exception
  {
    final ReplicationDomain domain = domainWhichPublishes(false);
    final PendingChanges pendingChanges = newPendingChanges(domain);

    assertNull(pendingChanges.putReplicaOfflineMsg(), "the broker refused the message");

    assertTrue(onlyMsgPublishedBy(domain) instanceof ReplicaOfflineMsg, "it was attempted");
  }

  /**
   * The message carries the newest CSN of the replica, so a change which is still in flight
   * holds it back - and what was never published must not be reported as sent: the shutdown of
   * a collocated replication server waits out the whole grace period of a message it was told
   * about and which never reaches the wire.
   * holds it back, and the broker is never even asked to publish it.
   */
  @Test
  public void replicaOfflineMsgQueuedBehindAnUncommittedChangeIsNotReportedAsSent() throws Exception
  {
    final ReplicationDomain domain = mock(ReplicationDomain.class);
    final ReplicationDomain domain = domainWhichPublishes(true);
    final PendingChanges pendingChanges = newPendingChanges(domain);
    pendingChanges.putLocalOperation(newLocalOperation());

@@ -88,7 +104,7 @@
  @Test
  public void replicaOfflineMsgWhichCouldNotBeSentIsNotPublishedLater() throws Exception
  {
    final ReplicationDomain domain = mock(ReplicationDomain.class);
    final ReplicationDomain domain = domainWhichPublishes(true);
    final PendingChanges pendingChanges = newPendingChanges(domain);
    final CSN changeCSN = pendingChanges.putLocalOperation(newLocalOperation());
    assertNull(pendingChanges.putReplicaOfflineMsg(), "nothing was published");
@@ -100,11 +116,40 @@
    assertTrue(published instanceof LDAPUpdateMsg, "published " + published);
  }

  /**
   * A change the broker refused leaves the pending changes all the same: the replica has done
   * it, its ServerState says so, and it is by finding that state ahead of the one its
   * replication server reports that the next session republishes the change from the historical
   * information of its entry. Only the offline announcement, which is stored nowhere, needs the
   * answer of the broker.
   */
  @Test
  public void changeTheBrokerRefusedStillLeavesThePendingChanges() throws Exception
  {
    final ReplicationDomain domain = domainWhichPublishes(false);
    final PendingChanges pendingChanges = newPendingChanges(domain);
    final CSN changeCSN = pendingChanges.putLocalOperation(newLocalOperation());
    assertEquals(pendingChanges.size(), 1);

    pendingChanges.commitAndPushCommittedChanges(changeCSN, mock(LDAPUpdateMsg.class));

    assertTrue(onlyMsgPublishedBy(domain) instanceof LDAPUpdateMsg, "the change was published");
    assertEquals(pendingChanges.size(), 0, "and is not queued for a second attempt");
  }

  private PendingChanges newPendingChanges(ReplicationDomain domain)
  {
    return new PendingChanges(new CSNGenerator(SERVER_ID, 0), domain);
  }

  /** A domain whose broker accepts, or refuses, whatever it is given to publish. */
  private ReplicationDomain domainWhichPublishes(boolean accepted)
  {
    final ReplicationDomain domain = mock(ReplicationDomain.class);
    when(domain.publish(any(UpdateMsg.class))).thenReturn(accepted);
    return domain;
  }

  /** A local operation, i.e. one this replica must publish to the other replicas. */
  private PluginOperation newLocalOperation()
  {