solution for bug 2437: clear the replica's state before registering for operation updates which needs to happen before requesting the initial load

java/com.sap.sailing.domain.shared.android/src/com/sap/sailing/domain/base/impl/SharedDomainFactoryImpl.java

@@ -12,6 +12,8 @@ import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.logging.Level;
+import java.util.logging.Logger;
 
 import com.sap.sailing.domain.base.BoatClass;
 import com.sap.sailing.domain.base.Competitor;
@@ -34,6 +36,7 @@ import com.sap.sailing.domain.common.WithID;
 import com.sap.sailing.domain.common.configuration.DeviceConfigurationMatcherType;
 
 public class SharedDomainFactoryImpl implements SharedDomainFactory {
+    private static final Logger logger = Logger.getLogger(SharedDomainFactoryImpl.class.getName());
     
     /**
      * Ensure that the <em>same</em> string is used as key that is also used to set the {@link Nationality}
@@ -321,6 +324,9 @@ public class SharedDomainFactoryImpl implements SharedDomainFactory {
 
     @Override
     public Competitor getOrCreateCompetitor(Serializable competitorId, String name, Color displayColor, DynamicTeam team, DynamicBoat boat) {
+        if (logger.isLoggable(Level.FINEST)) {
+            logger.log(Level.FINEST, "getting or creating competitor "+name+" with ID "+competitorId+" in domain factory "+this);
+        }
         return getCompetitorStore().getOrCreateCompetitor(competitorId, name, displayColor, team, boat);
     }
 

java/com.sap.sailing.domain.shared.android/src/com/sap/sailing/domain/base/impl/TransientCompetitorStoreImpl.java

@@ -10,6 +10,8 @@ import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
 import java.util.WeakHashMap;
+import java.util.logging.Level;
+import java.util.logging.Logger;
 
 import com.sap.sailing.domain.base.Competitor;
 import com.sap.sailing.domain.base.CompetitorStore;
@@ -23,6 +25,7 @@ import com.sap.sailing.util.impl.LockUtil;
 import com.sap.sailing.util.impl.NamedReentrantReadWriteLock;
 
 public class TransientCompetitorStoreImpl implements CompetitorStore, Serializable {
+    private static final Logger logger = Logger.getLogger(TransientCompetitorStoreImpl.class.getName());
     private static final long serialVersionUID = -4198298775476586931L;
     private final Map<Serializable, Competitor> competitorCache;
     private final Map<String, Competitor> competitorsByIdAsString;
@@ -67,6 +70,9 @@ public class TransientCompetitorStoreImpl implements CompetitorStore, Serializab
     private Competitor createCompetitor(Serializable id, String name, Color displayColor, DynamicTeam team, DynamicBoat boat) {
         Competitor result = new CompetitorImpl(id, name, displayColor, team, boat);
         addNewCompetitor(id, result);
+        if (logger.isLoggable(Level.FINEST)) {
+            logger.log(Level.FINEST, "Created competitor "+name+" with ID "+id, new Exception("Here is where it happened"));
+        }
         return result;
     }
 
@@ -151,6 +157,9 @@ public class TransientCompetitorStoreImpl implements CompetitorStore, Serializab
             competitorCache.clear();
             competitorsByIdAsString.clear();
             competitorsToUpdateDuringGetOrCreate.clear();
+            if (logger.isLoggable(Level.FINEST)) {
+                logger.log(Level.FINEST, "Clearing competitor store "+this, new Exception("here is where it happened"));
+            }
         } finally {
             LockUtil.unlockAfterWrite(lock);
         }
@@ -170,6 +179,7 @@ public class TransientCompetitorStoreImpl implements CompetitorStore, Serializab
     public void removeCompetitor(Competitor competitor) {
         LockUtil.lockForWrite(lock);
         try {
+            logger.fine("removing competitor "+competitor+" from competitor store "+this);
             competitorCache.remove(competitor.getId());
             competitorsByIdAsString.remove(competitor.getId().toString());  
             weakCompetitorDTOCache.remove(competitor);

java/com.sap.sailing.server.replication.test/src/com/sap/sailing/server/replication/test/AbstractServerReplicationTest.java

@@ -290,6 +290,7 @@ public abstract class AbstractServerReplicationTest {
                 }
             }.start();
             ObjectInputStream dis = resolveAgainst.createObjectInputStreamResolvingAgainstThisFactory(pis);
+            getRacingEventService().clearReplicaState();
             getRacingEventService().initiallyFillFrom(dis);
             dis.close();
         }

java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/ReplicationMasterDescriptorImpl.java

@@ -145,7 +145,7 @@ public class ReplicationMasterDescriptorImpl implements ReplicationMasterDescrip
                 // make sure to remove queue in order to avoid any exchanges filling it with messages
                 consumer.getChannel().queueUnbind(queueName, exchangeName, "");
                 consumer.getChannel().queueDelete(queueName);
-                consumer.getChannel().getConnection().close(1);
+                consumer.getChannel().getConnection().close(/* timeout in millis */ 1000);
             }
         } catch (Exception ex) {
             // ignore any exception during abort. close can yield a broad

java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/ReplicationServiceImpl.java

@@ -428,13 +428,15 @@ public class ReplicationServiceImpl implements ReplicationService, OperationExec
         replicator = new Replicator(master, this, /* startSuspended */ true, consumer, getRacingEventService().getBaseDomainFactory());
         // start receiving messages already now, but start in suspended mode
         replicatorThread = new Thread(replicator, "Replicator receiving from "+master.getMessagingHostname()+"/"+master.getExchangeName());
+        final RacingEventService racingEventService = getRacingEventService();
+        // clear racingEventService state here, before starting to receive and de-serialized operations which builds up new state, e.g., in competitor store
+        racingEventService.clearReplicaState(); // see also bug 2437
         replicatorThread.start();
         logger.info("Started replicator thread");
         InputStream is = initialLoadURL.openStream();
         final String queueName = new BufferedReader(new InputStreamReader(is)).readLine();
         try {
             RabbitInputStreamProvider rabbitInputStreamProvider = new RabbitInputStreamProvider(master.createChannel(), queueName);
-            final RacingEventService racingEventService = getRacingEventService();
             ObjectInputStream ois = racingEventService.getBaseDomainFactory()
                     .createObjectInputStreamResolvingAgainstThisFactory(
                             new GZIPInputStream(rabbitInputStreamProvider.getInputStream()));

java/com.sap.sailing.server/src/com/sap/sailing/server/RacingEventService.java

@@ -311,6 +311,17 @@ public interface RacingEventService extends TrackedRegattaRegistry, RegattaFetch
      */
     void serializeForInitialReplication(ObjectOutputStream oos) throws IOException;
 
+    /**
+     * Before {@link #initiallyFillFrom(ObjectInputStream) initially loading a replica's state from a master instance},
+     * the replica's old state needs to be "detached". This method clears all top-level data structures and stops all
+     * tracking currently going on.<p>
+     * 
+     * The reason this operation needs to be callable separate from {@link #initiallyFillFrom(ObjectInputStream)} is that
+     * it needs to happen before subscribing to the operation feed received from the master instance through the message bus
+     * which in turn needs to happen before receiving the initial load.
+     */
+    void clearReplicaState() throws MalformedURLException, IOException, InterruptedException;
+
     /**
      * Dual, reading operation for {@link #serializeForInitialReplication(ObjectOutputStream)}. In other words, when
      * this operation returns, this service instance is in a state "equivalent" to that of the service instance that
@@ -324,8 +335,8 @@ public interface RacingEventService extends TrackedRegattaRegistry, RegattaFetch
      * to the default leaderboard and for automatic linking to leaderboard columns. It is assumed that no explicit
      * replication of these operations will happen based on the changes performed on the replication master.<p>
      * 
-     * <b>Caution:</b> All relevant contents of this service instance will be replaced by the stream contents.
-     * @throws InterruptedException 
+     * <b>Caution:</b> All relevant contents of this service instance needs to be cleared before by a call to
+     * {@link #clearReplicaState()}. It will be replaced by the stream contents.
      */
     void initiallyFillFrom(ObjectInputStream ois) throws IOException, ClassNotFoundException, InterruptedException;
 

java/com.sap.sailing.server/src/com/sap/sailing/server/impl/RacingEventServiceImpl.java

@@ -2176,7 +2176,7 @@ public class RacingEventServiceImpl implements RacingEventServiceWithTestSupport
         logger.info(logoutput.toString());
     }
 
-    @SuppressWarnings("unchecked")
+    @SuppressWarnings("unchecked") // all the casts of ois.readObject()'s return value to Map<..., ...>
     // the type-parameters in the casts of the de-serialized collection objects can't be checked
     @Override
     public void initiallyFillFrom(ObjectInputStream ois) throws IOException, ClassNotFoundException,
@@ -2188,55 +2188,7 @@ public class RacingEventServiceImpl implements RacingEventServiceWithTestSupport
             // de-serialization; this will cause all classes to be visible that this bundle
             // (com.sap.sailing.server) can see
             Thread.currentThread().setContextClassLoader(getClass().getClassLoader());
-            LockUtil.lockForWrite(regattasByNameLock);
-            try {
-                regattasByName.clear();
-            } finally {
-                LockUtil.unlockAfterWrite(regattasByNameLock);
-            }
-            regattasObservedForDefaultLeaderboard.clear();
-
-            if (raceTrackersByRegatta != null && !raceTrackersByRegatta.isEmpty()) {
-                for (DynamicTrackedRegatta regatta : regattaTrackingCache.values()) {
-                    for (RaceTracker tracker : raceTrackersByRegatta.get(regatta)) {
-                        tracker.stop();
-                    }
-                }
-            }
-
-            logger.info("Clearing all data structures...");
-            LockUtil.lockForWrite(regattaTrackingCacheLock);
-            try {
-                regattaTrackingCache.clear();
-            } finally {
-                LockUtil.unlockAfterWrite(regattaTrackingCacheLock);
-            }
-            LockUtil.lockForWrite(raceTrackersByRegattaLock);
-            try {
-                raceTrackersByRegatta.clear();
-            } finally {
-                LockUtil.unlockAfterWrite(raceTrackersByRegattaLock);
-            }
-            LockUtil.lockForWrite(leaderboardGroupsByNameLock);
-            try {
-                leaderboardGroupsByName.clear();
-                leaderboardGroupsByID.clear();
-            } finally {
-                LockUtil.unlockAfterWrite(leaderboardGroupsByNameLock);
-            }
-            LockUtil.lockForWrite(leaderboardsByNameLock);
-            try {
-                leaderboardsByName.clear();
-            } finally {
-                LockUtil.unlockAfterWrite(leaderboardsByNameLock);
-            }
-            eventsById.clear();
-            mediaLibrary.clear();
-            competitorStore.clear();
-            remoteSailingServerSet.clear();
-
             StringBuffer logoutput = new StringBuffer();
-
             logger.info("Reading all events...");
             eventsById.putAll((Map<Serializable, Event>) ois.readObject());
             logoutput.append("\nReceived " + eventsById.size() + " NEW events\n");
@@ -2328,6 +2280,55 @@ public class RacingEventServiceImpl implements RacingEventServiceWithTestSupport
         }
     }
 
