bug5194: added com.sap.sse.replication.persistence to store replica descriptors

java/com.sap.sse.feature/feature.xml

@@ -220,7 +220,7 @@ Computes leaderboard information for sailing races and offers RESTful APIs to al
          version="0.0.0"
          unpack="false"/>
 
-      <plugin
+   <plugin
          id="com.sap.sse.jettyextensions"
          download-size="0"
          install-size="0"
@@ -228,4 +228,11 @@ Computes leaderboard information for sailing races and offers RESTful APIs to al
          fragment="true"
          unpack="false"/>
 
+   <plugin
+         id="com.sap.sse.replication.persistence"
+         download-size="0"
+         install-size="0"
+         version="0.0.0"
+         unpack="false"/>
+
 </feature>

java/com.sap.sse.replication.interfaces/META-INF/MANIFEST.MF

@@ -6,7 +6,8 @@ Bundle-Version: 1.0.0.qualifier
 Bundle-Vendor: SAP
 Automatic-Module-Name: com.sap.sse.replication.interfaces
 Bundle-RequiredExecutionEnvironment: JavaSE-1.8
-Export-Package: com.sap.sse.replication
+Export-Package: com.sap.sse.replication,
+ com.sap.sse.replication.impl
 Import-Package: com.rabbitmq.client;version="2.8.4",
  org.json.simple
 Require-Bundle: com.sap.sse.operationaltransformation;bundle-version="1.0.0",

java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/impl/ReplicaDescriptorImpl.java

