mirror of https://github.com/OpenIdentityPlatform/OpenDJ.git
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ExternalChangelogDomain.java
@@ -13,9 +13,13 @@ * * Copyright 2006-2010 Sun Microsystems, Inc. * Portions Copyright 2011-2016 ForgeRock AS. * Portions Copyright 2026 3A Systems, LLC. */ package org.opends.server.replication.plugin; import static org.opends.messages.ReplicationMessages.*; import static org.opends.server.util.StaticUtils.*; import java.util.List; import org.forgerock.i18n.LocalizableMessage; @@ -36,7 +40,8 @@ { private LDAPReplicationDomain domain; private boolean isEnabled; /** Published by a configuration change, read by the changelog threads without a lock. */ private volatile boolean isEnabled; /** * Constructor from a provided LDAPReplicationDomain. @@ -91,9 +96,24 @@ return ccr; } /* * Stored before the domain restarts its session for the attributes below, so that * what the restarted session publishes to the external changelog is decided by this * configuration rather than by the one it replaces. The failure of that restart is * reported rather than thrown at the configuration framework, which would leave the * listeners after this one uncalled - it does leave the attributes it had already * applied in place, as any listener of an entry written before it runs does. */ this.isEnabled = configuration.isEnabled(); domain.changeConfig(configuration.getECLInclude(), configuration.getECLIncludeForDeletes()); try { domain.changeConfig(configuration.getECLInclude(), configuration.getECLIncludeForDeletes()); } catch (Exception e) { return refused(configuration, stackTraceToSingleLineString(e)); } return new ConfigChangeResult(); } @@ -111,12 +131,19 @@ } catch (Exception e) { final ConfigChangeResult ccr = new ConfigChangeResult(); ccr.setResultCode(ResultCode.CONSTRAINT_VIOLATION); return ccr; return refused(configuration, stackTraceToSingleLineString(e)); } } private ConfigChangeResult refused(ExternalChangelogDomainCfg configuration, String reason) { final ConfigChangeResult ccr = new ConfigChangeResult(); ccr.setResultCode(ResultCode.CONSTRAINT_VIOLATION); ccr.addMessage(NOTE_ERR_UNABLE_TO_ENABLE_ECL.get( "External Changelog Domain " + configuration.dn(), reason)); return ccr; } /** {@inheritDoc} */ @Override opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java
@@ -391,6 +391,17 @@ @GuardedBy("serviceStateLock") private long sessionGeneration; /** * Set by {@link #restartService()} when it left the session of this domain alone, so * that the configuration change which asked for the restart can say so. * <p> * Cleared by that change before it applies anything, and read by it once it is done - * both under {@link #serviceStateLock}, the lock every restart of the session runs * under, so what it reads is what its own steps asked for. A restart which is * suppressed outside a configuration change leaves it set for the next one to clear. */ @GuardedBy("serviceStateLock") private boolean sessionRestartSuppressed; /** * Held while a replay thread applies a change of this domain, and taken exclusively by * this domain on its way down. * <p> @@ -501,7 +512,8 @@ * too early. */ private final RemotePendingChanges remotePendingChanges; private boolean solveConflictFlag = true; /** Published by a configuration change, read by the replay threads without the lock. */ private volatile boolean solveConflictFlag = true; private final InternalClientConnection conn = getRootConnection(); private final AtomicBoolean shutdown = new AtomicBoolean(); @@ -513,7 +525,8 @@ */ private final SortedMap<CSN, FakeOperation> replayOperations = new TreeMap<>(); private ExternalChangelogDomain eclDomain; /** Published by a configuration change, read by the changelog threads without a lock. */ private volatile ExternalChangelogDomain eclDomain; /** A boolean indicating if the thread used to save the persistentServerState is terminated. */ private volatile boolean done = true; @@ -744,15 +757,21 @@ // Get fractional configuration fractionalConfig = new FractionalConfig(getBaseDN()); readFractionalConfig(configuration, false); storeECLConfiguration(configuration); solveConflictFlag = isSolveConflict(configuration); // Checked before the ECL configuration, which reads the backend to create its default // entry: a domain on a backend of its own reports that rather than what it made of it. LocalBackend<?> backend = getBackend(); if (backend == null) { throw new ConfigException(ERR_SEARCHING_DOMAIN_BACKEND.get(getBaseDN())); } // The ECL domain is created here rather than handed a change it could refuse, so its // result can only be a success. createECLConfigurationEntryIfMissing(configuration); applyECLConfiguration(requireECLConfiguration(configuration)); solveConflictFlag = isSolveConflict(configuration); try { generationId = loadGenerationId(); @@ -887,6 +906,10 @@ disableService(); sessionGeneration++; } else if (needReconnection) { onSessionRestartSuppressed(); } // Set new configuration int newFractionalMode = newFractionalConfig.fractionalConfigToInt(); fractionalConfig.setFractional(newFractionalMode != @@ -4813,7 +4836,34 @@ public ConfigChangeResult applyConfigurationChange( ReplicationDomainCfg configuration) { this.config = configuration; final ConfigChangeResult ccr = new ConfigChangeResult(); /* * The step which can fail comes first, so that none of the domain configuration is * published before it succeeded. It is not free of writes of its own: a domain * without an external changelog configuration is given the default entry here, and * that entry stays written whether or not the rest of this succeeds. Refusing a * change before anything at all is written is what isConfigurationChangeAcceptable() * is for, and it is where a change carrying an unreadable configuration is refused * for good - the modified entry is written to the server configuration between that * method and this one, and it is not rolled back when this one reports an error. What * this reads is read again under the lock, so that a change of that entry which lands * in between is applied rather than reverted by this snapshot of it. */ try { createECLConfigurationEntryIfMissing(configuration); // Read once here so that a configuration which cannot be read fails before any of // this change is published, and once more under the lock so that what is applied is // not an outdated snapshot of it. requireECLConfiguration(configuration); } catch (Exception e) { ccr.setResultCode(ResultCode.OTHER); ccr.addMessage(configChangeFailed(configuration, e)); return ccr; } /* * Each of these stops and starts the session when what it changes calls for it, and * the configuration they change is read as the session comes up: hold the lock the @@ -4822,27 +4872,151 @@ */ synchronized (serviceStateLock) { changeConfig(configuration); // Whatever a restart was suppressed for before this change is none of its business. sessionRestartSuppressed = false; /* * Reported rather than thrown, all of it: this listener runs on an entry which is * already written, and an exception leaving it would abort the listeners after it * as well. What is applied when one of these fails stays applied - which is what * the framework says of a listener of an entry written before it runs. */ try { this.config = configuration; changeConfig(configuration); // Read assured + fractional configuration and each time reconnect if needed readAssuredConfig(configuration, true); readFractionalConfig(configuration, true); // Read assured + fractional configuration and each time reconnect if needed. A // domain which owns its session gets none of those reconnections. final boolean allowReconnection = !ownsItsSession(); readAssuredConfig(configuration, allowReconnection); readFractionalConfig(configuration, allowReconnection); solveConflictFlag = isSolveConflict(configuration); /* * Applied last, and still under the lock: this one restarts the session as well * when the attributes published to the external changelog changed, and the * session it starts replays on everything set above. It is read again here, so * that a change of that entry which landed in between is applied rather than * reverted by an older snapshot of it. */ ccr.aggregate(applyECLConfiguration(requireECLConfiguration(configuration))); } catch (Exception e) { ccr.setResultCode(ResultCode.OTHER); ccr.addMessage(configChangeFailed(configuration, e)); } /* * Read here rather than reported by restartService() itself, which has no result to * report through and is called from the external changelog configuration above as * well: what a change needed a session for, and did not get, is one thing to the * administrator whichever step of it asked. Left out of the failure path on purpose * - a change which could not be applied has a reason of its own to carry, and the * session it did not restart is not what the administrator has to act on. */ if (sessionRestartSuppressed && ccr.getResultCode() == ResultCode.SUCCESS) { ccr.setAdminActionRequired(true); ccr.addMessage(NOTE_REPLICATION_DOMAIN_SESSION_NOT_RESTARTED.get(getBaseDN())); } } solveConflictFlag = isSolveConflict(configuration); final ConfigChangeResult ccr = new ConfigChangeResult(); try { storeECLConfiguration(configuration); } catch(Exception e) { ccr.setResultCode(ResultCode.OTHER); } return ccr; } /** * Whether this domain, rather than a configuration change, decides when its session * runs: it is shutting down, or it is disabled for the length of a total update. */ private boolean ownsItsSession() { return shutdown.get() || disabled; } @Override protected void restartService() { synchronized (serviceStateLock) { if (ownsItsSession()) { /* * The domain is going away or is being imported into: a restart here would bring * a session, and the listener thread which goes with it, back up on a domain * whose ServerState is gone from memory. The session started when the domain is * enabled again reads the configuration this restart was asked for. * * Recorded rather than passed over in silence: the configuration a restart was * asked for is stored, and it is the session which is not brought up on it, so a * change which reports plain success would have the administrator believe the * domain is running on it already. A domain disabled for a total update comes up * on it when the total update ends; one which stays disabled - enable() gives up * when the data state it reads cannot be loaded, and nothing calls it again - * never does, and that is what the administrator is told to act on. */ onSessionRestartSuppressed(); return; } super.restartService(); } } /** * {@inheritDoc} * <p> * Every restart this domain leaves alone comes through here, whichever step of the * change asked for it: the broker properties which are renegotiated, the assured * configuration the replication server is told about as the session comes up, the * fractional configuration the session filters on, and the attributes the external * changelog publishes. */ @Override protected void onSessionRestartSuppressed() { synchronized (serviceStateLock) { sessionRestartSuppressed = true; } } /** * What the administrator is told a configuration change failed with. * <p> * The reason a {@link ConfigException} carries is passed on as it is: it names the step * which raised it, the external changelog configuration this listener reads included. * Anything else comes out of applying the domain configuration - the broker, the * assured and the fractional configuration - and is reported as such rather than as a * failure of the external changelog, which most of what this listener does has nothing * to do with. */ private LocalizableMessage configChangeFailed(ReplicationDomainCfg domCfg, Exception e) { if (e instanceof ConfigException) { return ((ConfigException) e).getMessageObject(); } return ERR_REPLICATION_DOMAIN_CONFIG_CHANGE_FAILED.get( domCfg.getBaseDN(), stackTraceToSingleLineString(e)); } /** * {@inheritDoc} * <p> * Taken under {@link #serviceStateLock} like every other configuration change: this one * comes from the external changelog domain - from the entry of its own, or from * {@link #applyECLConfiguration} - and it restarts the session as well. */ @Override public void changeConfig(Set<String> includeAttributes, Set<String> includeAttributesForDeletes) { synchronized (serviceStateLock) { super.changeConfig(includeAttributes, includeAttributesForDeletes); } } @Override public boolean isConfigurationChangeAcceptable( ReplicationDomainCfg configuration, List<LocalizableMessage> unacceptableReasons) @@ -4855,6 +5029,28 @@ return false; } /* * Check the external changelog configuration can be read. This is the one thing * applying the change can fail on, and this is where refusing it means something: the * modified entry is written to the server configuration between this method and * applyConfigurationChange(), and it is not rolled back when the latter reports an * error. Refused here, the entry is never written at all. * <p> * A domain whose external changelog configuration cannot be read therefore refuses * every change of its own entry until that configuration is repaired. What repairs * it is a change of the "cn=external changelog" entry, or its removal, neither of * which comes through here. */ try { readECLConfiguration(configuration); } catch (ConfigException e) { unacceptableReasons.add(e.getMessageObject()); return false; } // Check fractional configuration try { @@ -4906,7 +5102,7 @@ { try { DN eclConfigEntryDN = DN.valueOf("cn=external changeLog," + config.dn()); DN eclConfigEntryDN = eclConfigurationEntryDN(config); if (getServerContext().getConfigurationHandler().hasEntry(eclConfigEntryDN)) { getServerContext().getConfigurationHandler().deleteEntry(eclConfigEntryDN); @@ -4920,63 +5116,160 @@ } /** * Store the provided ECL configuration for the domain. * Reads the ECL configuration of the domain, changing nothing. * <p> * This is what {@link #isConfigurationChangeAcceptable} checks, so it leaves the server * configuration as it found it - the entry the domain is missing is created by * {@link #createECLConfigurationEntryIfMissing} once the change is accepted. It is read * off the provided configuration rather than off {@link #config}, which the caller of * the latter has not published yet. * * @param domCfg The provided configuration. * @return The ECL configuration, or {@code null} when the domain has none yet. * @throws ConfigException When it exists but could not be read. */ private ExternalChangelogDomainCfg readECLConfiguration(ReplicationDomainCfg domCfg) throws ConfigException { try { return domCfg.getExternalChangelogDomain(); } catch (Exception e) { if (!Boolean.TRUE.equals(hasECLConfigurationEntry(domCfg))) { /* * There is none to read - a default one is created when the change is applied - * or whether there is one could not be told, and a domain is not held back from * every change of its own by a failure which never reached its entry. */ return null; } throw new ConfigException(NOTE_ERR_UNABLE_TO_ENABLE_ECL.get( "Replication Domain on " + domCfg.getBaseDN(), stackTraceToSingleLineString(e)), e); } } /** Reads the ECL configuration which must be there, as the caller has just created it. */ private ExternalChangelogDomainCfg requireECLConfiguration(ReplicationDomainCfg domCfg) throws ConfigException { final ExternalChangelogDomainCfg eclDomCfg = readECLConfiguration(domCfg); if (eclDomCfg == null) { throw new ConfigException(NOTE_ERR_UNABLE_TO_ENABLE_ECL.get( "Replication Domain on " + domCfg.getBaseDN(), "its external changelog configuration is gone")); } return eclDomCfg; } private static DN eclConfigurationEntryDN(ReplicationDomainCfg domCfg) { return DN.valueOf("cn=external changelog," + domCfg.dn()); } /** Whether the ECL configuration entry is there, or {@code null} when it cannot be told. */ private Boolean hasECLConfigurationEntry(ReplicationDomainCfg domCfg) { try { final ConfigurationHandler configHandler = getServerContext().getConfigurationHandler(); // There may not be any config entry related to this domain in some unit test cases return configHandler.hasEntry(domCfg.dn()) && configHandler.hasEntry(eclConfigurationEntryDN(domCfg)); } catch (Exception e) { logger.traceException(e); return null; } } /** * Creates the entry the ECL configuration of the domain is stored in, with its default * values, when the server configuration does not carry one yet. * * @param domCfg The provided configuration. * @throws ConfigException When an error occurred. */ private void storeECLConfiguration(ReplicationDomainCfg domCfg) private void createECLConfigurationEntryIfMissing(ReplicationDomainCfg domCfg) throws ConfigException { ExternalChangelogDomainCfg eclDomCfg = null; if (readECLConfiguration(domCfg) != null) { return; } // create the ecl config if it does not exist // There may not be any config entry related to this domain in some // unit test cases try { DN configDn = config.dn(); DN configDn = domCfg.dn(); ConfigurationHandler configHandler = getServerContext().getConfigurationHandler(); if (configHandler.hasEntry(config.dn())) // domain with no config entry only when running unit tests if (configHandler.hasEntry(configDn)) { try { eclDomCfg = domCfg.getExternalChangelogDomain(); } catch(Exception e) { /* do nothing */ } // domain with no config entry only when running unit tests if (eclDomCfg == null) if (!configHandler.hasEntry(eclConfigurationEntryDN(domCfg))) { // no ECL config provided hence create a default one // create the default one DN eclConfigEntryDN = DN.valueOf("cn=external changelog," + configDn); if (!configHandler.hasEntry(eclConfigEntryDN)) if (getBackend() == null) { // no entry exist yet for the ECL config for this domain // create it String ldif = makeLdif( "dn: cn=external changelog," + configDn, "objectClass: top", "objectClass: ds-cfg-external-changelog-domain", "cn: external changelog", "ds-cfg-enabled: " + !getBackend().isPrivateBackend()); LDIFImportConfig ldifImportConfig = new LDIFImportConfig( new StringReader(ldif)); // No need to validate schema in replication ldifImportConfig.setValidateSchema(false); LDIFReader reader = new LDIFReader(ldifImportConfig); // Read to tell a private backend from a public one just below. throw new ConfigException(ERR_SEARCHING_DOMAIN_BACKEND.get(domCfg.getBaseDN())); } // no entry exist yet for the ECL config for this domain // create it String ldif = makeLdif( "dn: cn=external changelog," + configDn, "objectClass: top", "objectClass: ds-cfg-external-changelog-domain", "cn: external changelog", "ds-cfg-enabled: " + !getBackend().isPrivateBackend()); LDIFImportConfig ldifImportConfig = new LDIFImportConfig( new StringReader(ldif)); // No need to validate schema in replication ldifImportConfig.setValidateSchema(false); try (LDIFReader reader = new LDIFReader(ldifImportConfig)) { Entry eclEntry = reader.readEntry(); configHandler.addEntry(Converters.from(eclEntry)); ldifImportConfig.close(); } } } eclDomCfg = domCfg.getExternalChangelogDomain(); } catch (ConfigException e) { throw e; } catch (Exception e) { throw new ConfigException(NOTE_ERR_UNABLE_TO_ENABLE_ECL.get( "Replication Domain on " + domCfg.getBaseDN(), stackTraceToSingleLineString(e)), e); } } /** * Applies the provided ECL configuration to this domain. * <p> * This restarts the session when the attributes published to the external changelog * changed, so the configuration it is applied along must be in place already. * * @param eclDomCfg The ECL configuration read by {@link #requireECLConfiguration}. * @return What the ECL domain made of the change: it reports a change it cannot apply * rather than throwing it. * @throws ConfigException When applying it failed, the session it restarts included. */ private ConfigChangeResult applyECLConfiguration(ExternalChangelogDomainCfg eclDomCfg) throws ConfigException { try { if (eclDomain != null) { eclDomain.applyConfigurationChange(eclDomCfg); return eclDomain.applyConfigurationChange(eclDomCfg); } else { // Create the ECL domain object eclDomain = new ExternalChangelogDomain(this, eclDomCfg); } // Create the ECL domain object eclDomain = new ExternalChangelogDomain(this, eclDomCfg); return new ConfigChangeResult(); } catch (Exception e) { opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java
@@ -3393,13 +3393,36 @@ } } private void restartService() /** * Stops the session of this domain and starts it again, so that it comes up on the * configuration which has just changed. * <p> * A subclass may leave it alone: a domain which is shutting down, or which was disabled * for a total update, owns its session and is not given one back by a configuration * change. One which does reports it through {@link #onSessionRestartSuppressed()}. */ protected void restartService() { disableService(); enableService(); } /** * Called when what a change carries is negotiated as the session comes up, and the * session was not restarted for it. * <p> * The configuration is stored either way, and the session started next reads it - so * this says that the change is not live yet rather than that it was lost. A domain * which restarts its session for every change never reaches this; one which owns its * session while it is shutting down or disabled for a total update overrides it to tell * the administrator what is waiting for that session. */ protected void onSessionRestartSuppressed() { // Nothing to report: this domain restarts its session for whatever asks for it. } /** * This method should trigger an export of the replicated data. * to the provided outputStream. * When finished the outputStream should be flushed and closed. @@ -3837,9 +3860,11 @@ } /** * Gets and stores the assured replication configuration parameters. Returns a * boolean indicating if the passed configuration has changed compared to * previous values and the changes require a reconnection. * Gets and stores the assured replication configuration parameters. * <p> * The configuration is stored whether or not the session has to be restarted for it: * the assured timeout is read off it as the acknowledgements are waited for, and needs * no reconnection at all. * * @param config * The configuration object @@ -3852,12 +3877,28 @@ // Disconnect if required: changing configuration values before // disconnection would make assured replication used immediately and // disconnection could cause some timeouts error. if (needReconnection(config) && allowReconnection) final boolean needReconnection = needReconnection(config); final boolean needRestart = needReconnection && allowReconnection; if (needRestart) { disableService(); assuredConfig = config; } else if (needReconnection) { onSessionRestartSuppressed(); } /* * Stored whether or not the session was restarted for it, as the fractional * configuration is: the assured timeout is the one property a session does not have to * be restarted for, so a change carrying it alone - reported as applied and then * dropped, before - is applied here. A caller which does not allow the reconnection * has no session running assured replication either: the domain is being built, is * shutting down, or is disabled for the length of a total update, and the session its * enable() starts reads what is stored here. */ assuredConfig = config; if (needRestart) { enableService(); } } opendj-server-legacy/src/messages/org/opends/messages/replication.properties
@@ -665,3 +665,10 @@ for the replay of one of its changes to finish. A change which reaches the backend from now on \ is not recorded in the ServerState being saved, so the replication server sends it again and it \ is replayed a second time ERR_REPLICATION_DOMAIN_CONFIG_CHANGE_FAILED_326=Could not apply a configuration change to the \ replication domain on "%s": %s NOTE_REPLICATION_DOMAIN_SESSION_NOT_RESTARTED_327=The configuration change was applied to the \ replication domain on "%s", but the session to the replication server was not restarted for it: \ the domain is shutting down, or it is disabled for the length of a total update. The change is \ stored and takes effect when the session is started again, which a domain left disabled by a \ failed import or restore never does opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/LDAPReplicationDomainConfigChangeTest.java
New file @@ -0,0 +1,556 @@ /* * The contents of this file are subject to the terms of the Common Development and * Distribution License (the License). You may not use this file except in compliance with the * License. * * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the * specific language governing permission and limitations under the License. * * When distributing Covered Software, include this CDDL Header Notice in each file and include * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL * Header, with the fields enclosed by brackets [] replaced by your own identifying * information: "Portions copyright [year] [name of copyright owner]". * * Copyright 2026 3A Systems, LLC. */ package org.opends.server.replication.plugin; import static java.util.concurrent.TimeUnit.*; import static org.forgerock.opendj.ldap.ModificationType.*; import static org.opends.messages.ReplicationMessages.*; import static org.opends.server.TestCaseUtils.*; import static org.opends.server.protocols.internal.InternalClientConnection.*; import static org.testng.Assert.*; import java.lang.management.LockInfo; import java.lang.management.ManagementFactory; import java.lang.management.ThreadInfo; import java.lang.reflect.Field; import java.util.ArrayList; import java.util.List; import java.util.Set; import java.util.SortedSet; import java.util.TreeSet; import java.util.concurrent.CountDownLatch; import org.forgerock.i18n.LocalizableMessage; import org.forgerock.opendj.config.server.ConfigChangeResult; import org.forgerock.opendj.config.server.ConfigException; import org.forgerock.opendj.ldap.DN; import org.forgerock.opendj.ldap.ResultCode; import org.forgerock.opendj.server.config.meta.ReplicationDomainCfgDefn.AssuredType; import org.forgerock.opendj.server.config.meta.ReplicationDomainCfgDefn.IsolationPolicy; import org.forgerock.opendj.server.config.server.ExternalChangelogDomainCfg; import org.opends.server.TestCaseUtils; import org.opends.server.core.ModifyOperation; import org.opends.server.replication.ReplicationTestCase; import org.opends.server.replication.common.AssuredMode; import org.testng.annotations.Test; /** * Tests what a configuration change does, and does not, publish to a running * {@link LDAPReplicationDomain}. */ @SuppressWarnings("javadoc") public class LDAPReplicationDomainConfigChangeTest extends ReplicationTestCase { private static final int SERVER_ID = 1; private static final long HEARTBEAT_INTERVAL_IN_MS = 100000; private static final long ASSURED_TIMEOUT_IN_MS = 3000; private static final long NEW_ASSURED_TIMEOUT_IN_MS = 7000; /** The reason the external changelog configuration of the tests below cannot be read. */ private static final String UNDECODABLE_ECL_REASON = "the external changelog configuration cannot be decoded"; private static final String DOMAIN_CONFIG_NAME = "config change test"; /** * A configuration whose {@code cn=external changelog} child entry cannot be decoded: * the one thing {@code applyConfigurationChange()} does which can fail. */ private static final class UndecodableEclDomainFakeCfg extends DomainFakeCfg { private final DN configEntryDN; UndecodableEclDomainFakeCfg(DN baseDN, int serverId, SortedSet<String> replServers) { this(baseDN, serverId, replServers, null); } /** * @param configEntryDN * The entry this configuration is stored in, or {@code null} to keep the DN * of {@link DomainFakeCfg}, which the server configuration does not carry: * what an unreadable external changelog configuration does depends on * whether the entry which would carry it is there. */ UndecodableEclDomainFakeCfg(DN baseDN, int serverId, SortedSet<String> replServers, DN configEntryDN) { super(baseDN, serverId, replServers); this.configEntryDN = configEntryDN; } @Override public DN dn() { return configEntryDN != null ? configEntryDN : super.dn(); } @Override public ExternalChangelogDomainCfg getExternalChangelogDomain() throws ConfigException { throw new ConfigException(LocalizableMessage.raw(UNDECODABLE_ECL_REASON)); } } /** An ECL domain which refuses every change it is handed. */ private static final class RejectingExternalChangelogDomain extends ExternalChangelogDomain { RejectingExternalChangelogDomain(LDAPReplicationDomain domain, ExternalChangelogDomainCfg cfg) { super(domain, cfg); } @Override public ConfigChangeResult applyConfigurationChange(ExternalChangelogDomainCfg configuration) { final ConfigChangeResult ccr = new ConfigChangeResult(); ccr.setResultCode(ResultCode.CONSTRAINT_VIOLATION); ccr.addMessage(LocalizableMessage.raw("the ECL domain refused the change")); return ccr; } } @Test public void changeWhichCouldNotBeAppliedLeavesThePreviousConfigurationRunning() throws Exception { final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING); try { final SortedSet<String> replServers = unstartedReplicationServer(); final LDAPReplicationDomain domain = startDomain(new DomainFakeCfg(baseDN, SERVER_ID, replServers)); // Connected to no replication server, the default isolation policy rejects the updates. assertEquals(modifyBaseEntry(baseDN).getResultCode(), ResultCode.UNWILLING_TO_PERFORM); final DomainFakeCfg refused = new UndecodableEclDomainFakeCfg(baseDN, SERVER_ID, replServers); refused.setIsolationPolicy(IsolationPolicy.ACCEPT_ALL_UPDATES); domain.applyConfigurationChange(refused); assertEquals(modifyBaseEntry(baseDN).getResultCode(), ResultCode.UNWILLING_TO_PERFORM, "the domain went on running the isolation policy of a change it reported as failed"); } finally { MultimasterReplication.deleteDomain(baseDN); } } @Test public void changeWhichCouldNotBeAppliedSaysWhy() throws Exception { final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING); final DN configEntryDN = addDomainConfigurationEntry(baseDN); try { final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null); assertNotNull(domain, "the domain was not created from its configuration entry"); final ConfigChangeResult ccr = domain.applyConfigurationChange(new UndecodableEclDomainFakeCfg( baseDN, SERVER_ID, unstartedReplicationServer(), configEntryDN)); assertEquals(ccr.getResultCode(), ResultCode.OTHER); assertTrue(ccr.getMessages().toString().contains(UNDECODABLE_ECL_REASON), "the administrator was told the change failed, but not what failed: " + ccr.getMessages()); } finally { removeDomainConfigurationEntry(configEntryDN); } } @Test public void changeWhoseExternalChangelogConfigurationCannotBeReadIsRefusedBeforeItIsWritten() throws Exception { final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING); final DN configEntryDN = addDomainConfigurationEntry(baseDN); try { final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null); assertNotNull(domain, "the domain was not created from its configuration entry"); final List<LocalizableMessage> unacceptableReasons = new ArrayList<>(); final boolean acceptable = domain.isConfigurationChangeAcceptable( new UndecodableEclDomainFakeCfg( baseDN, SERVER_ID, unstartedReplicationServer(), configEntryDN), unacceptableReasons); assertFalse(acceptable, "the change was accepted, so the modified entry is written to the server " + "configuration before the domain finds out it cannot be applied"); assertTrue(unacceptableReasons.toString().contains(UNDECODABLE_ECL_REASON), "the administrator was not told what could not be read: " + unacceptableReasons); } finally { removeDomainConfigurationEntry(configEntryDN); } } @Test public void assuredConfigurationIsAppliedToADomainWhichOwnsItsSession() throws Exception { final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING); try { final SortedSet<String> replServers = unstartedReplicationServer(); final LDAPReplicationDomain domain = startDomain(assuredCfg(baseDN, replServers, ASSURED_TIMEOUT_IN_MS)); assertEquals(domain.getAssuredMode(), AssuredMode.SAFE_DATA_MODE); // Disabled is what an online import leaves the domain: it owns its session, so this // change is applied to it without a session being restarted for it. domain.disable(); waitForListenerThread(baseDN, false); final DomainFakeCfg safeRead = new DomainFakeCfg(baseDN, SERVER_ID, replServers, AssuredType.SAFE_READ, 1, -1, NEW_ASSURED_TIMEOUT_IN_MS, null); safeRead.setHeartbeatInterval(HEARTBEAT_INTERVAL_IN_MS); final ConfigChangeResult ccr = domain.applyConfigurationChange(safeRead); assertEquals(ccr.getResultCode(), ResultCode.SUCCESS, ccr.getMessages().toString()); assertFalse(hasListenerThread(baseDN), "the change started a session on a domain which was disabled for a total update"); assertEquals(domain.getAssuredMode(), AssuredMode.SAFE_READ_MODE, "the assured configuration was dropped although the change reported success"); assertEquals(domain.getAssuredTimeout(), NEW_ASSURED_TIMEOUT_IN_MS, "the assured timeout was dropped although the change reported success"); assertTrue(ccr.adminActionRequired(), "the assured configuration is negotiated as a session comes up, and this domain was" + " given no session to negotiate it over"); // Left as a total update leaves it: enabled back, on the configuration it was given. domain.enable(); } finally { MultimasterReplication.deleteDomain(baseDN); } } @Test public void changeIsRefusedWhenTheExternalChangelogDomainRejectsIt() throws Exception { final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING); try { final SortedSet<String> replServers = unstartedReplicationServer(); final DomainFakeCfg cfg = new DomainFakeCfg(baseDN, SERVER_ID, replServers); final LDAPReplicationDomain domain = startDomain(cfg); replaceEclDomain(domain, new RejectingExternalChangelogDomain(domain, cfg.getExternalChangelogDomain())); final ConfigChangeResult ccr = domain.applyConfigurationChange(new DomainFakeCfg(baseDN, SERVER_ID, replServers)); assertNotEquals(ccr.getResultCode(), ResultCode.SUCCESS, "the ECL domain refused the change and the domain reported it as applied"); assertFalse(ccr.getMessages().isEmpty(), "the refusal of the ECL domain was not passed on"); } finally { MultimasterReplication.deleteDomain(baseDN); } } @Test public void assuredTimeoutIsAppliedAlthoughItNeedsNoReconnection() throws Exception { final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING); try { final SortedSet<String> replServers = unstartedReplicationServer(); final LDAPReplicationDomain domain = startDomain( assuredCfg(baseDN, replServers, ASSURED_TIMEOUT_IN_MS)); assertEquals(domain.getAssuredTimeout(), ASSURED_TIMEOUT_IN_MS); // Only the timeout changes, which is the one assured property a session does not // have to be restarted for. final ConfigChangeResult ccr = domain.applyConfigurationChange(assuredCfg(baseDN, replServers, NEW_ASSURED_TIMEOUT_IN_MS)); assertEquals(domain.getAssuredTimeout(), NEW_ASSURED_TIMEOUT_IN_MS, "the new assured timeout was dropped although the change reported success"); assertFalse(ccr.adminActionRequired(), "a change which is live asked the administrator to act: " + ccr.getMessages()); } finally { MultimasterReplication.deleteDomain(baseDN); } } @Test public void changeNoSessionCouldBeRestartedForSaysItIsNotLiveYet() throws Exception { final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING); try { final LDAPReplicationDomain domain = startDomain(new DomainFakeCfg(baseDN, SERVER_ID, unstartedReplicationServer())); // Disabled is what an online import leaves the domain: it owns its session, and the // replication servers a domain talks to are negotiated as that session comes up. domain.disable(); waitForListenerThread(baseDN, false); final ConfigChangeResult ccr = domain.applyConfigurationChange( new DomainFakeCfg(baseDN, SERVER_ID, unstartedReplicationServer())); assertEquals(ccr.getResultCode(), ResultCode.SUCCESS, ccr.getMessages().toString()); assertTrue(ccr.adminActionRequired(), "the change was reported as fully applied although the session it needs was never" + " restarted for it, and a domain left disabled never restarts one"); assertTrue(ccr.getMessages().toString() .contains(NOTE_REPLICATION_DOMAIN_SESSION_NOT_RESTARTED.get(baseDN).toString()), "the administrator was not told which domain is waiting for a session: " + ccr.getMessages()); assertFalse(hasListenerThread(baseDN), "the change started a session on a domain which was disabled for a total update"); // Left as a total update leaves it: enabled back, on the configuration it was given. domain.enable(); } finally { MultimasterReplication.deleteDomain(baseDN); } } @Test public void externalChangelogConfigurationChangesTheSessionUnderTheServiceStateLock() throws Exception { final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING); Thread eclChange = null; try { final SortedSet<String> replServers = unstartedReplicationServer(); final LDAPReplicationDomain domain = startDomain(new DomainFakeCfg(baseDN, SERVER_ID, replServers)); final SortedSet<String> eclIncludes = new TreeSet<>(); eclIncludes.add("cn"); final CountDownLatch applied = new CountDownLatch(1); eclChange = new Thread(() -> { domain.changeConfig(eclIncludes, new TreeSet<String>()); applied.countDown(); }, "ECL configuration change"); /* * Asserted outside the block: a failure inside it would leave the thread running on * a domain the cleanup below is about to delete. What is waited for is the thread * blocking on the monitor rather than merely being slow, or the assertion would * hold for a change which simply took its time. */ final Object serviceStateLock = serviceStateLockOf(domain); final boolean blockedOnTheLock; final Set<String> eclIncludesWhileLocked; synchronized (serviceStateLock) { eclChange.start(); blockedOnTheLock = waitForBlockedOn(eclChange, serviceStateLock); eclIncludesWhileLocked = new TreeSet<>(domain.getEclIncludes()); } assertTrue(blockedOnTheLock, "the ECL configuration changed the session of the domain without holding serviceStateLock"); /* * What the assertion above alone does not tell apart: restartService(), at the end * of changeConfig(), takes this lock as well, so a change which applied the * attributes first and blocked on the lock afterwards would look just the same. The * whole of changeConfig() runs under the lock exactly when the attributes are still * unapplied while this thread holds it. */ assertFalse(eclIncludesWhileLocked.contains("cn"), "the ECL attributes were applied outside serviceStateLock, and only the session" + " restart which follows them was taken under it: " + eclIncludesWhileLocked); assertTrue(applied.await(30, SECONDS), "the ECL configuration change never completed"); assertTrue(domain.getEclIncludes().contains("cn"), "the ECL configuration change was reported as done and applied nothing"); } finally { if (eclChange != null) { eclChange.join(SECONDS.toMillis(30)); } MultimasterReplication.deleteDomain(baseDN); } } @Test public void externalChangelogConfigurationGivesNoSessionBackToADisabledDomain() throws Exception { final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING); try { final SortedSet<String> replServers = unstartedReplicationServer(); final LDAPReplicationDomain domain = startDomain(new DomainFakeCfg(baseDN, SERVER_ID, replServers)); // Disabled is what the domain is for the length of a total update: it owns its session. domain.disable(); waitForListenerThread(baseDN, false); final SortedSet<String> eclIncludes = new TreeSet<>(); eclIncludes.add("cn"); domain.changeConfig(eclIncludes, new TreeSet<String>()); assertFalse(hasListenerThread(baseDN), "the external changelog configuration started a session on a disabled domain"); assertTrue(domain.getEclIncludes().contains("cn"), "the attributes published to the external changelog were dropped"); // Left as a total update leaves it: enabled back, with its ServerState loaded again. domain.enable(); } finally { MultimasterReplication.deleteDomain(baseDN); } } /** * The listener thread of a domain is the one which says a session is running. Named * after the server id as well, or a domain another test class left behind on the same * base DN would answer for this one. */ private static boolean hasListenerThread(DN baseDN) { final String listenerName = "Replica DS(" + SERVER_ID + ") listener for domain \"" + baseDN + "\""; for (Thread thread : Thread.getAllStackTraces().keySet()) { if (thread.getName().contains(listenerName) && thread.isAlive()) { return true; } } return false; } private static void waitForListenerThread(DN baseDN, boolean running) throws Exception { final long deadline = System.currentTimeMillis() + SECONDS.toMillis(30); while (hasListenerThread(baseDN) != running && System.currentTimeMillis() < deadline) { Thread.sleep(100); } assertEquals(hasListenerThread(baseDN), running, "the listener thread of " + baseDN + " never " + (running ? "started" : "stopped")); } private DomainFakeCfg assuredCfg(DN baseDN, SortedSet<String> replServers, long assuredTimeout) { final DomainFakeCfg cfg = new DomainFakeCfg(baseDN, SERVER_ID, replServers, AssuredType.SAFE_DATA, 1, -1, assuredTimeout, null); // The same as the configuration in place, or the broker would restart the session for // the heartbeat interval and the change would no longer be the timeout alone. cfg.setHeartbeatInterval(HEARTBEAT_INTERVAL_IN_MS); return cfg; } /** A replication server which is not started, so that the domain never connects. */ private SortedSet<String> unstartedReplicationServer() throws Exception { final SortedSet<String> replServers = new TreeSet<>(); replServers.add("localhost:" + TestCaseUtils.findFreePort()); return replServers; } /** * Configures a domain the way the server does, through its configuration entries: what * an unreadable external changelog configuration does depends on whether the entry * which carries it is there, and the fake configurations of the other tests are stored * in no entry at all. * * @return the DN of the entry the configuration of the domain is stored in */ private DN addDomainConfigurationEntry(DN baseDN) throws Exception { addSynchroServerEntry( "dn: cn=" + DOMAIN_CONFIG_NAME + ",cn=domains," + SYNCHRO_PLUGIN_DN + "\n" + "objectClass: top\n" + "objectClass: ds-cfg-replication-domain\n" + "cn: " + DOMAIN_CONFIG_NAME + "\n" + "ds-cfg-base-dn: " + baseDN + "\n" + "ds-cfg-replication-server: localhost:" + TestCaseUtils.findFreePort() + "\n" + "ds-cfg-server-id: " + SERVER_ID + "\n"); final DN configEntryDN = synchroServerEntry.getName(); assertTrue(getServerContext().getConfigurationHandler() .hasEntry(DN.valueOf("cn=external changelog," + configEntryDN)), "the domain was configured without the external changelog entry these tests need"); return configEntryDN; } private void removeDomainConfigurationEntry(DN configEntryDN) throws Exception { // Deletes the "cn=external changelog" entry below it as well. deleteEntry(configEntryDN); configEntriesToCleanup.remove(configEntryDN); synchroServerEntry = null; } private LDAPReplicationDomain startDomain(DomainFakeCfg cfg) throws Exception { cfg.setHeartbeatInterval(HEARTBEAT_INTERVAL_IN_MS); final LDAPReplicationDomain domain = MultimasterReplication.createNewDomain(cfg); domain.start(); return domain; } private ModifyOperation modifyBaseEntry(DN baseDN) { return getRootConnection().processModify(modifyRequest(baseDN, REPLACE, "description", "test")); } /** * Whether the thread ends up waiting for this very monitor, rather than running to * completion or blocking on an unrelated one. */ private static boolean waitForBlockedOn(Thread thread, Object monitor) throws Exception { final long deadline = System.currentTimeMillis() + SECONDS.toMillis(10); while (System.currentTimeMillis() < deadline) { if (thread.getState() == Thread.State.TERMINATED) { return false; } final ThreadInfo info = ManagementFactory.getThreadMXBean().getThreadInfo(thread.getId()); final LockInfo blockedOn = info != null ? info.getLockInfo() : null; if (blockedOn != null && blockedOn.getIdentityHashCode() == System.identityHashCode(monitor)) { return true; } Thread.sleep(20); } return false; } private static Object serviceStateLockOf(LDAPReplicationDomain domain) throws Exception { final Field serviceStateLock = LDAPReplicationDomain.class.getDeclaredField("serviceStateLock"); serviceStateLock.setAccessible(true); return serviceStateLock.get(domain); } private static void replaceEclDomain(LDAPReplicationDomain domain, ExternalChangelogDomain eclDomain) throws Exception { final Field field = LDAPReplicationDomain.class.getDeclaredField("eclDomain"); field.setAccessible(true); field.set(domain, eclDomain); } }
