From dc20037ee9959ab05d841422015d7e4358162cbb Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Wed, 08 Jul 2026 16:16:40 +0000
Subject: [PATCH] Remove the unfixable testStateMachineFull and fix the dead replay pool (#700)
---
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/StateMachineTest.java | 387 +++----------------------------------------------------
1 files changed, 23 insertions(+), 364 deletions(-)
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/StateMachineTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/StateMachineTest.java
index c899e16..1f2d7e3 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/StateMachineTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/StateMachineTest.java
@@ -13,6 +13,7 @@
*
* Copyright 2009-2010 Sun Microsystems, Inc.
* Portions Copyright 2011-2016 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC
*/
package org.opends.server.replication.plugin;
@@ -198,7 +199,14 @@
{
ConfigurationChangeListener<ReplicationSynchronizationProviderCfg> mmr =
(ConfigurationChangeListener<ReplicationSynchronizationProviderCfg>) provider;
- mmr.applyConfigurationChange(mock(ReplicationSynchronizationProviderCfg.class));
+ // An unstubbed mock would return 0 for getNumUpdateReplayThreads()
+ // (Mockito's default for Integer instead of null), silently killing the
+ // replay thread pool of MultimasterReplication for the rest of the test:
+ // stub it so that the default number of replay threads is used.
+ ReplicationSynchronizationProviderCfg providerCfg = mock(ReplicationSynchronizationProviderCfg.class);
+ when(providerCfg.getNumUpdateReplayThreads()).thenReturn(null);
+ when(providerCfg.getConnectionTimeout()).thenReturn(5000L);
+ mmr.applyConfigurationChange(providerCfg);
}
return replicationDomain;
@@ -388,241 +396,21 @@
}
}
- /**
- * Go through the possible state machine transitions:
- *
- * NC = Not connected status
- * N = Normal status
- * D = Degraded status
- * FU = Full update status
- * BG = Bad generation id status
- *
- * The test path should be:
- * ->NC->D->N->NC->N->D->NC->D->N->BG->NC->N->D->BG->FU->NC->N->D->FU->NC->BG->NC->N->FU->NC->N->NC
- * @throws Exception If a problem occurred
+ /*
+ * Note: the former testStateMachineFull test (a walk through all state
+ * machine transitions) has been removed. It had been disabled forever and
+ * turned out to be unfixable in its historical form:
+ * - it relied on the DS entering the degraded status while replaying a
+ * change flood, which requires the replay to be durably slower than the
+ * publisher; on modern hardware both rates are comparable, making every
+ * degraded transition a race,
+ * - the unbounded floods needed to force the degraded status exhaust the
+ * test JVM heap with real backend entries before the status analyzer
+ * (5 second period) reliably observes a non-empty queue with the
+ * degraded threshold of 1.
+ * The degraded/normal transitions are still covered deterministically by
+ * testStateMachineStatusAnalyzer above.
*/
- @Test(enabled = false, groups = "slow")
- public void testStateMachineFull() throws Exception
- {
- String testCase = "testStateMachineFull";
-
- debugInfo("Starting " + testCase);
-
- initTest();
- BrokerReader br = null;
- BrokerWriter bw = null;
-
- try
- {
- int DEGRADED_STATUS_THRESHOLD = 1;
-
- // RS1 starts with 1 message as degraded status threshold value
- rs1 = createReplicationServer(testCase, DEGRADED_STATUS_THRESHOLD);
-
- // DS2 starts and connects to RS1
- ds2 = createReplicationBroker(DS2_ID, new ServerState(), EMPTY_DN_GENID);
- br = new BrokerReader(ds2, DS2_ID);
- waitUntiConnected(DS2_ID);
-
- // DS2 starts sending a lot of changes
- bw = new BrokerWriter(ds2, DS2_ID, false);
- bw.follow();
- Thread.sleep(1000); // Let some messages being queued in RS
-
- /*
- * DS1 starts and connects to RS1, server state exchange should lead to start in degraded status
- * as some changes should be in queued in the RS and the threshold value is 1 change in queue.
- */
- ds1 = createReplicationDomain(DS1_ID);
- waitUntiConnected(DS1_ID);
- waitUntilStatusEquals(ds1, ServerStatus.DEGRADED_STATUS);
-
- /* DS2 stops sending changes: DS1 should replay pending changes and should enter the normal status */
- bw.pause();
- // Sleep enough so that replay can be done and analyzer has time
- // to see that the queue length is now under the threshold value.
- waitUntilStatusEquals(ds1, ServerStatus.NORMAL_STATUS);
-
- /* RS1 stops to make DS1 go to not connected status (from normal status) */
- rs1.remove();
- waitUntilStatusEquals(ds1, ServerStatus.NOT_CONNECTED_STATUS);
-
- /*
- * DS2 restarts with up to date server state
- * (this allows to have restarting RS1 not sending him some updates he already sent)
- */
- ds2.stop();
- shutdown(bw);
- shutdown(br);
- ServerState curState = ds1.getServerState();
- ds2 = createReplicationBroker(DS2_ID, curState, EMPTY_DN_GENID);
- br = new BrokerReader(ds2, DS2_ID);
-
- /* RS1 restarts, DS1 should get back to normal status */
- rs1 = createReplicationServer(testCase, DEGRADED_STATUS_THRESHOLD);
- waitUntiConnected(DS2_ID);
- waitUntilStatusEquals(ds1, ServerStatus.NORMAL_STATUS);
-
- /* DS2 sends again a lot of changes to make DS1 degraded again */
- bw = new BrokerWriter(ds2, DS2_ID, false);
- bw.follow();
- Thread.sleep(8000); // Let some messages being queued in RS, and analyzer see the change
- waitUntilStatusEquals(ds1, ServerStatus.DEGRADED_STATUS);
-
- /* RS1 stops to make DS1 go to not connected status (from degraded status) */
- rs1.remove();
- bw.pause();
- waitUntilStatusEquals(ds1, ServerStatus.NOT_CONNECTED_STATUS);
-
- /*
- * DS2 restarts with up to date server state
- * (this allows to have restarting RS1 not sending him some updates he already sent)
- */
- ds2.stop();
- shutdown(bw);
- shutdown(br);
- curState = ds1.getServerState();
- ds2 = createReplicationBroker(DS2_ID, curState, EMPTY_DN_GENID);
- br = new BrokerReader(ds2, DS2_ID);
-
- /*
- * RS1 restarts, DS1 should reconnect in degraded status
- * (from not connected this time, not from state machine entry)
- */
- rs1 = createReplicationServer(testCase, DEGRADED_STATUS_THRESHOLD);
- // It is too difficult to tune the right sleep so disabling this test:
- // Sometimes the status analyzer may be fast and quickly change the status
- // of DS1 to NORMAL_STATUS
- //sleep(2000);
- //sleepAssertStatusEquals(30, ds1, ServerStatus.DEGRADED_STATUS);
- waitUntiConnected(DS2_ID);
-
- /* DS1 should come back in normal status after a while */
- waitUntilStatusEquals(ds1, ServerStatus.NORMAL_STATUS);
-
- /*
- * DS2 sends a reset gen id order with wrong gen id:
- * DS1 should go into bad generation id status
- */
- long BAD_GEN_ID = 999999L;
- resetGenId(ds2, BAD_GEN_ID); // ds2 will also go bad gen
- waitUntilStatusEquals(ds1, ServerStatus.BAD_GEN_ID_STATUS);
-
- /*
- * DS2 sends again a reset gen id order with right id: DS1 should be disconnected by RS
- * then reconnect and enter again in normal status. This goes through not connected status
- * but not possible to check as should reconnect immediately
- */
- resetGenId(ds2, EMPTY_DN_GENID); // ds2 will also be disconnected
- ds2.stop();
- shutdown(br); // Reader could reconnect broker, but gen id would be bad: need to recreate a
- // broker to send changex
- waitUntilStatusEquals(ds1, ServerStatus.NORMAL_STATUS);
-
- /* DS2 sends again a lot of changes to make DS1 degraded again */
- curState = ds1.getServerState();
- ds2 = createReplicationBroker(DS2_ID, curState, EMPTY_DN_GENID);
- waitUntiConnected(DS2_ID);
- bw = new BrokerWriter(ds2, DS2_ID, false);
- br = new BrokerReader(ds2, DS2_ID);
- bw.follow();
- Thread.sleep(8000); // Let some messages being queued in RS, and analyzer see the change
- waitUntilStatusEquals(ds1, ServerStatus.DEGRADED_STATUS);
-
- /*
- * DS2 sends reset gen id order with bad gen id: DS1 should go in bad gen id status
- * (from degraded status this time)
- */
- // -1 to allow next step full update and flush RS db so that DS1 can reconnect after full
- // update
- resetGenId(ds2, -1);
- waitUntilStatusEquals(ds1, ServerStatus.BAD_GEN_ID_STATUS);
- bw.pause();
-
- /*
- * DS2 engages full update (while DS1 in bad gen id status),
- * DS1 should go in full update status
- */
- BrokerInitializer bi = new BrokerInitializer(ds2, DS2_ID, false);
- bi.initFullUpdate(DS1_ID, 200);
- waitUntilStatusEquals(ds1, ServerStatus.FULL_UPDATE_STATUS);
-
- /*
- * DS2 terminates full update to DS1: DS1 should reconnect (goes through not connected status)
- * and come back to normal status (RS genid was -1 so RS will adopt new gen id)
- */
- bi.runFullUpdate();
- waitUntilStatusEquals(ds1, ServerStatus.NORMAL_STATUS);
-
- /* DS2 sends changes to DS1: DS1 should go in degraded status */
- ds2.stop(); // will need a new broker with another gen id restart it
- shutdown(bw);
- shutdown(br);
- long newGen = ds1.getGenerationID();
- curState = ds1.getServerState();
- ds2 = createReplicationBroker(DS2_ID, curState, newGen);
- waitUntiConnected(DS2_ID);
- bw = new BrokerWriter(ds2, DS2_ID, false);
- br = new BrokerReader(ds2, DS2_ID);
- bw.follow();
- Thread.sleep(8000); // Let some messages being queued in RS, and analyzer see the change
- waitUntilStatusEquals(ds1, ServerStatus.DEGRADED_STATUS);
-
- /* DS2 engages full update (while DS1 in degraded status), DS1 should go in full update status */
- bi = new BrokerInitializer(ds2, DS2_ID, false);
- bi.initFullUpdate(DS1_ID, 300);
- waitUntilStatusEquals(ds1, ServerStatus.FULL_UPDATE_STATUS);
- bw.pause();
-
- /*
- * DS2 terminates full update to DS1: DS1 should reconnect (goes through not connected status)
- * and come back to bad gen id status (RS genid was another gen id (300 entries instead of 200)
- */
- bi.runFullUpdate();
- waitUntilStatusEquals(ds1, ServerStatus.BAD_GEN_ID_STATUS);
-
- /*
- * DS2 sends reset gen id with gen id same as DS1:
- * DS1 will be disconnected by RS (not connected status) and come back to normal status
- */
- ds2.stop(); // will need a new broker with another gen id restart it
- shutdown(bw);
- shutdown(br);
- newGen = ds1.getGenerationID();
- curState = ds1.getServerState();
- ds2 = createReplicationBroker(DS2_ID, curState, newGen);
- waitUntiConnected(DS2_ID);
- br = new BrokerReader(ds2, DS2_ID);
- resetGenId(ds2, newGen); // Make DS1 reconnect in normal status
-
- waitUntilStatusEquals(ds1, ServerStatus.NORMAL_STATUS);
-
- /* DS2 engages full update (while DS1 in normal status), DS1 should go in full update status */
- bi = new BrokerInitializer(ds2, DS2_ID, false);
- bi.initFullUpdate(DS1_ID, 300); // 300 entries will compute same genid of the RS
- waitUntilStatusEquals(ds1, ServerStatus.FULL_UPDATE_STATUS);
-
- /*
- * DS2 terminates full update to DS1: DS1 should reconnect (goes through not connected status)
- * and come back to normal status (process full update with same data as before so RS already
- * has right gen id: version with 300 entries)
- */
- bi.runFullUpdate();
- ds2.stop();
- shutdown(br);
- waitUntilStatusEquals(ds1, ServerStatus.NORMAL_STATUS);
-
- /* RS1 stops, DS1 should go to not connected status */
- rs1.remove();
- waitUntilStatusEquals(ds1, ServerStatus.NOT_CONNECTED_STATUS);
- } finally
- {
- // Finalize test
- endTest();
- shutdown(bw);
- shutdown(br);
- }
- }
/**
* Set up the environment.
@@ -661,136 +449,7 @@
paranoiaCheck();
}
- /**
- * Sends a reset genid message through the given replication broker, with the
- * given new generation id.
- */
- private void resetGenId(ReplicationBroker rb, long newGenId)
- {
- ResetGenerationIdMsg resetMsg = new ResetGenerationIdMsg(newGenId);
- rb.publish(resetMsg);
- }
- /**
- * Utility class for making a full update through a broker. No separated thread
- * Usage:
- * BrokerInitializer bi = new BrokerInitializer(rb, sid, nEntries);
- * bi.initFullUpdate(); // Initializes a full update session by sending InitializeTargetMsg
- * bi.runFullUpdate(); // loops sending nEntries entries and finalizes the full update by sending the EntryDoneMsg
- */
- private class BrokerInitializer
- {
- private ReplicationBroker rb;
- private int serverId = -1;
- private long userId;
- /** Server id of server to initialize. */
- private int destId = -1;
- /** Number of entries to send to dest. */
- private long nEntries = -1;
- private boolean createReader;
-
- /**
- * If the BrokerInitializer is to be used for a lot of entries to send
- * (which is often the case), the reader thread should be enabled to make
- * the window subsystem work and allow the broker to send as much entries as
- * he wants. If not enabled, the user is responsible to call the receive
- * method of the broker himself.
- */
- private BrokerReader reader;
-
- /** Creates a broker initializer. Also creates a reader according to request */
- public BrokerInitializer(ReplicationBroker rb, int serverId,
- boolean createReader)
- {
- this.rb = rb;
- this.serverId = serverId;
- this.createReader = createReader;
- }
-
- /** Initializes a full update session by sending InitializeTargetMsg. */
- public void initFullUpdate(int destId, long nEntries)
- {
- // Also create reader ?
- if (createReader)
- {
- reader = new BrokerReader(rb, serverId);
- }
-
- debugInfo("Broker " + serverId + " initializer sending InitializeTargetMsg to server " + destId);
-
- this.destId = destId;
- this.nEntries = nEntries;
-
- // Send init msg to warn dest server it is going do be initialized
- RoutableMsg initTargetMsg = new InitializeTargetMsg(
- EXAMPLE_DN_, serverId, destId, serverId, nEntries, initWindow);
- rb.publish(initTargetMsg);
-
- // Send top entry for the domain
- String topEntry = "dn: " + EXAMPLE_DN + "\n"
- + "objectClass: top\n"
- + "objectClass: domain\n"
- + "dc: example\n"
- + "entryUUID: 11111111-1111-1111-1111-111111111111\n\n";
- EntryMsg entryMsg = new EntryMsg(serverId, destId, topEntry.getBytes(), 1);
- rb.publish(entryMsg);
- }
-
- private EntryMsg createNextEntryMsg()
- {
- String userEntryUUID = "11111111-1111-1111-1111-111111111111";
- long curId = ++userId;
- String userdn = "uid=full_update_user" + curId + "," + EXAMPLE_DN;
- String entryWithUUIDldif = "dn: " + userdn + "\n" + "objectClass: top\n" +
- "objectClass: person\n" + "objectClass: organizationalPerson\n" +
- "objectClass: inetOrgPerson\n" +
- "uid: full_update_user" + curId + "\n" +
- "homePhone: 951-245-7634\n" +
- "description: This is the description for Aaccf Amar.\n" + "st: NC\n" +
- "mobile: 027-085-0537\n" +
- "postalAddress: Aaccf Amar$17984 Thirteenth Street" +
- "$Rockford, NC 85762\n" + "mail: user.1@example.com\n" +
- "cn: Aaccf Amar\n" + "l: Rockford\n" + "pager: 508-763-4246\n" +
- "street: 17984 Thirteenth Street\n" + "telephoneNumber: 216-564-6748\n" +
- "employeeNumber: 1\n" + "sn: Amar\n" + "givenName: Aaccf\n" +
- "postalCode: 85762\n" + "userPassword: password\n" + "initials: AA\n" +
- "entryUUID: " + userEntryUUID + "\n\n";
- // -> WARNING: EntryMsg PDUs are concatenated before calling import on LDIF
- // file so need \n\n to separate LDIF entries to conform to LDIF file format
-
- // Create an entry message
- return new EntryMsg(serverId, destId, entryWithUUIDldif.getBytes(),
- (int) userId);
- }
-
- /**
- * Loops sending entries for full update (EntryMsg messages). When
- * terminates, sends the EntryDoneMsg to finalize full update. Number of
- * sent entries is determined at initFullUpdate call time.
- */
- public void runFullUpdate()
- {
- debugInfo("Broker " + serverId + " initializer starting sending entries to server " + destId);
-
- for(long i = 0 ; i<nEntries ; i++) {
- EntryMsg entryMsg = createNextEntryMsg();
- rb.publish(entryMsg);
- }
-
- debugInfo("Broker " + serverId + " initializer stopping sending entries");
-
- debugInfo("Broker " + serverId + " initializer sending EntryDoneMsg");
- DoneMsg doneMsg = new DoneMsg(serverId, destId);
- rb.publish(doneMsg);
-
- if (createReader)
- {
- shutdown(reader);
- }
-
- debugInfo("Broker " + serverId + " initializer thread is dying");
- }
- }
/** Thread for sending a lot of changes through a broker. */
private class BrokerWriter extends Thread
--
Gitblit v1.10.0