@@ -27,10 +27,18 @@ public class ReplicaDescriptorImpl implements ReplicaDescriptor {
     /**
      * Sets the registration time to now.
      */
-    public ReplicaDescriptorImpl(InetAddress ipAddress, UUID serverUuid, String additionalInformation, String[] replicableIdsAsStrings) {
+    public ReplicaDescriptorImpl(InetAddress ipAddress, UUID id, String additionalInformation, String[] replicableIdsAsStrings) {
+        this(ipAddress, id, MillisecondsTimePoint.now(), additionalInformation, replicableIdsAsStrings);
+    }
+
+    /**
+     * Sets the registration time to now.
+     */
+    public ReplicaDescriptorImpl(InetAddress ipAddress, UUID id, TimePoint registrationTime,
+            String additionalInformation, String[] replicableIdsAsStrings) {
         assert replicableIdsAsStrings != null && replicableIdsAsStrings.length > 0;
-        this.uuid = serverUuid;
-        this.registrationTime = MillisecondsTimePoint.now();
+        this.uuid = id;
+        this.registrationTime = registrationTime;
         this.ipAddress = ipAddress;
         this.additionalInformation = additionalInformation;
         this.replicableIdsAsStrings = replicableIdsAsStrings;

java/com.sap.sse.replication.persistence/.classpath

@@ -0,0 +1,7 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<classpath>
+	<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/JavaSE-1.8"/>
+	<classpathentry kind="con" path="org.eclipse.pde.core.requiredPlugins"/>
+	<classpathentry kind="src" path="src"/>
+	<classpathentry kind="output" path="bin"/>
+</classpath>

java/com.sap.sse.replication.persistence/.project

@@ -0,0 +1,28 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<projectDescription>
+	<name>com.sap.sse.replication.persistence</name>
+	<comment></comment>
+	<projects>
+	</projects>
+	<buildSpec>
+		<buildCommand>
+			<name>org.eclipse.jdt.core.javabuilder</name>
+			<arguments>
+			</arguments>
+		</buildCommand>
+		<buildCommand>
+			<name>org.eclipse.pde.ManifestBuilder</name>
+			<arguments>
+			</arguments>
+		</buildCommand>
+		<buildCommand>
+			<name>org.eclipse.pde.SchemaBuilder</name>
+			<arguments>
+			</arguments>
+		</buildCommand>
+	</buildSpec>
+	<natures>
+		<nature>org.eclipse.pde.PluginNature</nature>
+		<nature>org.eclipse.jdt.core.javanature</nature>
+	</natures>
+</projectDescription>

java/com.sap.sse.replication.persistence/.settings/org.eclipse.jdt.core.prefs

@@ -0,0 +1,7 @@
+eclipse.preferences.version=1
+org.eclipse.jdt.core.compiler.codegen.inlineJsrBytecode=enabled
+org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.8
+org.eclipse.jdt.core.compiler.compliance=1.8
+org.eclipse.jdt.core.compiler.problem.assertIdentifier=error
+org.eclipse.jdt.core.compiler.problem.enumIdentifier=error
+org.eclipse.jdt.core.compiler.source=1.8

java/com.sap.sse.replication.persistence/META-INF/MANIFEST.MF

@@ -0,0 +1,16 @@
+Manifest-Version: 1.0
+Bundle-ManifestVersion: 2
+Bundle-Name: Persistence
+Bundle-SymbolicName: com.sap.sse.replication.persistence
+Bundle-Version: 1.0.0.qualifier
+Bundle-Vendor: SAP
+Automatic-Module-Name: com.sap.sse.replication.persistence
+Bundle-RequiredExecutionEnvironment: JavaSE-1.8
+Require-Bundle: com.sap.sse.replication.interfaces,
+ org.mongodb.mongo-java-driver;bundle-version="3.6.4",
+ com.sap.sse.common
+Import-Package: com.sap.sse.mongodb,
+ org.osgi.framework;version="1.8.0"
+Export-Package: com.sap.sse.replication.persistence
+Bundle-ActivationPolicy: lazy
+Bundle-Activator: com.sap.sse.replication.persistence.impl.Activator

java/com.sap.sse.replication.persistence/build.properties

@@ -0,0 +1,4 @@
+source.. = src/
+output.. = bin/
+bin.includes = META-INF/,\
+               .

java/com.sap.sse.replication.persistence/pom.xml

@@ -0,0 +1,12 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd" xmlns="http://maven.apache.org/POM/4.0.0"
+    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <artifactId>root</artifactId>
+    <groupId>com.sap.sailing</groupId>
+    <version>1.0.0-SNAPSHOT</version>
+  </parent>
+  <artifactId>com.sap.sse.replication.persistence</artifactId>
+  <packaging>eclipse-plugin</packaging>  
+</project>

java/com.sap.sse.replication.persistence/src/com/sap/sse/replication/persistence/DomainObjectFactory.java

@@ -0,0 +1,7 @@
+package com.sap.sse.replication.persistence;
+
+import com.sap.sse.replication.ReplicaDescriptor;
+
+public interface DomainObjectFactory {
+    Iterable<ReplicaDescriptor> loadReplicaDescriptors();
+}

java/com.sap.sse.replication.persistence/src/com/sap/sse/replication/persistence/MongoObjectFactory.java

@@ -0,0 +1,9 @@
+package com.sap.sse.replication.persistence;
+
+import com.sap.sse.replication.ReplicaDescriptor;
+
+public interface MongoObjectFactory {
+    void storeReplicaDescriptor(ReplicaDescriptor replicaDescriptor);
+
+    void removeReplicaDescriptor(ReplicaDescriptor replicaDescriptor);
+}

java/com.sap.sse.replication.persistence/src/com/sap/sse/replication/persistence/PersistenceFactory.java

@@ -0,0 +1,17 @@
+package com.sap.sse.replication.persistence;
+
+import com.sap.sse.mongodb.MongoDBService;
+import com.sap.sse.replication.persistence.impl.PersistenceFactoryImpl;
+
+public interface PersistenceFactory {
+    PersistenceFactory INSTANCE = new PersistenceFactoryImpl();
+
+    DomainObjectFactory getDomainObjectFactory(MongoDBService mongoDbService);
+
+    MongoObjectFactory getMongoObjectFactory(MongoDBService mongoDbService);
+
+    DomainObjectFactory getDefaultDomainObjectFactory();
+
+    MongoObjectFactory getDefaultMongoObjectFactory();
+
+}

java/com.sap.sse.replication.persistence/src/com/sap/sse/replication/persistence/impl/Activator.java

@@ -0,0 +1,19 @@
+package com.sap.sse.replication.persistence.impl;
+
+import org.osgi.framework.BundleActivator;
+import org.osgi.framework.BundleContext;
+
+import com.sap.sse.mongodb.MongoDBService;
+
+public class Activator implements BundleActivator {
+    @Override
+    public void start(BundleContext context) throws Exception {
+        for (CollectionNames name : CollectionNames.values()) {
+            MongoDBService.INSTANCE.registerExclusively(CollectionNames.class, name.name());
+        }
+    }
+
+    @Override
+    public void stop(BundleContext context) throws Exception {
+    }
+}

java/com.sap.sse.replication.persistence/src/com/sap/sse/replication/persistence/impl/CollectionNames.java

@@ -0,0 +1,5 @@
+package com.sap.sse.replication.persistence.impl;
+
+public enum CollectionNames {
+    REPLICA_DESCRIPTORS
+}

java/com.sap.sse.replication.persistence/src/com/sap/sse/replication/persistence/impl/DomainObjectFactoryImpl.java

@@ -0,0 +1,64 @@
+package com.sap.sse.replication.persistence.impl;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import org.bson.Document;
+
+import com.mongodb.client.MongoCollection;
+import com.mongodb.client.MongoDatabase;
+import com.sap.sse.common.TimePoint;
+import com.sap.sse.common.impl.MillisecondsTimePoint;
+import com.sap.sse.replication.ReplicaDescriptor;
+import com.sap.sse.replication.impl.ReplicaDescriptorImpl;
+import com.sap.sse.replication.persistence.DomainObjectFactory;
+
+public class DomainObjectFactoryImpl implements DomainObjectFactory {
+    private static final Logger logger = Logger.getLogger(DomainObjectFactoryImpl.class.getName());
+    private final MongoDatabase db;
+    
+    public DomainObjectFactoryImpl(MongoDatabase db) {
+        super();
+        this.db = db;
+    }
+
+    private MongoCollection<Document> getReplicaDescriptorCollection() {
+        final MongoCollection<Document> result = db.getCollection(CollectionNames.REPLICA_DESCRIPTORS.name());
+        return result;
+    }
+
+    @Override
+    public Iterable<ReplicaDescriptor> loadReplicaDescriptors() {
+        final Map<UUID, ReplicaDescriptor> result = new HashMap<>();
+        for (final Document o : getReplicaDescriptorCollection().find()) {
+            final ReplicaDescriptor replicaDescriptor = loadReplicaDescriptor(o);
+            if (replicaDescriptor != null) {
+                result.put(replicaDescriptor.getUuid(), replicaDescriptor);
+            }
+        }
+        return result.values();
+    }
+
+    private ReplicaDescriptor loadReplicaDescriptor(Document o) {
+        final UUID uuid = UUID.fromString(o.getString(FieldNames.REPLICA_ID_AS_STRING.name()));
+        InetAddress ipAddress;
+        String ipAddressString = o.getString(FieldNames.REPLICA_IP_ADDRESS.name());
+        try {
+            ipAddress = InetAddress.getByName(ipAddressString);
+        } catch (UnknownHostException e) {
+            final String message = "Internal error: the IP address "+ipAddressString+
+                    " cannot be converted into an InetAddress";
+            logger.log(Level.WARNING, message, e);
+            throw new RuntimeException(message, e);
+        }
+        final TimePoint registrationTime = new MillisecondsTimePoint(o.getLong(FieldNames.REPLICA_REGISTRATION_TIME_MILLIS));
+        final String additionalInformation = o.getString(FieldNames.REPLICA_ADDITIONAL_INFORMATION.name());
+        final String[] replicableIdsAsStrings = o.get(FieldNames.REPLICA_REPLICABLE_IDS_AS_STRINGS.name(), new String[0]);
+        return new ReplicaDescriptorImpl(ipAddress, uuid, registrationTime, additionalInformation, replicableIdsAsStrings);
+    }
+}

java/com.sap.sse.replication.persistence/src/com/sap/sse/replication/persistence/impl/FieldNames.java

@@ -0,0 +1,6 @@
+package com.sap.sse.replication.persistence.impl;
+
+public enum FieldNames {
+    REPLICA_ID_AS_STRING, REPLICA_IP_ADDRESS, REPLICA_REGISTRATION_TIME_MILLIS,
+    REPLICA_ADDITIONAL_INFORMATION, REPLICA_REPLICABLE_IDS_AS_STRINGS;
+}

java/com.sap.sse.replication.persistence/src/com/sap/sse/replication/persistence/impl/MongoObjectFactoryImpl.java

@@ -0,0 +1,45 @@
+package com.sap.sse.replication.persistence.impl;
+
+import org.bson.Document;
+
+import com.mongodb.WriteConcern;
+import com.mongodb.client.MongoCollection;
+import com.mongodb.client.MongoDatabase;
+import com.mongodb.client.model.UpdateOptions;
+import com.sap.sse.replication.ReplicaDescriptor;
+import com.sap.sse.replication.persistence.MongoObjectFactory;
+
+public class MongoObjectFactoryImpl implements MongoObjectFactory {
+    private final MongoDatabase db;
+    
+    public MongoObjectFactoryImpl(MongoDatabase db) {
+        super();
+        this.db = db;
+    }
+    
+    private MongoCollection<Document> getReplicaDescriptorCollection() {
+        final MongoCollection<Document> result = db.getCollection(CollectionNames.REPLICA_DESCRIPTORS.name());
+        result.createIndex(new Document(FieldNames.REPLICA_ID_AS_STRING.name(), 1));
+        return result;
+    }
+
+    @Override
+    public void storeReplicaDescriptor(ReplicaDescriptor replicaDescriptor) {
+        final Document replicaDescriptorDoc = new Document();
+        replicaDescriptorDoc.put(FieldNames.REPLICA_ID_AS_STRING.name(), replicaDescriptor.getUuid().toString());
+        replicaDescriptorDoc.put(FieldNames.REPLICA_IP_ADDRESS.name(), replicaDescriptor.getIpAddress().getHostAddress());
+        replicaDescriptorDoc.put(FieldNames.REPLICA_REGISTRATION_TIME_MILLIS.name(), replicaDescriptor.getRegistrationTime().asMillis());
+        replicaDescriptorDoc.put(FieldNames.REPLICA_ADDITIONAL_INFORMATION.name(), replicaDescriptor.getAdditionalInformation());
+        replicaDescriptorDoc.put(FieldNames.REPLICA_REPLICABLE_IDS_AS_STRINGS.name(), replicaDescriptor.getReplicableIdsAsStrings());
+        getReplicaDescriptorCollection().withWriteConcern(WriteConcern.ACKNOWLEDGED).replaceOne(replicaDescriptorDoc,
+                replicaDescriptorDoc, new UpdateOptions().upsert(true));
+    }
+    
+    @Override
+    public void removeReplicaDescriptor(ReplicaDescriptor replicaDescriptor) {
+        if (replicaDescriptor != null) {
+            final Document replicaDescriptorDoc = new Document(FieldNames.REPLICA_ID_AS_STRING.name(), replicaDescriptor.getUuid().toString());
+            getReplicaDescriptorCollection().withWriteConcern(WriteConcern.ACKNOWLEDGED).deleteOne(replicaDescriptorDoc);
+        }
+    }
+}

java/com.sap.sse.replication.persistence/src/com/sap/sse/replication/persistence/impl/PersistenceFactoryImpl.java

@@ -0,0 +1,29 @@
+package com.sap.sse.replication.persistence.impl;
+
+import com.sap.sse.replication.persistence.PersistenceFactory;
+import com.sap.sse.mongodb.MongoDBConfiguration;
+import com.sap.sse.mongodb.MongoDBService;
+import com.sap.sse.replication.persistence.DomainObjectFactory;
+import com.sap.sse.replication.persistence.MongoObjectFactory;
+
+public class PersistenceFactoryImpl implements PersistenceFactory {
+    @Override
+    public DomainObjectFactory getDomainObjectFactory(MongoDBService mongoDbService) {
+        return new DomainObjectFactoryImpl(mongoDbService.getDB());
+    }
+
+    @Override
+    public MongoObjectFactory getMongoObjectFactory(MongoDBService mongoDbService) {
+        return new MongoObjectFactoryImpl(mongoDbService.getDB());
+    }
+
+    @Override
+    public DomainObjectFactory getDefaultDomainObjectFactory() {
+        return getDomainObjectFactory(MongoDBConfiguration.getDefaultConfiguration().getService());
+    }
+
+    @Override
+    public MongoObjectFactory getDefaultMongoObjectFactory() {
+        return getMongoObjectFactory(MongoDBConfiguration.getDefaultConfiguration().getService());
+    }
+}

java/com.sap.sse.replication/META-INF/MANIFEST.MF

@@ -24,6 +24,7 @@ Bundle-ClassPath: .
 Comment: The Web-ContextPath is referred to by ReplicationMasterDescriptorImpl.REPLICATION_SERVLET
 Web-ContextPath: /replication
 Import-Package: com.rabbitmq.client;version="2.8.4",
+ com.sap.sse.replication.persistence,
  javax.servlet;version="2.6.0",
  javax.servlet.http;version="2.6.0",
  org.osgi.framework;version="1.6.0",

java/com.sap.sse.replication/src/com/sap/sse/replication/impl/Activator.java

@@ -4,15 +4,18 @@ import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Optional;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
 import org.osgi.framework.BundleActivator;
 import org.osgi.framework.BundleContext;
 
+import com.sap.sse.replication.ReplicaDescriptor;
 import com.sap.sse.replication.Replicable;
 import com.sap.sse.replication.ReplicablesProvider;
 import com.sap.sse.replication.ReplicationService;
+import com.sap.sse.replication.persistence.PersistenceFactory;
 import com.sap.sse.security.util.RemoteServerUtil;
 
 /**
@@ -80,6 +83,7 @@ public class Activator implements BundleActivator {
     public static final String ENV_VAR_NAME_REPLICATION_PORT = "REPLICATION_PORT";
     
     public static final String PROPERTY_NAME_REPLICATE_ON_START = "replicate.on.start";
+    public static final String PROPERTY_NAME_RESTORE_REPLICAS = "replicate.restore.replicas";
     public static final String PROPERTY_NAME_REPLICATE_MASTER_SERVLET_HOST = "replicate.master.servlet.host";
     public static final String PROPERTY_NAME_REPLICATE_MASTER_SERVLET_PORT = "replicate.master.servlet.port";
     public static final String PROPERTY_NAME_REPLICATE_MASTER_QUEUE_HOST = "replicate.master.queue.host";
@@ -127,11 +131,18 @@ public class Activator implements BundleActivator {
         } catch (NumberFormatException nfe) {
             logger.severe("Couldn't parse the replication port specification \""+exchangePortAsString+"\". Using default.");
         }
-        replicationInstancesManager = new ReplicationInstancesManager();
+        final Iterable<ReplicaDescriptor> replicasToAssumeConnectedToThisMaster;
+        if (Boolean.valueOf(System.getProperty(PROPERTY_NAME_RESTORE_REPLICAS, "false"))) {
+            replicasToAssumeConnectedToThisMaster = PersistenceFactory.INSTANCE.getDefaultDomainObjectFactory().loadReplicaDescriptors();
+        } else {
+            replicasToAssumeConnectedToThisMaster = null;
+        }
+        replicationInstancesManager = new ReplicationInstancesManager(replicasToAssumeConnectedToThisMaster);
         final OSGiReplicableTracker replicablesProvider = new OSGiReplicableTracker(bundleContext);
         serverReplicationMasterService = new ReplicationServiceImpl(
+                Optional.of(PersistenceFactory.INSTANCE.getDefaultMongoObjectFactory()),
                 exchangeName, exchangeHost, exchangePort, replicationInstancesManager, replicablesProvider);
-        String replicateOnStart = System.getProperty(PROPERTY_NAME_REPLICATE_ON_START);
+        final String replicateOnStart = System.getProperty(PROPERTY_NAME_REPLICATE_ON_START);
         final boolean autoReplicationRequested = replicateOnStart != null && !replicateOnStart.isEmpty();
         if (autoReplicationRequested) {
             // set before registering the service; clients discovering it therefore cannot accidentally

java/com.sap.sse.replication/src/com/sap/sse/replication/impl/ReplicationInstancesManager.java

@@ -56,13 +56,22 @@ public class ReplicationInstancesManager {
     private ReplicationMasterDescriptor replicationMasterDescriptor;
 
     public ReplicationInstancesManager() {
+        this(/* replicas to assume connected to this master */ null);
+    }
+    
+    public ReplicationInstancesManager(Iterable<ReplicaDescriptor> replicasToAssumeConnectedToThisMaster) {
         replicaDescriptors = new HashMap<>();
         replicationCounts = new HashMap<ReplicaDescriptor, Map<Class<? extends OperationWithResult<?, ?>>,Integer>>();
         totalMessageCount = new HashMap<>();
         totalNumberOfOperations = new HashMap<>();
         totalQueueMessagesRawSizeInBytes = new HashMap<ReplicaDescriptor, Long>();
+        if (replicasToAssumeConnectedToThisMaster != null) {
+            for (final ReplicaDescriptor replicaDescriptor : replicasToAssumeConnectedToThisMaster) {
+                registerReplica(replicaDescriptor);
+            }
+        }
     }
-    
+
     /**
      * Tells if at least one replica is currently registered.
      * 
@@ -76,6 +85,10 @@ public class ReplicationInstancesManager {
     public Iterable<ReplicaDescriptor> getReplicaDescriptors() {
         return Collections.unmodifiableCollection(replicaDescriptors.values());
     }
+    
+    public ReplicaDescriptor getReplicaDescriptor(UUID replicaId) {
+        return replicaDescriptors.get(replicaId);
+    }
 
     public ReplicationMasterDescriptor getReplicationMasterDescriptor() {
         return replicationMasterDescriptor;

java/com.sap.sse.replication/src/com/sap/sse/replication/impl/ReplicationServiceImpl.java

@@ -17,6 +17,7 @@ import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Random;
 import java.util.Timer;
 import java.util.TimerTask;
@@ -47,6 +48,7 @@ import com.sap.sse.replication.ReplicationMasterDescriptor;
 import com.sap.sse.replication.ReplicationReceiver;
 import com.sap.sse.replication.ReplicationService;
 import com.sap.sse.replication.ReplicationStatus;
+import com.sap.sse.replication.persistence.MongoObjectFactory;
 import com.sap.sse.util.HttpUrlConnectionHelper;
 
 import net.jpountz.lz4.LZ4BlockInputStream;
@@ -96,7 +98,7 @@ public class ReplicationServiceImpl implements ReplicationService, OperationsToM
     /**
      * The UUIDs with which this replica is registered by the master identified by the corresponding key
      */
-    private final Map<ReplicationMasterDescriptor, String> replicaUUIDs;
+    private final ConcurrentMap<ReplicationMasterDescriptor, String> replicaUUIDs;
 
     /**
      * Channel used by a master server to publish replication operations; <code>null</code> in servers that don't have
@@ -232,6 +234,13 @@ public class ReplicationServiceImpl implements ReplicationService, OperationsToM
      */
     private boolean replicationStarting;
 
+    /**
+     * An optional link to a persistence layer that allows this service to record replicas
+     * registered at this server, so that optionally during a re-start those replica
+     * links can be re-established.
+     */
+    private final Optional<MongoObjectFactory> mongoObjectFactory;
+
     private static class InitialLoadRequest {
         private final Channel channelForInitialLoad;
         private final Iterable<Replicable<?, ?>> replicables;
@@ -296,12 +305,19 @@ public class ReplicationServiceImpl implements ReplicationService, OperationsToM
     public ReplicationServiceImpl(String exchangeName, String exchangeHost, int exchangePort,
             final ReplicationInstancesManager replicationInstancesManager, ReplicablesProvider replicablesProvider)
             throws IOException {
+        this(/* defaultMongoObjectFactory */ Optional.empty(), exchangeName, exchangeHost, exchangePort, replicationInstancesManager, replicablesProvider);
+    }
+
+    public ReplicationServiceImpl(Optional<MongoObjectFactory> optionalMongoObjectFactory, String exchangeName,
+            String exchangeHost, int exchangePort, ReplicationInstancesManager replicationInstancesManager,
+            ReplicablesProvider replicablesProvider) {
+        this.mongoObjectFactory = optionalMongoObjectFactory;
         timer = new Timer("ReplicationServiceImpl timer for delayed task sending", /* isDaemon */ true);
         unsentOperationsSenderJob = new UnsentOperationsSenderJob();
         executionListenersByReplicableIdAsString = new ConcurrentHashMap<>();
         initialLoadChannels = new ConcurrentHashMap<>();
         this.replicationInstancesManager = replicationInstancesManager;
-        replicaUUIDs = new HashMap<ReplicationMasterDescriptor, String>();
+        replicaUUIDs = new ConcurrentHashMap<ReplicationMasterDescriptor, String>();
         this.replicablesProvider = replicablesProvider;
         this.exchangeName = exchangeName;
         this.exchangeHost = exchangeHost;
@@ -382,10 +398,19 @@ public class ReplicationServiceImpl implements ReplicationService, OperationsToM
                 }
             }
             replicationInstancesManager.registerReplica(replica);
+            recordRegisteredReplicaPersistently(replica);
         }
         logger.info("Registered replica " + replica);
     }
 
