From 76088205f8f8e7f9083bf238aa5648fa8b6a5905 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Tue, 22 Sep 2026 10:57:31 +0000
Subject: [PATCH] [#1039] Report a total update whose session stops before the DoneMsg as a failed import (#1044)
---
opendj-server-legacy/src/test/java/org/opends/server/replication/InitOnLineTest.java | 86 +++++++++++++++++++++++++++++++++++++++++++
opendj-server-legacy/src/messages/org/opends/messages/replication.properties | 4 ++
opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java | 28 +++++++------
3 files changed, 105 insertions(+), 13 deletions(-)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java
index 3dd9cfd..472f68c 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java
+++ b/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
diff --git a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
index 3129b6e..594524f 100644
--- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
+++ b/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
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/InitOnLineTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/InitOnLineTest.java
index 4703794..7f5431c 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/InitOnLineTest.java
+++ b/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
--
Gitblit v1.10.0