[#1039] Report a total update whose session stops before the DoneMsg as a f...

Valery Kharseko

20 hours ago 76088205f8f8e7f9083bf238aa5648fa8b6a5905

[#1039] Report a total update whose session stops before the DoneMsg as a failed import (#1044)

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

@@ -2127,19 +2127,21 @@

        if (msg == null)
        {
          if (broker.shuttingDown())
          {
            // The server is in the shutdown process
            return null;
          }
          else
          {
            // Handle connection issues
            ieCtx.setExceptionIfNoneSet(new DirectoryException(
                ResultCode.OTHER, ERR_INIT_RS_DISCONNECTION_DURING_IMPORT
                    .get(broker.getReplicationServer())));
            return null;
          }
          /*
           * The stream ended before the DoneMsg of the exporter: the broker lost its
           * connection, or it was stopped under the import - by the shutdown of the server,
           * or by a restart of the session which every road takes through disableService().
           * Either way the import is a failure and is recorded as one (issue #1039): the
           * import which ends on the entries which had arrived would otherwise be reported
           * as finished, with the generationId of the exporter loaded from the base entry
           * among them, and the replica would come up as a peer of the exporter over part of
           * its data. A failed import has its generationId computed over the data instead.
           */
          final LocalizableMessage cause = broker.shuttingDown()
              ? ERR_INIT_SESSION_STOPPED_DURING_IMPORT.get(getBaseDN(), getServerId(), ieCtx.importSource)
              : ERR_INIT_RS_DISCONNECTION_DURING_IMPORT.get(broker.getReplicationServer());
          ieCtx.setExceptionIfNoneSet(new DirectoryException(ResultCode.OTHER, cause));
          return null;
        }

        // Check good ordering of msg received

 opendj-server-legacy/src/messages/org/opends/messages/replication.properties

@@ -705,3 +705,7 @@
 comes back with the last state it did write and replays the changes since
ERR_STATE_CHECKPOINTER_NOT_STOPPED_324=The state checkpointer of domain "%s" has not stopped within \
 %d ms : the shutdown of the domain goes on without it
ERR_INIT_SESSION_STOPPED_DURING_IMPORT_329=Domain %s (server id: %s) : the session to the \
 replication server was stopped before the initialization from server %s completed. The \
 entries which had arrived are imported, and the generation id of the data is computed over \
 them rather than taken from the exporter

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

@@ -587,6 +587,92 @@
  }

  /**
   * A total update into this replica whose session stops before the DoneMsg arrives is a
   * failed import, and the generationId of the exporter does not stay over the part of its
   * data which arrived (issue #1039). Every road which stops the session - the restart a
   * failed replay or a configuration change asks for, the shutdown of the server - goes
   * through disableService(), and the import read the stopped broker as the end of the
   * stream: it kept the entries which had arrived, loaded the exporter's generationId from
   * the base entry among them, and completed its task as a success.
   */
  @Test(enabled=true)
  public void initializeImportSessionStoppedBeforeDone() throws Exception
  {
    String testCase = "initializeImportSessionStoppedBeforeDone";
    log("Starting " + testCase);
    try
    {
      replServer1 = createReplicationServer(replServer1ID, testCase);
      connectServer1ToReplServer(replServer1ID);
      server2 = openReplicationSession(baseDN,
          server2ID, 100, getReplServerPort(replServer1ID), 10000);

      // In S1 launch the total update, and S2 receives the request
      addTask(taskInitFromS2, ResultCode.SUCCESS, null);
      ReplicationMsg msg = server2.receive();
      Assertions.assertThat(msg).isInstanceOf(InitializeRequestMsg.class);

      // S2 announces every entry and sends two of them: the base entry, carrying the
      // generationId of S2 the way the base entry of a real export does, and one more.
      // The DoneMsg never comes.
      final long exporterGenerationId = 7777777L;
      final String baseEntry = updatedEntries[0].substring(0, updatedEntries[0].length() - 1)
          + "ds-sync-generation-id: " + exporterGenerationId + "\n\n";
      server2.publish(new InitializeTargetMsg(baseDN, server2ID, server1ID, server1ID,
          updatedEntries.length, initWindow));
      server2.publish(new EntryMsg(server2ID, server1ID, baseEntry.getBytes(), 1));
      server2.publish(new EntryMsg(server2ID, server1ID, updatedEntries[1].getBytes(), 2));

      // The import has read both and is waiting for the next one
      waitTaskLeft(taskInitFromS2, updatedEntries.length - 2);

      // The session stops under the import, the way every road to a stopped session does
      replDomain.disableService();

      waitTaskCompleted(taskInitFromS2, STOPPED_BY_ERROR, updatedEntries.length - 2, 2);
      // ...for the reason it failed, not for the lost connection every other early end reports
      final String reason =
          ERR_INIT_SESSION_STOPPED_DURING_IMPORT.get(baseDN, server1ID, server2ID).toString();
      Assertions.assertThat(getEntry(taskInitFromS2.getName(), 1000, true)
              .parseAttribute(ATTR_TASK_LOG_MESSAGES).asSetOfString())
          .as("the task does not report the stopped session as the reason its import failed")
          .anyMatch(record -> record.contains(reason));
      assertNotEquals(replDomain.getGenerationID(), exporterGenerationId,
          "the generationId of the exporter stayed over the part of its data which arrived");
      Entry base = getEntry(baseDN, 1000, true);
      assertNotEquals(base.parseAttribute("ds-sync-generation-id").asString(),
          String.valueOf(exporterGenerationId),
          "the generationId of the exporter stayed stored on the base entry of a cut import");

      log("Successfully ending " + testCase);
    }
    finally
    {
      afterTest(testCase);
    }
  }

  /** Waits until the task reports the given number of entries still to be imported. */
  private void waitTaskLeft(Entry taskEntry, long expectedLeft) throws Exception
  {
    final long deadline = System.currentTimeMillis() + 20000;
    String left;
    do
    {
      final SearchRequest request = newSearchRequest(taskEntry.getName(), SearchScope.BASE_OBJECT);
      Entry resultEntry = connection.processSearch(request).getSearchEntries().getFirst();
      left = resultEntry.parseAttribute(ATTR_TASK_INITIALIZE_LEFT).asString();
      if (String.valueOf(expectedLeft).equals(left))
      {
        return;
      }
      Thread.sleep(100);
    }
    while (System.currentTimeMillis() < deadline);
    fail("the import did not reach " + expectedLeft + " entries left within 20s, last read " + left);
  }

  /**
   * Tests the export side of the Initialize task
   * Test steps :
   * - add entries in S1, make S2 publish InitRequest