+    private void recordRegisteredReplicaPersistently(ReplicaDescriptor replica) {
+        mongoObjectFactory.ifPresent(mof->mof.storeReplicaDescriptor(replica));
+    }
+
+    private void removeRegisteredReplicaPersistently(ReplicaDescriptor replica) {
+        mongoObjectFactory.ifPresent(mof->mof.removeReplicaDescriptor(replica));
+    }
+
     private void addAsListenerToReplicables(String[] replicableIdsAsStringForReplicablesToReplicate) {
         for (final String replicableIdAsStringForReplicableToReplicate : replicableIdsAsStringForReplicablesToReplicate) {
             Replicable<?, ?> replicable = getReplicable(replicableIdAsStringForReplicableToReplicate, /* wait */ true);
@@ -411,6 +436,7 @@ public class ReplicationServiceImpl implements ReplicationService, OperationsToM
         synchronized (replicationInstancesManager) {
             final boolean hadReplicas = replicationInstancesManager.hasReplicas();
             final Iterable<String> oldReplicablesInReplication = replicationInstancesManager.getAllReplicableIdsAtLeastOneReplicaIsReplicating();
+            removeRegisteredReplicaPersistently(replicationInstancesManager.getReplicaDescriptor(replicaUuid));
             final ReplicaDescriptor unregisteredReplica = replicationInstancesManager.unregisterReplica(replicaUuid);
             final Iterable<String> newReplicablesInReplication = replicationInstancesManager.getAllReplicableIdsAtLeastOneReplicaIsReplicating();
             for (final String idAsStringOfReplicableNoReplicaIsInterestedInAnymore : Util.removeAll(newReplicablesInReplication, Util.addAll(oldReplicablesInReplication, new HashSet<>()))) {

java/pom.xml

@@ -101,6 +101,7 @@
         <module>com.sap.sse.security.userstore.mongodb</module>
         <module>com.sap.sse.operationaltransformation</module>
         <module>com.sap.sse.replication</module>
+        <module>com.sap.sse.replication.persistence</module>
         <module>com.sap.sse.replication.interfaces</module>
         <module>com.sap.sse.threadmanager</module>
         <module>com.sap.sse.filestorage</module>