From 08727ec559e7018316d3b25b04a828ee0a127b2a Mon Sep 17 00:00:00 2001
From: Jean-Noel Rouvignac <jean-noel.rouvignac@forgerock.com>
Date: Tue, 28 Jan 2014 14:33:41 +0000
Subject: [PATCH] OPENDJ-1262 NPE in ChangeNumberIndex during server startup
---
opends/src/server/org/opends/server/replication/plugin/MultimasterReplication.java | 59 +++++++++++++++++++++++++++++++++++++++++++++++------------
1 files changed, 47 insertions(+), 12 deletions(-)
diff --git a/opends/src/server/org/opends/server/replication/plugin/MultimasterReplication.java b/opends/src/server/org/opends/server/replication/plugin/MultimasterReplication.java
index 1a2e8c8..f78bf74 100644
--- a/opends/src/server/org/opends/server/replication/plugin/MultimasterReplication.java
+++ b/opends/src/server/org/opends/server/replication/plugin/MultimasterReplication.java
@@ -22,7 +22,7 @@
*
*
* Copyright 2006-2010 Sun Microsystems, Inc.
- * Portions Copyright 2011-2013 ForgeRock AS
+ * Portions Copyright 2011-2014 ForgeRock AS
*/
package org.opends.server.replication.plugin;
@@ -30,6 +30,7 @@
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicReference;
import org.opends.messages.Message;
import org.opends.server.admin.server.ConfigurationAddListener;
@@ -67,7 +68,7 @@
ExportTaskListener
{
private ReplicationServerListener replicationServerListener = null;
- private static Map<DN, LDAPReplicationDomain> domains =
+ private static final Map<DN, LDAPReplicationDomain> domains =
new ConcurrentHashMap<DN, LDAPReplicationDomain>(4) ;
/**
@@ -80,7 +81,7 @@
/**
* The list of ReplayThread threads.
*/
- private static List<ReplayThread> replayThreads =
+ private static final List<ReplayThread> replayThreads =
new ArrayList<ReplayThread>();
/**
@@ -88,7 +89,16 @@
*/
private static int replayThreadNumber = 10;
- private boolean isRegistered = false;
+ /**
+ * enum that symbolizes the state of the multimaster replication.
+ */
+ private static enum State
+ {
+ STARTING, RUNNING, STOPPING
+ }
+
+ private static final AtomicReference<State> state =
+ new AtomicReference<State>(State.STARTING);
/**
* The configurable connection/handshake timeout.
@@ -338,9 +348,12 @@
try
{
LDAPReplicationDomain rd = createNewDomain(configuration);
- if (isRegistered)
+ if (State.RUNNING.equals(state.get()))
{
rd.start();
+ if (State.STOPPING.equals(state.get())) {
+ rd.shutdown();
+ }
}
return new ConfigChangeResult(ResultCode.SUCCESS, false);
} catch (ConfigException e)
@@ -540,19 +553,16 @@
@Override
public void finalizeSynchronizationProvider()
{
- isRegistered = false;
+ setState(State.STOPPING);
- // shutdown all the domains
for (LDAPReplicationDomain domain : domains.values())
{
domain.shutdown();
}
domains.clear();
- // Stop replay threads
stopReplayThreads();
- // shutdown the ReplicationServer Service if necessary
if (replicationServerListener != null)
replicationServerListener.shutdown();
@@ -757,13 +767,20 @@
@Override
public void completeSynchronizationProvider()
{
- isRegistered = true;
-
- // start all the domains
for (LDAPReplicationDomain domain : domains.values())
{
domain.start();
}
+ setState(State.RUNNING);
+ }
+
+ private void setState(State newState)
+ {
+ state.set(newState);
+ synchronized (state)
+ {
+ state.notifyAll();
+ }
}
/**
@@ -806,6 +823,24 @@
*/
public static boolean isECLEnabledDomain(DN baseDN)
{
+ if (State.STARTING.equals(state.get()))
+ {
+ synchronized (state)
+ {
+ while (State.STARTING.equals(state.get()))
+ {
+ try
+ {
+ state.wait();
+ }
+ catch (InterruptedException ignored)
+ {
+ // loop and check state again
+ }
+ }
+ }
+ }
+ // if state is STOPPING, then we need to return from this method
for (LDAPReplicationDomain domain : domains.values())
{
if (domain.isECLEnabled() && domain.getBaseDN().equals(baseDN))
--
Gitblit v1.10.0