+    @Override
+    public void clearReplicaState() throws MalformedURLException, IOException, InterruptedException {
+        logger.info("Clearing all data structures...");
+        LockUtil.lockForWrite(regattasByNameLock);
+        try {
+            regattasByName.clear();
+        } finally {
+            LockUtil.unlockAfterWrite(regattasByNameLock);
+        }
+        regattasObservedForDefaultLeaderboard.clear();
+
+        if (raceTrackersByRegatta != null && !raceTrackersByRegatta.isEmpty()) {
+            for (DynamicTrackedRegatta regatta : regattaTrackingCache.values()) {
+                for (RaceTracker tracker : raceTrackersByRegatta.get(regatta)) {
+                    tracker.stop();
+                }
+            }
+        }
+        LockUtil.lockForWrite(regattaTrackingCacheLock);
+        try {
+            regattaTrackingCache.clear();
+        } finally {
+            LockUtil.unlockAfterWrite(regattaTrackingCacheLock);
+        }
+        LockUtil.lockForWrite(raceTrackersByRegattaLock);
+        try {
+            raceTrackersByRegatta.clear();
+        } finally {
+            LockUtil.unlockAfterWrite(raceTrackersByRegattaLock);
+        }
+        LockUtil.lockForWrite(leaderboardGroupsByNameLock);
+        try {
+            leaderboardGroupsByName.clear();
+            leaderboardGroupsByID.clear();
+        } finally {
+            LockUtil.unlockAfterWrite(leaderboardGroupsByNameLock);
+        }
+        LockUtil.lockForWrite(leaderboardsByNameLock);
+        try {
+            leaderboardsByName.clear();
+        } finally {
+            LockUtil.unlockAfterWrite(leaderboardsByNameLock);
+        }
+        eventsById.clear();
+        mediaLibrary.clear();
+        competitorStore.clear();
+        remoteSailingServerSet.clear();
+    }
+
     // Used for TESTING only
     @Override
     public Event addEvent(String eventName, String eventDescription, TimePoint startDate, TimePoint endDate,

java/target/configuration/logging_debug.properties

@@ -17,6 +17,9 @@ java.util.logging.ConsoleHandler.formatter = java.util.logging.SimpleFormatter
 
 # Overwrite the log leves for specific loggers here
 com.sap.sailing.level = FINE
+# log at level FINEST for TransientCompetitorStoreIMpl
+#com.sap.sailing.domain.base.impl.TransientCompetitorStoreImpl.level = FINEST
+#com.sap.sailing.domain.base.impl.SharedDomainFactoryImpl.level = FINEST
 org.eclipse.jetty.level = INFO
 org.apache.activemq.level = INFO