Merge branch 'master' into ubilabs--rcapp--develop

java/com.sap.sailing.dashboards.gwt/src/main/java/com/sap/sailing/dashboards/gwt/client/EventLogo.java

@@ -21,7 +21,6 @@ public class EventLogo extends Image {
 
     public static EventLogo getEventLogoFromEventId(SailingServiceAsync sailingServiceAsync, String eventId) {
         final EventLogo eventLogo = new EventLogo();
-        eventLogo.getElement().setAttribute("src", "someURL");
         try {
             final UUID eventUUID = UUID.fromString(eventId);
             sailingServiceAsync.getEventById(eventUUID, true, new AsyncCallback<EventDTO>() {
@@ -29,17 +28,11 @@ public class EventLogo extends Image {
                 public void onSuccess(final EventDTO event) {
                     logger.log(Level.INFO, "GOT EVENT: ");
                     final String logoUrl = event.getLogoImageURL();
-//                    Scheduler.get().scheduleDeferred(new ScheduledCommand() {
-//
-//                        @Override
-//                        public void execute() {
                             if (logoUrl != null) {
                                 eventLogo.getElement().setAttribute("src", logoUrl);
                             } else {
                                 eventLogo.getElement().getStyle().setVisibility(Visibility.HIDDEN);
                             }
-//                        }
-//                    });
                 }
 
                 @Override

java/com.sap.sailing.dashboards.gwt/src/main/java/com/sap/sailing/dashboards/gwt/client/RibDashboardPanel.java

@@ -91,13 +91,7 @@ public class RibDashboardPanel extends Composite implements RibDashboardDataRetr
         windcharthint.getElement().setInnerText(stringConstants.dashboardWindChartHint());
         windloadinghintleft.setInnerText(stringConstants.dashboardWindBotLoading());
         windloadinghintright.setInnerText(stringConstants.dashboardWindBotLoading());
-    
+        loadAndDisplayEventLogo(sailingServiceAsync);
-        String eventId = Window.Location.getParameter(EVENT_ID_PARAMETER);
-        EventLogo eventLogo = EventLogo.getEventLogoFromEventId(sailingServiceAsync, eventId);
-        if(eventLogo != null){
-            Document.get().getBody().appendChild(eventLogo.getElement());
-        }
-        
     }
 
     private void initLogos() {
@@ -107,6 +101,16 @@ public class RibDashboardPanel extends Composite implements RibDashboardDataRetr
         Document.get().getBody().appendChild(logo.getElement());
     }
     
+    private void loadAndDisplayEventLogo(SailingServiceAsync sailingServiceAsync){
+        String eventId = Window.Location.getParameter(EVENT_ID_PARAMETER);
+        if (eventId != null) {
+            EventLogo eventLogo = EventLogo.getEventLogoFromEventId(sailingServiceAsync, eventId);
+            if (eventLogo != null) {
+                Document.get().getBody().appendChild(eventLogo.getElement());
+            }
+        }
+    }
+
     private void initWindBotComponents(List<String> windBotIDs,
             WindBotDataRetrieverProvider windBotDataRetrieverProvider) {
         int windBotAddedCounterTOConsiderForPanels = 0;

java/com.sap.sailing.dashboards.gwt/src/main/java/com/sap/sailing/dashboards/gwt/client/startanalysis/StartlineAnalysisComponent.java

@@ -240,7 +240,7 @@ public class StartlineAnalysisComponent extends Composite implements HasWidgets
         ribDashboardServiceAsync.getCompetitorsInLeaderboard(leaderboardName, new AsyncCallback<List<CompetitorDTO>>() {
             @Override
             public void onSuccess(List<CompetitorDTO> result) {
-                if (!competitorSelectionPopup.isShown())
+                if (result != null && !competitorSelectionPopup.isShown())
                     competitorSelectionPopup.show(result);
             }
 

java/com.sap.sailing.dashboards.gwt/src/main/java/com/sap/sailing/dashboards/gwt/server/RibDashboardServiceImpl.java

@@ -236,13 +236,16 @@ public class RibDashboardServiceImpl extends RemoteServiceServlet implements Rib
     public List<StartAnalysisDTO> getStartAnalysisListForCompetitorIDAndLeaderboardName(String competitorIdAsString,
             String leaderboardName) {
         List<StartAnalysisDTO> startAnalysisDTOs = new ArrayList<StartAnalysisDTO>();
-        Competitor competitor = baseDomainFactory.getCompetitorStore().getExistingCompetitorByIdAsString(
+        if (leaderboardName != null) {
-                competitorIdAsString);
+            Competitor competitor = baseDomainFactory.getCompetitorStore().getExistingCompetitorByIdAsString(
-        List<TrackedRace> trackedRacesForLeaderBoardName = getTrackedRacesFromLeaderboard(leaderboardName);
+                    competitorIdAsString);
-        for (TrackedRace trackedRace : trackedRacesForLeaderBoardName) {
+            List<TrackedRace> trackedRacesForLeaderBoardName = getTrackedRacesFromLeaderboard(leaderboardName);
-            StartAnalysisDTO startAnalysisDTO = startAnalysisCreationController.checkStartAnalysisForCompetitorInTrackedRace(competitor, trackedRace);
+            for (TrackedRace trackedRace : trackedRacesForLeaderBoardName) {
-            if (startAnalysisDTO != null)
+                StartAnalysisDTO startAnalysisDTO = startAnalysisCreationController
-                startAnalysisDTOs.add(startAnalysisDTO);
+                        .checkStartAnalysisForCompetitorInTrackedRace(competitor, trackedRace);
+                if (startAnalysisDTO != null)
+                    startAnalysisDTOs.add(startAnalysisDTO);
+            }
         }
         return startAnalysisDTOs;
     }
@@ -250,8 +253,12 @@ public class RibDashboardServiceImpl extends RemoteServiceServlet implements Rib
     @Override
     public List<CompetitorDTO> getCompetitorsInLeaderboard(String leaderboardName) {
         logger.log(Level.INFO, "getCompetitorsInLeaderboard(...) Request.");
+        if(leaderboardName != null){
         Leaderboard lb = getRacingEventService().getLeaderboardByName(leaderboardName);
         return baseDomainFactory.getCompetitorDTOList(lb.getCompetitorsFromBestToWorst(new MillisecondsTimePoint(
                 new Date())));
+        }else{
+            return null;
+        }
     }
 }

java/com.sap.sailing.domain/src/com/sap/sailing/domain/tracking/RaceTracker.java

@@ -42,7 +42,11 @@ public interface RaceTracker {
 
     /**
      * Stops tracking the races.
-     * @param preemptive TODO
+     * 
+     * @param preemptive
+     *            if <code>false</code>, the tracker will continue to process data already received but will stop
+     *            receiving new data. If <code>true</code>, the tracker will stop processing data immediately, ignoring
+     *            (dropping) all data already received but not yet processed.
      */
     void stop(boolean preemptive) throws MalformedURLException, IOException, InterruptedException;
 

java/com.sap.sailing.domain/src/com/sap/sailing/domain/tracking/impl/TrackedRaceImpl.java

@@ -257,7 +257,7 @@ public abstract class TrackedRaceImpl extends TrackedRaceWithWindEssentials impl
      * these race states can be garbage-collected when the race log is no longer attached. The race states are created
      * lazily, synchronizing on this weak hash map.
      */
-    protected final transient WeakHashMap<RaceLog, ReadonlyRaceState> raceStates;
+    protected transient WeakHashMap<RaceLog, ReadonlyRaceState> raceStates;
 
     protected transient ConcurrentHashMap<Serializable, RegattaLog> attachedRegattaLogs;
 
@@ -494,6 +494,7 @@ public abstract class TrackedRaceImpl extends TrackedRaceWithWindEssentials impl
      */
     private void readObject(ObjectInputStream ois) throws ClassNotFoundException, IOException, PatchFailedException {
         ois.defaultReadObject();
+        raceStates = new WeakHashMap<>();
         attachedRaceLogs = new ConcurrentHashMap<>();
         markPassingsTimes = new ArrayList<com.sap.sse.common.Util.Pair<Waypoint, com.sap.sse.common.Util.Pair<TimePoint, TimePoint>>>();
         // The short time wind cache needs to be there before operations such as maneuver recalculation try to access it
@@ -3155,26 +3156,30 @@ public abstract class TrackedRaceImpl extends TrackedRaceWithWindEssentials impl
             if (!legs.isEmpty()) {
                 if (allMarksHavePositions && numberOfMarks == 2) {
                     final TrackedLeg legDeterminingDirection = getLegDeterminingDirectionInWhichToPassWaypoint(waypoint);
-                    Distance crossTrackErrorOfMark0OnLineFromMark1ToNextWaypoint = markPositions.get(0)
+                    if (legDeterminingDirection == null) {
-                            .crossTrackError(markPositions.get(1), legDeterminingDirection.getLegBearing(timePoint));
+                        result = null;
-                    final Position portMarkPositionWhileApproachingLine;
-                    final Position starboardMarkPositionWhileApproachingLine;
-                    final Mark starboardMarkWhileApproachingLine;
-                    final Mark portMarkWhileApproachingLine;
-                    if (crossTrackErrorOfMark0OnLineFromMark1ToNextWaypoint.getMeters() < 0) {
-                        portMarkWhileApproachingLine = Util.get(waypoint.getMarks(), 0);
-                        portMarkPositionWhileApproachingLine = markPositions.get(0);
-                        starboardMarkWhileApproachingLine = Util.get(waypoint.getMarks(), 1);
-                        starboardMarkPositionWhileApproachingLine = markPositions.get(1);
                     } else {
-                        portMarkWhileApproachingLine = Util.get(waypoint.getMarks(), 1);
+                        Distance crossTrackErrorOfMark0OnLineFromMark1ToNextWaypoint = markPositions.get(0)
-                        portMarkPositionWhileApproachingLine = markPositions.get(1);
+                                .crossTrackError(markPositions.get(1), legDeterminingDirection.getLegBearing(timePoint));
-                        starboardMarkWhileApproachingLine = Util.get(waypoint.getMarks(), 0);
+                        final Position portMarkPositionWhileApproachingLine;
-                        starboardMarkPositionWhileApproachingLine = markPositions.get(0);
+                        final Position starboardMarkPositionWhileApproachingLine;
+                        final Mark starboardMarkWhileApproachingLine;
+                        final Mark portMarkWhileApproachingLine;
+                        if (crossTrackErrorOfMark0OnLineFromMark1ToNextWaypoint.getMeters() < 0) {
+                            portMarkWhileApproachingLine = Util.get(waypoint.getMarks(), 0);
+                            portMarkPositionWhileApproachingLine = markPositions.get(0);
+                            starboardMarkWhileApproachingLine = Util.get(waypoint.getMarks(), 1);
+                            starboardMarkPositionWhileApproachingLine = markPositions.get(1);
+                        } else {
+                            portMarkWhileApproachingLine = Util.get(waypoint.getMarks(), 1);
+                            portMarkPositionWhileApproachingLine = markPositions.get(1);
+                            starboardMarkWhileApproachingLine = Util.get(waypoint.getMarks(), 0);
+                            starboardMarkPositionWhileApproachingLine = markPositions.get(0);
+                        }
+                        result = new LineMarksWithPositions(portMarkPositionWhileApproachingLine,
+                                starboardMarkPositionWhileApproachingLine, starboardMarkWhileApproachingLine,
+                                portMarkWhileApproachingLine);
                     }
-                    result = new LineMarksWithPositions(portMarkPositionWhileApproachingLine,
-                            starboardMarkPositionWhileApproachingLine, starboardMarkWhileApproachingLine,
-                            portMarkWhileApproachingLine);
                 } else {
                     result = null; // either the position(s) or one or more marks is/are unknown, or the waypoint is not a two-mark waypoint
                 }

java/com.sap.sailing.feature/feature.xml

@@ -24,10 +24,10 @@ Computes leaderboard information for sailing races and offers RESTful APIs to al
       <import plugin="com.googlecode.java-diff-utils" version="1.3.0" match="greaterOrEqual"/>
       <import plugin="org.eclipse.osgi" version="3.10.1" match="greaterOrEqual"/>
       <import plugin="javax.servlet" version="3.1.0" match="greaterOrEqual"/>
-      <import plugin="com.mongodb.driver" version="2.6.2" match="greaterOrEqual"/>
+      <import plugin="org.mongodb.mongo-java-driver" version="2.13.0" match="greaterOrEqual"/>
-      <import plugin="com.google.gwt.osgi" version="2.6.1" match="greaterOrEqual"/>
+      <import plugin="com.google.gwt.osgi" version="2.7.0" match="greaterOrEqual"/>
       <import plugin="org.json.simple" version="1.1.0" match="greaterOrEqual"/>
-      <import plugin="org.moxieapps.gwt.highcharts" version="1.1.4" match="greaterOrEqual"/>
+      <import plugin="org.moxieapps.gwt.highcharts" version="1.6.0" match="greaterOrEqual"/>
       <import plugin="org.apache.geronimo.specs.geronimo-jms_1.1_spec"/>
       <import plugin="com.rabbitmq.client"/>
       <import plugin="org.apache.geronimo.specs.geronimo-j2ee-management_1.1_spec"/>
@@ -339,13 +339,6 @@ Computes leaderboard information for sailing races and offers RESTful APIs to al
          version="0.0.0"
          unpack="false"/>
 
-   <plugin
-         id="com.sap.sse.gateway"
-         download-size="0"
-         install-size="0"
-         version="0.0.0"
-         unpack="false"/>
-
    <plugin
          id="com.sap.sse.gwt.adminconsole"
          download-size="0"

java/com.sap.sailing.gwt.ui/src/main/java/com/sap/sailing/gwt/ui/client/SailingService.java

@@ -140,7 +140,7 @@ public interface SailingService extends RemoteService, FileStorageManagementGwtS
     WindInfoForRaceDTO getAveragedWindInfo(RegattaAndRaceIdentifier raceIdentifier, Date from, long millisecondsStepWidth,
             int numberOfFixes, double latDeg, double lngDeg, Collection<String> windSources) throws NoWindException;
     
-    Boolean getPolarResults(RegattaAndRaceIdentifier raceIdentifier);
+    boolean getPolarResults(RegattaAndRaceIdentifier raceIdentifier);
 
     SimulatorResultsDTO getSimulatorResults(RegattaAndRaceIdentifier raceIdentifier, Date from, Date prevStartTime);
 

java/com.sap.sailing.gwt.ui/src/main/java/com/sap/sailing/gwt/ui/server/SailingServiceImpl.java

@@ -1423,13 +1423,19 @@ public class SailingServiceImpl extends ProxiedRemoteServiceServlet implements S
     }
 
     @Override
-    public Boolean getPolarResults(RegattaAndRaceIdentifier raceIdentifier) {
+    public boolean getPolarResults(RegattaAndRaceIdentifier raceIdentifier) {
+        final boolean result;
         TrackedRace trackedRace = getExistingTrackedRace(raceIdentifier);
-        BoatClass boatClass = trackedRace.getRace().getBoatClass();
+        if (trackedRace == null) {
-        PolarDataService polarData = getService().getPolarDataService();
+            result = false;
-        PolarDiagram polarDiagram = new PolarDiagramGPS(boatClass, polarData);
+        } else {
-        return new Boolean(polarDiagram != null);
+            BoatClass boatClass = trackedRace.getRace().getBoatClass();
-    };
+            PolarDataService polarData = getService().getPolarDataService();
+            PolarDiagram polarDiagram = new PolarDiagramGPS(boatClass, polarData);
+            result = polarDiagram != null;
+        }
+        return result;
+    }
 
     @Override
     public SimulatorResultsDTO getSimulatorResults(RegattaAndRaceIdentifier raceIdentifier, Date from, Date prevStartTime) {

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

@@ -190,7 +190,7 @@ public class InitialLoadReplicationObjectIdentityTest extends AbstractServerRepl
         replicationDescriptorPair.getA().initialLoad();
         replicator.setSuspended(false);
         synchronized (replicator) {
-            while (!replicator.isQueueEmpty()) {
+            while (!replicator.isQueueEmptyOrStopped()) {
                 replicator.wait();
             }
         }

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

@@ -60,7 +60,7 @@ public class PrematureOperationReceiptTest extends AbstractServerReplicationTest
         }
         replicator.setSuspended(false);
         synchronized (replicator) {
-            while (!replicator.isQueueEmpty()) {
+            while (!replicator.isQueueEmptyOrStopped()) {
                 replicator.wait();
             }
         }

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

@@ -129,6 +129,7 @@ public class TrackRaceReplicationTest extends AbstractServerReplicationTest {
 
     @Test
     public void testReassignmentToLeaderboardReplication() throws Exception {
+        logger.info("Running testReassignmentToLeaderboardReplication()");
         final String leaderboardName = "Test Leaderboard";
         Leaderboard masterLeaderboard = master.apply(new CreateFlexibleLeaderboard(leaderboardName, null, new int[0], new LowPoint(), null));
         final String columnName = "R1";
@@ -153,10 +154,12 @@ public class TrackRaceReplicationTest extends AbstractServerReplicationTest {
         RaceColumn column = replicaLeaderboard.getRaceColumnByName(columnName);
         assertNotNull(column);
         assertSame(replicaTrackedRace, column.getTrackedRace(defaultFleet));
+        logger.info("Done running testReassignmentToLeaderboardReplication()");
     }
     
     @Test
     public void testRaceTimeReplication() throws InterruptedException, Exception {
+        logger.info("Running testRaceTimeReplication()");
         startTracking();
         // now wait at least until the start of tracking time has been received on the master copy
         boolean receivedStartAndEndOfTracking = master.getTrackedRace(raceIdentifier).getStartOfTracking() != null &&
@@ -176,21 +179,27 @@ public class TrackRaceReplicationTest extends AbstractServerReplicationTest {
         ((DynamicTrackedRace) masterTrackedRace).setStartTimeReceived(now);
         Thread.sleep(1000);
         assertEquals(now, replicaTrackedRace.getStartOfRace());
+        logger.info("Done running testRaceTimeReplication()");
     }
 
     @Test
     public void testWindSourcesToExcludeReplication() throws InterruptedException, Exception {
-        startTracking();
+        try {
-        Thread.sleep(1000);
+            logger.info("Running testWindSourcesToExcludeReplication()");
-        TrackedRace replicaTrackedRace = replica.getTrackedRace(raceIdentifier);
+            startTracking();
-        assertTrue(Util.isEmpty(masterTrackedRace.getWindSourcesToExclude()));
+            Thread.sleep(1000);
-        assertTrue(Util.isEmpty(replicaTrackedRace.getWindSourcesToExclude()));
+            TrackedRace replicaTrackedRace = replica.getTrackedRace(raceIdentifier);
-        masterTrackedRace.setWindSourcesToExclude(Collections.singleton(new WindSourceImpl(WindSourceType.WEB)));
+            assertTrue(Util.isEmpty(masterTrackedRace.getWindSourcesToExclude()));
-        assertEquals(1, Util.size(masterTrackedRace.getWindSourcesToExclude()));
+            assertTrue(Util.isEmpty(replicaTrackedRace.getWindSourcesToExclude()));
-        assertEquals(WindSourceType.WEB, masterTrackedRace.getWindSourcesToExclude().iterator().next().getType());
+            masterTrackedRace.setWindSourcesToExclude(Collections.singleton(new WindSourceImpl(WindSourceType.WEB)));
-        Thread.sleep(1000);
+            assertEquals(1, Util.size(masterTrackedRace.getWindSourcesToExclude()));
-        assertEquals(1, Util.size(replicaTrackedRace.getWindSourcesToExclude()));
+            assertEquals(WindSourceType.WEB, masterTrackedRace.getWindSourcesToExclude().iterator().next().getType());
-        assertEquals(WindSourceType.WEB, replicaTrackedRace.getWindSourcesToExclude().iterator().next().getType());
+            Thread.sleep(1000);
+            assertEquals(1, Util.size(replicaTrackedRace.getWindSourcesToExclude()));
+            assertEquals(WindSourceType.WEB, replicaTrackedRace.getWindSourcesToExclude().iterator().next().getType());
+        } finally {
+            logger.info("Done running testWindSourcesToExcludeReplication()");
+        }
     }
 
     @After

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

@@ -380,6 +380,8 @@ public class RacingEventServiceImpl implements RacingEventService, ClearStateTes
     
     private final Set<ClassLoader> masterDataClassLoaders = new HashSet<ClassLoader>();
     
+    private final JoinedClassLoader joinedClassLoader;
+
     /**
      * Constructs a {@link DomainFactory base domain factory} that uses this object's {@link #competitorStore competitor
      * store} for competitor management. This base domain factory is then also used for the construction of the
@@ -457,6 +459,7 @@ public class RacingEventServiceImpl implements RacingEventService, ClearStateTes
             TypeBasedServiceFinderFactory serviceFinderFactory) {
         logger.info("Created " + this);
         this.masterDataClassLoaders.add(this.getClass().getClassLoader());
+        joinedClassLoader = new JoinedClassLoader(masterDataClassLoaders);
         this.operationsSentToMasterForReplication = new HashSet<>();
         if (windStore == null) {
             try {
@@ -2291,7 +2294,7 @@ public class RacingEventServiceImpl implements RacingEventService, ClearStateTes
     
     @Override
     public ClassLoader getDeserializationClassLoader() {
-        return getCombinedMasterDataClassLoader();
+        return joinedClassLoader;
     }
 
     @Override

java/com.sap.sse.replication.testsupport/src/com/sap/sse/replication/testsupport/AbstractServerReplicationTestSetUp.java

@@ -179,6 +179,16 @@ public abstract class AbstractServerReplicationTestSetUp<ReplicableInterface ext
              * original exception propagate */
             logger.log(Level.SEVERE, "Exception trying to connect to initial load test servlet to STOP it", ex);
         }
+        final ReplicationReceiver replicaReplicatorReplicator = replicaReplicator.getReplicator();
+        if (replicaReplicatorReplicator != null) {
+            synchronized (replicaReplicatorReplicator) {
+                while (!replicaReplicatorReplicator.isQueueEmptyOrStopped()) {
+                    logger.info("Waiting for replication queue to drain...");
+                    replicaReplicatorReplicator.wait();
+                }
+            }
+        }
+        logger.info("Replication queue has been drained...");
     }
     
     /**
@@ -216,6 +226,11 @@ public abstract class AbstractServerReplicationTestSetUp<ReplicableInterface ext
             return masterDescriptor;
         }
         
+        @Override
+        public ReplicationReceiver getReplicator() {
+            return super.getReplicator();
+        }
+
         private Thread startInitialLoadTransmissionServlet() throws InterruptedException {
             final boolean[] listening = new boolean[] { false };
             Thread initialLoadTestServerThread = new Thread("Replication initial load test server") {
@@ -322,7 +337,7 @@ public abstract class AbstractServerReplicationTestSetUp<ReplicableInterface ext
         
         public void waitUntilQueueIsEmpty() throws InterruptedException, IllegalAccessException {
             synchronized (getReplicator()) {
-                while (!getReplicator().isQueueEmpty()) {
+                while (!getReplicator().isQueueEmptyOrStopped()) {
                     getReplicator().wait();
                 }
             }

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

@@ -7,6 +7,8 @@ import java.net.URL;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.UUID;
+import java.util.logging.Level;
+import java.util.logging.Logger;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 
@@ -19,6 +21,8 @@ import com.sap.sse.replication.Replicable;
 import com.sap.sse.replication.ReplicationMasterDescriptor;
 
 public class ReplicationMasterDescriptorImpl implements ReplicationMasterDescriptor {
+    private static final Logger logger = Logger.getLogger(ReplicationMasterDescriptorImpl.class.getName());
+    
     private static final String REPLICATION_SERVLET = "/replication/replication";
     private final String masterServletHostname;
     private final String exchangeName;
@@ -161,7 +165,7 @@ public class ReplicationMasterDescriptorImpl implements ReplicationMasterDescrip
         } catch (Exception ex) {
             // ignore any exception during abort. close can yield a broad
             // number of exceptions that we don't want to know or to log.
-            ex.printStackTrace();
+            logger.log(Level.SEVERE, "Exception while closing replication channel consumer", ex);
         }
     }
 

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

@@ -147,7 +147,6 @@ public class ReplicationReceiver implements Runnable {
      */
     @Override
     public void run() {
-        ClassLoader oldClassLoader = Thread.currentThread().getContextClassLoader();
         long messageCount = 0;
         long operationCount = 0;
         final boolean logsFine = logger.isLoggable(Level.FINE);
@@ -156,6 +155,11 @@ public class ReplicationReceiver implements Runnable {
                 Delivery delivery = consumer.nextDelivery();
                 messageCount++;
                 if (_queue != null) {
+                    synchronized (this) {
+                        if (getInboundMessageQueue().isEmpty()) {
+                            notifyAll(); // wake up anyone waiting for isQueueEmpty()
+                        }
+                    }
                     if (logsFine || messageCount % 10l == 0) {
                         try {
                             logger.log(messageCount%10l==0 ? Level.INFO : Level.FINE,
@@ -174,7 +178,6 @@ public class ReplicationReceiver implements Runnable {
                 String replicableIdAsString = new DataInputStream(uncompressedInputStream).readUTF();
                 Replicable<?, ?> replicable = replicableProvider.getReplicable(replicableIdAsString, /* wait */ false);
                 if (replicable != null) {
-                    Thread.currentThread().setContextClassLoader(replicable.getClass().getClassLoader());
                     ObjectInputStream ois = new ObjectInputStream(uncompressedInputStream); // no special stream required; only reading a generic byte[]
                     int operationsInMessage = 0;
                     try {
@@ -248,11 +251,13 @@ public class ReplicationReceiver implements Runnable {
             } catch (Exception e) {
                 logger.info("Exception while processing replica: "+e.getMessage());
                 logger.log(Level.SEVERE, "run", e);
-            } finally {
-                Thread.currentThread().setContextClassLoader(oldClassLoader);
             }
         }
         logger.info("Stopped replicator thread. This server will no longer receive events from a master.");
+        synchronized (this) {
+            stopped = true;
+            notifyAll();
+        }
     }
 
     /**
@@ -313,9 +318,6 @@ public class ReplicationReceiver implements Runnable {
             queue = new ArrayList<>();
             queueByReplicableIdAsString.put(replicableIdAsString, queue);
         }
-        if (queue.isEmpty()) {
-            notifyAll();
-        }
         queue.add(new Pair<String, OperationWithResult<?, ?>>(replicable.getId().toString(), operation));
         assert !queue.isEmpty();
     }
@@ -359,6 +361,7 @@ public class ReplicationReceiver implements Runnable {
         logger.info("Signaled Replicator thread to stop asap.");
         stopped = true;
         master.stopConnection();
+        notifyAll(); // notify those waiting for stopped
     }
     
     public synchronized boolean isBeingStopped() {
@@ -374,8 +377,9 @@ public class ReplicationReceiver implements Runnable {
     /**
      * @return <code>true</code> if all queues for all replicables are empty
      */
-    public boolean isQueueEmpty() throws IllegalAccessException {
+    public boolean isQueueEmptyOrStopped() throws IllegalAccessException {
-        return (_queue == null || getInboundMessageQueue().isEmpty()) && !queueByReplicableIdAsString.values().stream().anyMatch(q->!q.isEmpty());
+        return isBeingStopped() ||
+                (_queue == null || getInboundMessageQueue().isEmpty()) && !queueByReplicableIdAsString.values().stream().anyMatch(q->!q.isEmpty());
     }
 
 }

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

@@ -246,7 +246,12 @@ public class ReplicationServiceImpl implements ReplicationService {
         replicator = null;
         serverUUID = UUID.randomUUID();
         logger.info("Setting " + serverUUID.toString() + " as unique replication identifier.");
+    }
     
+    @Override
+    protected void finalize() {
+        logger.info("terminating timer "+timer);
+        timer.cancel();
     }
 
     protected ServiceTracker<Replicable<?, ?>, Replicable<?, ?>> getReplicableTracker() {