bug5728: manually-crafted test fix sequences must not undergo outlier filtering; fighting intermittent MongoDB causality errors

java/com.sap.sailing.domain.persistence/src/com/sap/sailing/domain/persistence/racelog/tracking/impl/MetadataCollection.java

@@ -2,7 +2,6 @@ package com.sap.sailing.domain.persistence.racelog.tracking.impl;
 
 import java.util.HashMap;
 import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.logging.Logger;
 
 import org.bson.Document;
@@ -43,7 +42,7 @@ import com.sap.sse.common.TypeBasedServiceFinder;
 public class MetadataCollection extends MongoFixHandler {
     private static final Logger logger = Logger.getLogger(MetadataCollection.class.getName());
     private final MongoCollection<Document> metadataCollection;
-    private final ConcurrentHashMap<DeviceIdentifier, MetadataUpdater> metadataUpdaters;
+    private final HashMap<DeviceIdentifier, MetadataUpdater> metadataUpdaters;
     private final WriteConcern writeConcern;
     private final ReadConcern readConcern;
 
@@ -52,7 +51,7 @@ public class MetadataCollection extends MongoFixHandler {
             TypeBasedServiceFinder<DeviceIdentifierMongoHandler> deviceServiceFinder, ReadConcern readConcern, WriteConcern writeConcern) {
         super(fixServiceFinder, deviceServiceFinder);
         this.metadataCollection = mongoOF.getGPSFixMetadataCollection();
-        this.metadataUpdaters = new ConcurrentHashMap<>();
+        this.metadataUpdaters = new HashMap<>();
         this.readConcern = readConcern;
         this.writeConcern = writeConcern;
     }
@@ -131,13 +130,13 @@ public class MetadataCollection extends MongoFixHandler {
         return result;
     }
 
-    <FixT extends Timed> void enqueueMetadataUpdate(DeviceIdentifier device, final Object dbDeviceId,
+    synchronized <FixT extends Timed> void enqueueMetadataUpdate(DeviceIdentifier device, final Object dbDeviceId,
             final int nrOfTotalFixes, TimeRange fixesTimeRange, FixT latestFix) throws TransformationException {
         final MetadataUpdater metadataUpdaterForDevice = metadataUpdaters.computeIfAbsent(device, d->new MetadataUpdater(this, device));
         metadataUpdaterForDevice.enqueueMetadataUpdate(device, dbDeviceId, nrOfTotalFixes, fixesTimeRange, latestFix);
     }
     
-    private void waitForPendingMetadataUpdates(DeviceIdentifier device) {
+    private synchronized void waitForPendingMetadataUpdates(DeviceIdentifier device) {
         final MetadataUpdater metadataUpdaterForDevice = metadataUpdaters.get(device);
         if (metadataUpdaterForDevice != null) {
             metadataUpdaterForDevice.waitForPendingUpdates();
@@ -157,6 +156,7 @@ public class MetadataCollection extends MongoFixHandler {
         MongoObjectFactoryImpl.storeTimeRange(newTimeRange, newMetadata, FieldNames.TIMERANGE);
         updateOperation.append("$set", newMetadata);
         updateOperation.append("$inc", new Document(FieldNames.NUM_FIXES.name(), update.getNrOfTotalFixes()));
+        logger.fine(()->"Updating sensor fix store metadata with update operation "+updateOperation);
         metadataCollection.withWriteConcern(writeConcern).updateOne(com.sap.sailing.shared.persistence.impl.MongoObjectFactoryImpl.getDeviceQuery(deviceServiceFinder, update.getDevice()), updateOperation, new UpdateOptions().upsert(true));
     }
 }

java/com.sap.sailing.domain.persistence/src/com/sap/sailing/domain/persistence/racelog/tracking/impl/MetadataUpdater.java

@@ -96,8 +96,6 @@ public class MetadataUpdater {
                     } else {
                         setNextUpdate(null);
                     }
-                    updatesProcessed++;
-                    MetadataUpdater.this.notifyAll();
                 }
                 if (theNextUpdate != null) {
                     boolean success = false;
@@ -106,6 +104,10 @@ public class MetadataUpdater {
                         try {
                             metadataCollection.update(theNextUpdate);
                             success = true;
+                            synchronized (MetadataUpdater.this) {
+                                updatesProcessed++;
+                                MetadataUpdater.this.notifyAll();
+                            }
                         } catch (Exception e) {
                             logger.severe("Unable to write update "+theNextUpdate+" to the metadata collection for device "+forDevice+
                                     ": "+e.getMessage()+"; retrying "+retryCount+" more times.");

java/com.sap.sailing.domain.racelogtrackingadapter.test/META-INF/MANIFEST.MF

@@ -12,7 +12,10 @@ Require-Bundle: org.junit;bundle-version="4.8.2",
  org.objenesis;bundle-version="1.3.0",
  org.hamcrest;bundle-version="1.1.0",
  com.sap.sse.common,
- com.sap.sailing.domain
+ com.sap.sailing.domain,
+ org.mongodb.driver-core,
+ com.sap.sse.mongodb,
+ org.mongodb.driver-sync
 Import-Package: com.sap.sailing.domain.abstractlog.regatta.events,
  com.sap.sailing.domain.persistence.impl,
  com.sap.sailing.domain.persistence.racelog.tracking.impl,

java/com.sap.sailing.domain.racelogtrackingadapter.test/src/com/sap/sailing/domain/racelogtracking/test/AbstractGPSFixStoreTest.java

@@ -12,6 +12,7 @@ import org.junit.Before;
 
 import com.mongodb.ReadConcern;
 import com.mongodb.WriteConcern;
+import com.mongodb.client.ClientSession;
 import com.sap.sailing.domain.abstractlog.race.impl.RaceLogImpl;
 import com.sap.sailing.domain.abstractlog.regatta.RegattaLog;
 import com.sap.sailing.domain.abstractlog.regatta.events.impl.RegattaLogDefineMarkEventImpl;
@@ -54,6 +55,7 @@ import com.sap.sailing.server.impl.RacingEventServiceImpl;
 import com.sap.sailing.server.interfaces.RacingEventService;
 import com.sap.sse.common.impl.DegreeBearingImpl;
 import com.sap.sse.common.impl.MillisecondsTimePoint;
+import com.sap.sse.mongodb.MongoDBService;
 
 public class AbstractGPSFixStoreTest extends RaceLogTrackingTestHelper {
     protected RacingEventService service;
@@ -67,6 +69,7 @@ public class AbstractGPSFixStoreTest extends RaceLogTrackingTestHelper {
     protected final Competitor comp = DomainFactory.INSTANCE.getOrCreateCompetitor("comp", "comp", null, null, null, null, null, /* timeOnTimeFactor */ null, /* timeOnDistanceAllowanceInSecondsPerNauticalMile */ null, null, /* storePersistently */ true);
     protected final Boat boat = DomainFactory.INSTANCE.getOrCreateBoat("boat", "boat", boatClass, "GER 234", null, /* storePersistently */ true);
     protected final Mark mark = DomainFactory.INSTANCE.getOrCreateMark("mark");
+    private ClientSession clientSession;
 
     protected GPSFixMoving createFix(long millis, double lat, double lng, double knots, double degrees) {
         return new GPSFixMovingImpl(new DegreePosition(lat, lng),
@@ -84,6 +87,7 @@ public class AbstractGPSFixStoreTest extends RaceLogTrackingTestHelper {
         raceLog = new RaceLogImpl("racelog");
         regattaLog = new RegattaLogImpl("regattalog");
         dropPersistedData();
+        clientSession = MongoDBService.INSTANCE.startCausallyConsistentSession();
         store = new MongoSensorFixStoreImpl(service.getMongoObjectFactory(), service.getDomainObjectFactory(),
                 serviceFinderFactory, ReadConcern.MAJORITY, WriteConcern.MAJORITY);
     }
@@ -91,14 +95,15 @@ public class AbstractGPSFixStoreTest extends RaceLogTrackingTestHelper {
     @After
     public void after() {
         dropPersistedData();
+        clientSession.close();
     }
 
     private void dropPersistedData() {
         MongoObjectFactoryImpl mongoOF = (MongoObjectFactoryImpl) service.getMongoObjectFactory();
-        mongoOF.getGPSFixCollection().drop();
+        mongoOF.getGPSFixCollection().withWriteConcern(WriteConcern.MAJORITY).drop();
-        mongoOF.getGPSFixMetadataCollection().drop();
+        mongoOF.getGPSFixMetadataCollection().withWriteConcern(WriteConcern.MAJORITY).drop();
-        mongoOF.getRaceLogCollection().drop();
+        mongoOF.getRaceLogCollection().withWriteConcern(WriteConcern.MAJORITY).drop();
-        mongoOF.getRegattaLogCollection().drop();
+        mongoOF.getRegattaLogCollection().withWriteConcern(WriteConcern.MAJORITY).drop();
     }
 
     protected void map(RegattaLog regattaLog, Competitor comp, DeviceIdentifier device, long from, long to) {

java/com.sap.sailing.domain.racelogtrackingadapter.test/src/com/sap/sailing/domain/racelogtracking/test/impl/SensorFixStoreAndLoadTest.java

@@ -447,7 +447,7 @@ public class SensorFixStoreAndLoadTest {
                 }
                 throw new IllegalArgumentException("Unknown event type");
             }
-        });
+        }, /* removeOutliersFromCompetitorTracks */ false);
     }
 
     protected void testNumberOfRawFixes(Track<?> track, long expected) {

java/com.sap.sailing.domain.racelogtrackingadapter.test/src/com/sap/sailing/domain/racelogtracking/test/impl/TrackedRaceLoadsFixesTest.java

@@ -90,7 +90,7 @@ public class TrackedRaceLoadsFixesTest extends AbstractGPSFixStoreTest {
         final DynamicTrackedRaceImpl trackedRace = createDynamicTrackedRace(boatClass, raceDefinition);
         trackedRace.setStartOfTrackingReceived(new MillisecondsTimePoint(1000));
         trackedRace.setEndOfTrackingReceived(new MillisecondsTimePoint(2000));
-        new FixLoaderAndTracker(trackedRace, store, null);
+        new FixLoaderAndTracker(trackedRace, store, null, /* removeOutliersFromCompetitorTracks */ false);
 
         trackedRace.attachRaceLog(raceLog);
         trackedRace.attachRegattaLog(regattaLog);
@@ -134,7 +134,7 @@ public class TrackedRaceLoadsFixesTest extends AbstractGPSFixStoreTest {
 
         DynamicTrackedRace trackedRace = createDynamicTrackedRace(boatClass, raceDefinition);
 
-        new FixLoaderAndTracker(trackedRace, store, null);
+        new FixLoaderAndTracker(trackedRace, store, null, /* removeOutliersFromCompetitorTracks */ false);
         
         raceLog.add(new RaceLogStartOfTrackingEventImpl(TimePoint.BeginningOfTime, author, 0));
         trackedRace.attachRaceLog(raceLog);
@@ -529,7 +529,7 @@ public class TrackedRaceLoadsFixesTest extends AbstractGPSFixStoreTest {
         DynamicTrackedRace trackedRace = createDynamicTrackedRace(boatClass, raceDefinition);
         trackedRace.attachRaceLog(raceLog);
         trackedRace.attachRegattaLog(regattaLog);
-        new FixLoaderAndTracker(trackedRace, store, null);
+        new FixLoaderAndTracker(trackedRace, store, null, /* removeOutliersFromCompetitorTracks */ false);
         for(Consumer<DynamicTrackedRace> afterTrackingStarted : afterTrackingStartedConsumers) {
             trackedRace.waitForLoadingToFinish();
             afterTrackingStarted.accept(trackedRace);

java/com.sap.sailing.domain.racelogtrackingadapter/src/com/sap/sailing/domain/racelogtracking/impl/fixtracker/FixLoaderAndTracker.java

@@ -154,6 +154,12 @@ public class FixLoaderAndTracker implements TrackingDataLoader {
 
     private final SensorFixMapperFactory sensorFixMapperFactory;
     
+    /**
+     * If set to {@code true} in the constructor, fixes loaded for competitor tracks will be subject to
+     * outlier removal using the {@link OutlierFilter}.
+     */
+    private final boolean removeOutliersFromCompetitorTracks;
+    
     /**
      * This flag is used to tell the loaders/trackers whether preemptive stopping has been requested. If switched
      * to {@code true}, running loaders will stop loading fixes and return immediately.
@@ -168,6 +174,7 @@ public class FixLoaderAndTracker implements TrackingDataLoader {
     private AtomicBoolean willBeRemovedAfterStopping = new AtomicBoolean(false);
     
     private AtomicBoolean stopRequested = new AtomicBoolean(false);
+    
     private final AbstractRaceChangeListener raceChangeListener = new AbstractRaceChangeListener() {
         @Override
         public void startOfTrackingChanged(TimePoint oldStartOfTracking, TimePoint newStartOfTracking) {
@@ -196,6 +203,7 @@ public class FixLoaderAndTracker implements TrackingDataLoader {
             deviceMappings.addRegattaLog(regattaLog);
         }
     };
+    
     private final FixReceivedListener<Timed> listener = new FixReceivedListener<Timed>() {
         @Override
         public Iterable<Triple<RegattaAndRaceIdentifier, Boolean, Duration>> fixReceived(DeviceIdentifier device, Timed fix, boolean returnManeuverChanges, boolean returnLiveDelay) {
@@ -394,10 +402,11 @@ public class FixLoaderAndTracker implements TrackingDataLoader {
     }
 
     public FixLoaderAndTracker(DynamicTrackedRace trackedRace, SensorFixStore sensorFixStore,
-            SensorFixMapperFactory sensorFixMapperFactory) {
+            SensorFixMapperFactory sensorFixMapperFactory, boolean removeOutliersFromCompetitorTracks) {
         this.sensorFixStore = sensorFixStore;
         this.sensorFixMapperFactory = sensorFixMapperFactory;
         this.trackedRace = trackedRace;
+        this.removeOutliersFromCompetitorTracks = removeOutliersFromCompetitorTracks;
         startTracking();
     }
 
@@ -561,8 +570,6 @@ public class FixLoaderAndTracker implements TrackingDataLoader {
                 /**
                  * First loads the fixes into a temporary track which is then subject to outlier filtering (see
                  * {@link OutlierFilter}). The fixes that make it through outlier filtering are then inserted 
-                 * @param competitor
-                 * @param event
                  */
                 private void loadForCompetitor(Competitor competitor, RegattaLogDeviceMappingEvent<?> event) {
                     DynamicGPSFixTrack<Competitor, GPSFixMoving> track = trackedRace.getTrack(competitor);
@@ -574,17 +581,22 @@ public class FixLoaderAndTracker implements TrackingDataLoader {
                                 new DynamicGPSFixMovingTrackImpl<Competitor>(track.getTrackedItem(), track.getMillisecondsOverWhichToAverageSpeed());
                         loadedFixes.suspendValidityAndMaxSpeedCaching(); // no validity nor max speed required on this track
                         try {
+                            final DynamicGPSFixTrack<Competitor, GPSFixMoving> filteredTrack;
                             sensorFixStore.<GPSFixMoving> loadFixes(fix -> loadedFixes.add(fix, true), event.getDevice(),
                                     timeRangeToLoad.from(), timeRangeToLoad.to(), /* toIsInclusive */ false,
                                     stopCallback, progressConsumer);
-                            final Pair<Integer, DynamicGPSFixTrack<Competitor, GPSFixMoving>> filtered = new OutlierFilter().findAndRemoveInconsistenciesOnRawFixes(loadedFixes);
+                            if (removeOutliersFromCompetitorTracks) {
-                            loadedFixes.lockForRead();
+                                final Pair<Integer, DynamicGPSFixTrack<Competitor, GPSFixMoving>> filtered = new OutlierFilter().findAndRemoveInconsistenciesOnRawFixes(loadedFixes);
-                            try {
+                                loadedFixes.lockForRead();
-                                logger.info("Filtered competitor track for outliers; "+filtered.getA()+" outliers removed in track with "+Util.size(loadedFixes.getRawFixes())+" fixes");
+                                try {
-                            } finally {
+                                    logger.info("Filtered competitor track for outliers; "+filtered.getA()+" outliers removed in track with "+Util.size(loadedFixes.getRawFixes())+" fixes");
-                                loadedFixes.unlockAfterRead();
+                                } finally {
+                                    loadedFixes.unlockAfterRead();
+                                }
+                                filteredTrack = filtered.getB();
+                            } else {
+                                filteredTrack = loadedFixes;
                             }
-                            final DynamicGPSFixTrack<Competitor, GPSFixMoving> filteredTrack = filtered.getB();
                             track.suspendValidityAndMaxSpeedCaching();
                             filteredTrack.lockForRead();
                             try {

java/com.sap.sailing.domain.racelogtrackingadapter/src/com/sap/sailing/domain/racelogtracking/impl/fixtracker/RaceLogFixTrackerManager.java

@@ -128,7 +128,7 @@ public class RaceLogFixTrackerManager implements TrackingDataLoader {
     private synchronized void startTrackerIfNotAlreadyStarted() {
         if (tracker == null) {
             logger.fine("Starting fix tracker for TrackedRace: " + trackedRace.getRaceIdentifier());
-            tracker = new FixLoaderAndTracker(trackedRace, sensorFixStore, sensorFixMapperFactory);
+            tracker = new FixLoaderAndTracker(trackedRace, sensorFixStore, sensorFixMapperFactory, /* removeOutliersFromCompetitorTracks */ true);
         }
     }
 

java/com.sap.sse.mongodb/src/com/sap/sse/mongodb/MongoDBService.java

@@ -49,4 +49,6 @@ public interface MongoDBService {
      * Those weak references that got cleared will be removed from the cache.
      */
     ClientSession startAutoRefreshingSession();
+
+    ClientSession startCausallyConsistentSession();
 }

java/com.sap.sse.mongodb/src/com/sap/sse/mongodb/internal/MongoDBServiceImpl.java

@@ -13,6 +13,7 @@ import java.util.logging.Logger;
 import org.bson.BsonArray;
 import org.bson.BsonDocument;
 
+import com.mongodb.ClientSessionOptions;
 import com.mongodb.ConnectionString;
 import com.mongodb.client.ClientSession;
 import com.mongodb.client.MongoClient;
@@ -72,6 +73,12 @@ public class MongoDBServiceImpl implements MongoDBService {
         }
     }
     
+    @Override
+    public ClientSession startCausallyConsistentSession() {
+        ensureConfigurationDefaultingToTest();
+        return getMongo(getConfiguration()).startSession(ClientSessionOptions.builder().causallyConsistent(true).build());
+    }
+    
     @Override
     public ClientSession startAutoRefreshingSession() {
         ensureConfigurationDefaultingToTest();

java/target/configuration/logging_debug.properties

@@ -49,3 +49,5 @@ com.sap.sse.replication.impl.OperationQueueByKeyExecutor.level = INFO
 
 # YB logging
 com.sap.sailing.domain.yellowbrickadapter.level = ALL
+
+com.sap.sailing.domain.persistence.racelog.tracking.impl.MetadataCollection.level = FINE