external-software/github_OpenIdentityPlatform_OpenDJ.git

			@@ -39,8 +39,10 @@
			import java.util.Set;
			import java.util.concurrent.CopyOnWriteArraySet;
			import java.util.concurrent.atomic.AtomicBoolean;
			import java.util.concurrent.atomic.AtomicInteger;

			import org.forgerock.i18n.LocalizableMessage;
			import org.forgerock.i18n.LocalizableMessageBuilder;
			import org.forgerock.i18n.slf4j.LocalizedLogger;
			import org.forgerock.opendj.config.server.ConfigChangeResult;
			import org.forgerock.opendj.config.server.ConfigException;
			@@ -92,7 +94,18 @@
			{
			private String serverURL;

			private ServerSocket listenSocket;
			/**
			* Number of attempts to bind the listen port before giving up. The port may be held for a
			* short while by a socket which is being closed, so a few retries make the start-up
			* resilient to such transient conditions.
			*/
			private static final int LISTEN_BIND_ATTEMPTS = 5;
			/** Delay between two attempts to bind the listen port. */
			private static final long LISTEN_BIND_RETRY_DELAY_MS = 200;
			/** Timeout of the diagnostic probe performed when the listen port cannot be bound. */
			private static final int LISTEN_PORT_PROBE_TIMEOUT_MS = 200;

			private volatile ServerSocket listenSocket;
			private Thread listenThread;
			private Thread connectThread;

			@@ -109,8 +122,16 @@
			/** The backend that allow to search the changes (external changelog). */
			private ChangelogBackend changelogBackend;

			/**
			* Whether this instance registered the virtual attribute rules of the external changelog.
			* They are registered globally, by attribute name, so an instance which did not register
			* them must not deregister them: it would strip them from the instance which did.
			*/
			private boolean externalChangelogRegistered;

			private final AtomicBoolean shutdown = new AtomicBoolean();
			private boolean stopListen;
			/** Written by the thread applying a configuration change, read by the listen thread. */
			private volatile boolean stopListen;
			private final ReplSessionSecurity replSessionSecurity;

			private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();
			@@ -124,6 +145,16 @@
			*/
			private static final Set<Integer> localPorts = new CopyOnWriteArraySet<>();

			/**
			* Number of attempts to bind a listen port which failed in this VM.
			* <p>
			* This is required for unit testing: it lets a test which holds a listen port release it
			* as soon as a replication server has actually failed to bind it, instead of after a
			* delay which would make the test vacuous when it is too long, and flaky when it is too
			* short.
			*/
			static final AtomicInteger listenPortBindFailures = new AtomicInteger();

			/** Monitors for synchronizing domain creation with the connect thread. */
			private final Object domainTicketLock = new Object();
			private final Object connectThreadLock = new Object();
			@@ -177,15 +208,25 @@
			this.dsrsShutdownSync = dsrsShutdownSync;
			this.domainPredicate = predicate;

			enableExternalChangeLog();
			ServerContext serverContext = DirectoryServer.getInstance().getServerContext();
			cryptoSuite = serverContext.getCryptoManager().
			newCryptoSuite(cfg.getCipherTransformation(), cfg.getCipherKeyLength(), cfg.isConfidentialityEnabled());
			try
			{
			enableExternalChangeLog();
			ServerContext serverContext = DirectoryServer.getInstance().getServerContext();
			cryptoSuite = serverContext.getCryptoManager().
			newCryptoSuite(cfg.getCipherTransformation(), cfg.getCipherKeyLength(), cfg.isConfidentialityEnabled());

			this.changelogDB = new FileChangelogDB(this, config.getReplicationDBDirectory(), cryptoSuite);
			this.changelogDB = new FileChangelogDB(this, config.getReplicationDBDirectory(), cryptoSuite);

			replSessionSecurity = new ReplSessionSecurity();
			initialize();
			replSessionSecurity = new ReplSessionSecurity();
			initialize();
			}
			catch (ConfigException e)
			{
			// This instance is never returned to the caller, so nothing will ever shut it down:
			// release what the initialization managed to acquire before failing.
			abortInitialization();
			throw e;
			}
			cfg.addChangeListener(this);

			localPorts.add(getReplicationPort());
			@@ -441,8 +482,14 @@
			return true;
			}

			/** Initialization function for the replicationServer. */
			private void initialize()
			/**
			* Initialization function for the replicationServer.
			*
			* @throws ConfigException
			* when the replication server cannot be started, in particular when its listen
			* port cannot be bound.
			*/
			private void initialize() throws ConfigException
			{
			shutdown.set(false);

			@@ -451,8 +498,8 @@
			this.changelogDB.initializeDB();

			setServerURL();
			listenSocket = new ServerSocket();
			listenSocket.bind(new InetSocketAddress(getReplicationPort()));
			// Assigned before the threads are created, so that a failure below still releases it.
			listenSocket = bindListenPort(getReplicationPort());

			// creates working threads: we must first connect, then start to listen.
			if (logger.isTraceEnabled())
			@@ -467,8 +514,7 @@
			logger.trace("RS " + getMonitorInstanceName() + " creates listen thread");
			}

			listenThread = new ReplicationServerListenThread(this);
			listenThread.start();
			startListenThread(listenSocket);

			if (logger.isTraceEnabled())
			{
			@@ -476,14 +522,265 @@
			}
			} catch (UnknownHostException e)
			{
			logger.error(ERR_UNKNOWN_HOSTNAME);
			// Not logged here: the caller reports the ConfigException, logging it once.
			logger.traceException(e);
			throw new ConfigException(ERR_UNKNOWN_HOSTNAME.get(), e);
			} catch (IOException e)
			{
			logger.error(ERR_COULD_NOT_BIND_CHANGELOG, getReplicationPort(), e.getMessage());
			// A replication server whose listen port is not bound is dead: every consumer would
			// otherwise only learn about it as a "connection refused" somewhere else.
			logger.traceException(e);
			throw new ConfigException(bindFailureMessage(getReplicationPort(), e), e);
			}
			}

			/**
			* Binds the given port, retrying a few times when it is momentarily unavailable.
			*
			* @param port
			* the port to bind
			* @return the bound listen socket
			* @throws IOException
			* if the port could not be bound within {@link #LISTEN_BIND_ATTEMPTS} attempts
			*/
			private ServerSocket bindListenPort(int port) throws IOException
			{
			for (int attempt = 1; ; attempt++)
			{
			final ServerSocket socket = new ServerSocket();
			try
			{
			socket.bind(new InetSocketAddress(port));
			if (attempt > 1)
			{
			logger.info(NOTE_BOUND_CHANGELOG_AFTER_RETRY, port, attempt);
			}
			return socket;
			}
			catch (IOException e)
			{
			close(socket);
			listenPortBindFailures.incrementAndGet();
			if (attempt >= LISTEN_BIND_ATTEMPTS)
			{
			throw e;
			}
			// The port is probed only when giving up: probing it on every attempt would cost a
			// connection to whoever holds it, and delay the next attempt for nothing.
			logger.warn(WARN_RETRYING_BIND_CHANGELOG, port, getExceptionMessage(e), LISTEN_BIND_RETRY_DELAY_MS);
			try
			{
			Thread.sleep(LISTEN_BIND_RETRY_DELAY_MS);
			}
			catch (InterruptedException e2)
			{
			Thread.currentThread().interrupt();
			throw e;
			}
			}
			}
			}

			/**
			* Stops the listen thread and releases the listen port.
			*
			* @throws InterruptedException
			* if this thread is interrupted while waiting for the listen thread to stop
			*/
			private void stopListenThread() throws InterruptedException
			{
			stopListen = true;
			close(listenSocket);
			if (listenThread != null)
			{
			listenThread.join();
			listenThread = null;
			}
			}

			/**
			* Starts a listen thread on the provided listen socket.
			* <p>
			* {@code stopListen} is only cleared here, i.e. once the listen port is bound, so a
			* failure to bind leaves this replication server consistently stopped rather than with a
			* listen thread which would spin on a closed socket.
			*
			* @param boundListenSocket
			* the bound socket the listen thread will accept connections on
			*/
			private void startListenThread(ServerSocket boundListenSocket)
			{
			listenSocket = boundListenSocket;
			stopListen = false;
			listenThread = new ReplicationServerListenThread(this);
			listenThread.start();
			}

			/**
			* Switches the listen port to the one of the provided configuration.
			* <p>
			* The new port is bound while the current one is still open and serving, so a failure
			* leaves this replication server listening on its current port, with its current
			* configuration: there is nothing to roll back, and no window during which this
			* replication server advertises a port that nothing listens to.
			*
			* @param newConfig
			* the configuration being applied, whose listen port differs from the current one
			* @param ccr
			* the result of the configuration change, to which a failure is added
			* @return {@code true} when this replication server listens on the new port, in which
			* case {@code newConfig} has become its configuration
			*/
			private boolean switchListenPort(ReplicationServerCfg newConfig, ConfigChangeResult ccr)
			{
			final ReplicationServerCfg previousConfig = this.config;
			final String previousServerURL = serverURL;
			final int newPort = newConfig.getReplicationPort();
			ServerSocket newListenSocket = null;
			try
			{
			// The current listen socket is still open and serving while the new port is bound.
			newListenSocket = bindListenPort(newPort);

			this.config = newConfig;
			setServerURL();

			stopListenThread();
			startListenThread(newListenSocket);
			newListenSocket = null;

			localPorts.remove(previousConfig.getReplicationPort());
			localPorts.add(newPort);
			return true;
			}
			catch (UnknownHostException e)
			{
			logger.traceException(e);
			ccr.setResultCode(ResultCode.OPERATIONS_ERROR);
			ccr.addMessage(ERR_UNKNOWN_HOSTNAME.get());
			}
			catch (IOException e)
			{
			// The new port could not be bound, the current listen socket was left untouched.
			logger.traceException(e);
			ccr.setResultCode(ResultCode.OPERATIONS_ERROR);
			ccr.addMessage(bindFailureMessage(newPort, e));
			}
			catch (InterruptedException e)
			{
			// The previous listen thread may still be running, so do not hand it a new socket:
			// stopListen is still set, which makes that thread stop as soon as it wakes up.
			Thread.currentThread().interrupt();
			logger.traceException(e);
			ccr.setResultCode(ResultCode.OPERATIONS_ERROR);
			ccr.addMessage(ERR_COULD_NOT_STOP_LISTEN_THREAD.get(getExceptionMessage(e)));
			}
			// The failure is reported through the ConfigChangeResult, which the configuration
			// handler logs: nothing of the new configuration was applied.
			this.config = previousConfig;
			serverURL = previousServerURL;
			close(newListenSocket);
			return false;
			}

			/**
			* Returns the message of a failed bind of the listen port, i.e. the error itself plus a
			* best effort diagnostic of what holds the port.
			*
			* @param port
			* the port which could not be bound
			* @param cause
			* the error returned by the failed bind
			* @return the message describing the failure
			*/
			private static LocalizableMessage bindFailureMessage(int port, IOException cause)
			{
			return new LocalizableMessageBuilder(ERR_COULD_NOT_BIND_CHANGELOG.get(port, getExceptionMessage(cause)))
			.append(" ").append(describeListenPortHolder(port)).toMessage();
			}

			/**
			* Returns a best effort diagnostic of what holds the given port.
			* <p>
			* The port is probed over the loopback interface only, so the diagnostic reports what was
			* observed there and nothing more: a socket bound to another address holds the port
			* without ever accepting a loopback connection, and cannot be told apart from a socket
			* which does not accept connections at all, e.g. a client socket which was given that
			* port as its local port.
			*
			* @param port
			* the port which could not be bound
			* @return the message describing what was observed on the loopback interface
			*/
			private static LocalizableMessage describeListenPortHolder(int port)
			{
			final String probedAddress = InetAddress.getLoopbackAddress().getHostAddress() + ":" + port;
			try (Socket probe = new Socket())
			{
			// Without SO_REUSEADDR a probe which self-connects would, once closed, hold the port
			// in TIME_WAIT and defeat the very bind it is diagnosing.
			probe.setReuseAddress(true);
			probe.connect(new InetSocketAddress(InetAddress.getLoopbackAddress(), port), LISTEN_PORT_PROBE_TIMEOUT_MS);
			if (isSelfConnection(probe))
			{
			// The kernel can only give the probed port to the probe as its local port while
			// that port is free: whoever held it released it in the meantime.
			return ERR_COULD_NOT_BIND_CHANGELOG_PORT_FREE.get(probedAddress);
			}
			return ERR_COULD_NOT_BIND_CHANGELOG_LISTENING.get(probedAddress);
			}
			catch (IOException e)
			{
			logger.traceException(e);
			return ERR_COULD_NOT_BIND_CHANGELOG_NOT_ACCEPTING.get(probedAddress);
			}
			}

			/**
			* Releases what the initialization acquired before it failed.
			* <p>
			* It runs on a partially constructed instance, so every field it uses may still be
			* unassigned.
			*/
			private void abortInitialization()
			{
			shutdown.set(true);
			if (connectThread != null)
			{
			connectThread.interrupt();
			}
			close(listenSocket);
			if (listenThread != null)
			{
			listenThread.interrupt();
			}
			shutdownExternalChangelog();
			if (this.changelogDB != null)
			{
			try
			{
			this.changelogDB.shutdownDB();
			}
			catch (ChangelogException ignored)
			{
			logger.traceException(ignored);
			}
			}
			}

			/**
			* Indicates whether this replication server has bound its listen port, i.e. whether it
			* can accept connections from directory servers and from other replication servers.
			*
			* @return {@code true} if this replication server is listening
			*/
			public boolean isListening()
			{
			final ServerSocket socket = listenSocket;
			return socket != null && socket.isBound() && !socket.isClosed();
			}

			/**
			* Enable the external changelog if it is not already enabled.
			* <p>
			* The external changelog is provided by the changelog backend.
			@@ -515,6 +812,9 @@
			getExceptionMessage(e)));
			}

			// Set before the rules are registered, so that a partial registration is released
			// too: this instance is then the one which registered whatever is registered.
			externalChangelogRegistered = true;
			registerVirtualAttributeRules();
			}
			catch (Exception e)
			@@ -535,7 +835,14 @@
			changelogBackend.finalizeBackend();
			changelogBackend = null;
			}
			deregisterVirtualAttributeRules();
			if (externalChangelogRegistered)
			{
			// Virtual attribute rules are registered globally, by attribute name: deregistering
			// rules which this instance did not register would strip them from the instance
			// which did, e.g. when this one took the early return of enableExternalChangeLog().
			externalChangelogRegistered = false;
			deregisterVirtualAttributeRules();
			}
			}

			private List<VirtualAttributeRule> getVirtualAttributesRules() throws DirectoryException
			@@ -863,17 +1170,14 @@
			{
			int port = configuration.getReplicationPort();

			try
			try (ServerSocket tmpSocket = new ServerSocket())
			{
			ServerSocket tmpSocket = new ServerSocket();
			tmpSocket.bind(new InetSocketAddress(port));
			tmpSocket.close();
			return true;
			}
			catch (Exception e)
			{
			LocalizableMessage message = ERR_COULD_NOT_BIND_CHANGELOG.get(port, e.getMessage());
			unacceptableReasons.add(message);
			unacceptableReasons.add(ERR_COULD_NOT_BIND_CHANGELOG.get(port, getExceptionMessage(e)));
			return false;
			}
			}
			@@ -884,11 +1188,21 @@
			{
			final ConfigChangeResult ccr = new ConfigChangeResult();

			final Set<HostPort> oldRSAddresses = getConfiguredRSAddresses();
			final ReplicationServerCfg oldConfig = this.config;

			// Changing the listen port requires to stop the listen thread and restart it. It is
			// done first, and the new port is bound before the current one is released, so that a
			// change which cannot be applied leaves this replication server as it was, instead of
			// half configured and, worse, without any listener.
			if (configuration.getReplicationPort() != oldConfig.getReplicationPort()
			&& !switchListenPort(configuration, ccr))
			{
			return ccr;
			}

			// Some of those properties change don't need specific code.
			// They will be applied for next connections. Some others have immediate effect
			final Set<HostPort> oldRSAddresses = getConfiguredRSAddresses();

			final ReplicationServerCfg oldConfig = this.config;
			this.config = configuration;

			disconnectRemovedReplicationServers(oldRSAddresses);
			@@ -915,39 +1229,6 @@
			cryptoSuite.newParameters(config.getCipherTransformation(), config.getCipherKeyLength(),
			config.isConfidentialityEnabled());

			// changing the listen port requires to stop the listen thread
			// and restart it.
			if (getReplicationPort() != oldConfig.getReplicationPort())
			{
			stopListen = true;
			try
			{
			close(listenSocket);
			if (listenThread != null)
			{
			listenThread.join();
			}
			stopListen = false;

			setServerURL();
			listenSocket = new ServerSocket();
			listenSocket.bind(new InetSocketAddress(getReplicationPort()));

			listenThread = new ReplicationServerListenThread(this);
			listenThread.start();
			}
			catch (IOException e)
			{
			logger.traceException(e);
			logger.error(ERR_COULD_NOT_CLOSE_THE_SOCKET, e);
			}
			catch (InterruptedException e)
			{
			logger.traceException(e);
			logger.error(ERR_COULD_NOT_STOP_LISTEN_THREAD, e);
			}
			}

			// Update period value for monitoring publishers
			if (oldConfig.getMonitoringPeriod() != config.getMonitoringPeriod())
			{
			@@ -1031,7 +1312,13 @@
			public boolean isConfigurationChangeAcceptable(
			ReplicationServerCfg configuration, List<LocalizableMessage> unacceptableReasons)
			{
			return true;
			if (configuration.getReplicationPort() == getReplicationPort())
			{
			return true;
			}
			// The change is persisted before it is applied, so rejecting a port which cannot be
			// bound is the only way to keep the configuration and the listen port in sync.
			return isConfigurationAcceptable(configuration, unacceptableReasons);
			}

			/**