fixed distributed indicator value cache handling (patch-fix 1.3.2)

src/main/java/ch/ethz/seb/sebserver/webservice/servicelayer/session/impl/ClientIndicatorFactory.java

@@ -26,8 +26,10 @@ import ch.ethz.seb.sebserver.gbl.model.exam.Indicator;
 import ch.ethz.seb.sebserver.gbl.model.exam.Indicator.IndicatorType;
 import ch.ethz.seb.sebserver.gbl.model.session.ClientConnection;
 import ch.ethz.seb.sebserver.gbl.profile.WebServiceProfile;
+import ch.ethz.seb.sebserver.gbl.util.Utils;
 import ch.ethz.seb.sebserver.webservice.servicelayer.dao.IndicatorDAO;
 import ch.ethz.seb.sebserver.webservice.servicelayer.session.ClientIndicator;
+import ch.ethz.seb.sebserver.webservice.servicelayer.session.impl.indicator.DistributedIndicatorValueService;
 import ch.ethz.seb.sebserver.webservice.servicelayer.session.impl.indicator.PingIntervalClientIndicator;
 
 @Lazy
@@ -39,27 +41,66 @@ public class ClientIndicatorFactory {
 
     private final ApplicationContext applicationContext;
     private final IndicatorDAO indicatorDAO;
+    private final DistributedIndicatorValueService distributedPingCache;
+    private final boolean distributedSetup;
     private final boolean enableCaching;
 
     @Autowired
     public ClientIndicatorFactory(
             final ApplicationContext applicationContext,
             final IndicatorDAO indicatorDAO,
+            final DistributedIndicatorValueService distributedPingCache,
             @Value("${sebserver.webservice.distributed:false}") final boolean distributedSetup,
             @Value("${sebserver.webservice.api.exam.enable-indicator-cache:true}") final boolean enableCaching) {
 
         this.applicationContext = applicationContext;
         this.indicatorDAO = indicatorDAO;
+        this.distributedPingCache = distributedPingCache;
+        this.distributedSetup = distributedSetup;
         this.enableCaching = distributedSetup ? false : enableCaching;
     }
 
-    public List<ClientIndicator> createFor(final ClientConnection clientConnection) {
+    public void initializeDistributedCaches(final ClientConnection clientConnection) {
-        return createFor(clientConnection, false);
+        try {
+
+            if (!this.distributedSetup || clientConnection.examId == null) {
+                return;
             }
 
-    public List<ClientIndicator> createFor(
+            final Collection<Indicator> examIndicators = this.indicatorDAO
-            final ClientConnection clientConnection,
+                    .allForExam(clientConnection.examId)
-            final boolean enableCachingOverride) {
+                    .getOrThrow();
+
+            boolean pingIndicatorAvailable = false;
+            for (final Indicator indicatorDef : examIndicators) {
+
+                this.distributedPingCache.createIndicatorForConnection(
+                        clientConnection.id,
+                        indicatorDef.type,
+                        indicatorDef.type == IndicatorType.LAST_PING ? Utils.getMillisecondsNow() : 0L);
+
+                if (!pingIndicatorAvailable) {
+                    pingIndicatorAvailable = indicatorDef.type == IndicatorType.LAST_PING;
+                }
+            }
+
+            // If there is no ping interval indicator set from the exam, we add a hidden one
+            // to at least create missing ping events and track missing state
+            if (!pingIndicatorAvailable) {
+                this.distributedPingCache.createIndicatorForConnection(
+                        clientConnection.id,
+                        IndicatorType.LAST_PING,
+                        Utils.getMillisecondsNow());
+            }
+
+        } catch (final Exception e) {
+            log.error("Unexpected error while trying to initialize distributed indicator value cache for: {}",
+                    clientConnection,
+                    e);
+        }
+    }
+
+    public List<ClientIndicator> createFor(final ClientConnection clientConnection) {
 
         final List<ClientIndicator> result = new ArrayList<>();
         if (clientConnection.examId == null) {
@@ -73,7 +114,6 @@ public class ClientIndicatorFactory {
                     .getOrThrow();
 
             boolean pingIndicatorAvailable = false;
-
             for (final Indicator indicatorDef : examIndicators) {
                 try {
 
@@ -88,11 +128,12 @@ public class ClientIndicatorFactory {
                             indicatorDef,
                             clientConnection.id,
                             clientConnection.status.clientActiveStatus,
-                            this.enableCaching || enableCachingOverride);
+                            this.enableCaching);
 
                     result.add(indicator);
                 } catch (final Exception e) {
-                    log.warn("No Indicator with type: {} found as registered bean. Ignore this one.", indicatorDef.type,
+                    log.warn("No Indicator with type: {} found as registered bean. Ignore this one.",
+                            indicatorDef.type,
                             e);
                 }
             }
@@ -117,7 +158,7 @@ public class ClientIndicatorFactory {
                         indicator,
                         clientConnection.id,
                         clientConnection.status.clientActiveStatus,
-                        this.enableCaching || enableCachingOverride);
+                        this.enableCaching);
                 result.add(pingIndicator);
             }
 

src/main/java/ch/ethz/seb/sebserver/webservice/servicelayer/session/impl/ExamSessionCacheService.java

@@ -148,9 +148,7 @@ public class ExamSessionCacheService {
         if (clientConnection == null) {
             return null;
         } else {
-            return this.internalClientConnectionDataFactory.createClientConnectionData(
+            return this.internalClientConnectionDataFactory.createClientConnectionData(clientConnection);
-                    clientConnection,
-                    this.getRunningExam(clientConnection.examId) != null);
         }
     }
 

src/main/java/ch/ethz/seb/sebserver/webservice/servicelayer/session/impl/ExamSessionServiceImpl.java

@@ -281,8 +281,8 @@ public class ExamSessionServiceImpl implements ExamSessionService {
             throw new IllegalStateException("Missing exam identifier or requested exam is not running");
         }
 
-        if (log.isDebugEnabled()) {
+        if (log.isTraceEnabled()) {
-            log.debug("Trying to get exam from InMemorySEBConfig");
+            log.trace("Trying to get exam from InMemorySEBConfig");
         }
 
         final InMemorySEBConfig sebConfigForExam = this.examSessionCacheService
@@ -295,14 +295,14 @@ public class ExamSessionServiceImpl implements ExamSessionService {
 
         try {
 
-            if (log.isDebugEnabled()) {
+            if (log.isTraceEnabled()) {
-                log.debug("SEB exam configuration download request, start writing SEB exam configuration");
+                log.trace("SEB exam configuration download request, start writing SEB exam configuration");
             }
 
             out.write(sebConfigForExam.getData());
 
-            if (log.isDebugEnabled()) {
+            if (log.isTraceEnabled()) {
-                log.debug("SEB exam configuration download request, finished writing SEB exam configuration");
+                log.trace("SEB exam configuration download request, finished writing SEB exam configuration");
             }
 
         } catch (final IOException e) {

src/main/java/ch/ethz/seb/sebserver/webservice/servicelayer/session/impl/InternalClientConnectionDataFactory.java

@@ -32,16 +32,7 @@ public class InternalClientConnectionDataFactory {
         this.sebClientNotificationService = sebClientNotificationService;
     }
 
-    public ClientConnectionDataInternal createClientConnectionData(
+    public ClientConnectionDataInternal createClientConnectionData(final ClientConnection clientConnection) {
-            final ClientConnection clientConnection,
-            final boolean examRunning) {
-
-        if (!examRunning) {
-            return new ClientConnectionDataInternal(
-                    clientConnection,
-                    () -> false,
-                    this.clientIndicatorFactory.createFor(clientConnection, true));
-        }
 
         if (clientConnection.status == ConnectionStatus.CLOSED
                 || clientConnection.status == ConnectionStatus.DISABLED) {

src/main/java/ch/ethz/seb/sebserver/webservice/servicelayer/session/impl/SEBClientConnectionServiceImpl.java

@@ -74,8 +74,8 @@ public class SEBClientConnectionServiceImpl implements SEBClientConnectionServic
     private final SEBClientConfigDAO sebClientConfigDAO;
     private final SEBClientInstructionService sebInstructionService;
     private final ExamAdminService examAdminService;
-    // TODO get rid of this dependency and use application events for signaling client connection state changes
     private final DistributedIndicatorValueService distributedPingCache;
+    private final ClientIndicatorFactory clientIndicatorFactory;
     private final boolean isDistributedSetup;
 
     protected SEBClientConnectionServiceImpl(
@@ -84,7 +84,8 @@ public class SEBClientConnectionServiceImpl implements SEBClientConnectionServic
             final SEBClientConfigDAO sebClientConfigDAO,
             final SEBClientInstructionService sebInstructionService,
             final ExamAdminService examAdminService,
-            final DistributedIndicatorValueService distributedPingCache) {
+            final DistributedIndicatorValueService distributedPingCache,
+            final ClientIndicatorFactory clientIndicatorFactory) {
 
         this.examSessionService = examSessionService;
         this.examSessionCacheService = examSessionService.getExamSessionCacheService();
@@ -96,6 +97,7 @@ public class SEBClientConnectionServiceImpl implements SEBClientConnectionServic
         this.examAdminService = examAdminService;
         this.distributedPingCache = distributedPingCache;
         this.isDistributedSetup = sebInstructionService.getWebserviceInfo().isDistributed();
+        this.clientIndicatorFactory = clientIndicatorFactory;
     }
 
     @Override
@@ -165,6 +167,11 @@ public class SEBClientConnectionServiceImpl implements SEBClientConnectionServic
                     null))
                     .getOrThrow();
 
+            // initialize distributed indicator value caches if possible and needed
+            if (clientConnection.examId != null && this.isDistributedSetup) {
+                this.clientIndicatorFactory.initializeDistributedCaches(clientConnection);
+            }
+
             // load client connection data into cache
             final ClientConnectionDataInternal activeClientConnection = this.examSessionService
                     .getConnectionDataInternal(connectionToken);
@@ -262,6 +269,11 @@ public class SEBClientConnectionServiceImpl implements SEBClientConnectionServic
                             null))
                     .getOrThrow();
 
+            // initialize distributed indicator value caches if possible and needed
+            if (examId != null && this.isDistributedSetup) {
+                this.clientIndicatorFactory.initializeDistributedCaches(clientConnection);
+            }
+
             final ClientConnectionDataInternal activeClientConnection =
                     reloadConnectionCache(connectionToken);
 
@@ -402,6 +414,11 @@ public class SEBClientConnectionServiceImpl implements SEBClientConnectionServic
             // check exam integrity for established connection
             checkExamIntegrity(establishedClientConnection.examId);
 
+            // initialize distributed indicator value caches if possible and needed
+            if (examId != null && this.isDistributedSetup) {
+                this.clientIndicatorFactory.initializeDistributedCaches(clientConnection);
+            }
+
             // if proctoring is enabled for exam, mark for room update
             if (proctoringEnabled) {
                 this.clientConnectionDAO.markForProctoringUpdate(updatedClientConnection.id);
@@ -869,13 +886,6 @@ public class SEBClientConnectionServiceImpl implements SEBClientConnectionServic
                     connection.getIndicatorMapping(EventType.ERROR_LOG)
                             .forEach(indicator -> indicator.notifyValueChange(clientEventRecord));
                 }
-
-                if (this.isDistributedSetup) {
-                    // mark for update and flush the cache
-                    this.clientConnectionDAO.save(connection.clientConnection);
-                    this.examSessionCacheService.evictClientConnection(
-                            connection.clientConnection.connectionToken);
-                }
             }
         };
     }

src/main/java/ch/ethz/seb/sebserver/webservice/servicelayer/session/impl/indicator/AbstractClientIndicator.java

@@ -69,37 +69,34 @@ public abstract class AbstractClientIndicator implements ClientIndicator {
         this.cachingEnabled = cachingEnabled;
 
         if (!this.cachingEnabled && this.active) {
-            try {
+            this.ditributedIndicatorValueRecordId = this.distributedPingCache
-                this.ditributedIndicatorValueRecordId = this.distributedPingCache.initIndicatorForConnection(
+                    .getIndicatorForConnection(connectionId, getType());
-                        connectionId,
+//            if (this.ditributedIndicatorValueRecordId == null) {
-                        getType(),
+//                tryRecoverIndicatorRecord();
-                        initValue());
+//            }
-            } catch (final Exception e) {
+//            try {
-                tryRecoverIndicatorRecord();
+//                this.ditributedIndicatorValueRecordId = this.distributedPingCache.initIndicatorForConnection(
-            }
+//                        connectionId,
+//                        getType(),
+//                        initValue());
+//            } catch (final Exception e) {
+//                tryRecoverIndicatorRecord();
+//            }
         }
 
         this.currentValue = computeValueAt(Utils.getMillisecondsNow());
         this.initialized = true;
     }
 
-    protected long initValue() {
-        return 0;
-    }
-
     protected void tryRecoverIndicatorRecord() {
-
+        this.ditributedIndicatorValueRecordId = this.distributedPingCache.getIndicatorForConnection(
-        if (log.isWarnEnabled()) {
-            log.warn("*** Missing indicator value record for connection: {}. Try to recover...", this.connectionId);
-        }
-
-        try {
-            this.ditributedIndicatorValueRecordId = this.distributedPingCache.initIndicatorForConnection(
                 this.connectionId,
-                    getType(),
+                getType());
-                    initValue());
+
-        } catch (final Exception e) {
+        if (this.ditributedIndicatorValueRecordId == null) {
-            log.error("Failed to recover indicator value record for connection: {}", this.connectionId, e);
+            log.warn("Failed to recover from missing indicator value cache record: {} type: {}",
+                    this.connectionId,
+                    getType());
         }
     }
 

src/main/java/ch/ethz/seb/sebserver/webservice/servicelayer/session/impl/indicator/AbstractLogLevelCountIndicator.java

@@ -55,8 +55,8 @@ public abstract class AbstractLogLevelCountIndicator extends AbstractLogIndicato
     @Override
     public double computeValueAt(final long timestamp) {
 
-        if (log.isDebugEnabled()) {
+        if (log.isTraceEnabled()) {
-            log.debug("computeValueAt: {}", timestamp);
+            log.trace("computeValueAt: {}", timestamp);
         }
 
         try {

src/main/java/ch/ethz/seb/sebserver/webservice/servicelayer/session/impl/indicator/AbstractLogNumberIndicator.java

@@ -73,8 +73,8 @@ public abstract class AbstractLogNumberIndicator extends AbstractLogIndicator {
     @Override
     public double computeValueAt(final long timestamp) {
 
-        if (log.isDebugEnabled()) {
+        if (log.isTraceEnabled()) {
-            log.debug("computeValueAt: {}", timestamp);
+            log.trace("computeValueAt: {}", timestamp);
         }
 
         try {

src/main/java/ch/ethz/seb/sebserver/webservice/servicelayer/session/impl/indicator/DistributedIndicatorValueService.java

@@ -129,70 +129,39 @@ public class DistributedIndicatorValueService implements DisposableBean {
         }
     }
 
-    /** This initializes a SEB client indicator on the persistent storage for a given SEB client
+    /** This creates a distributed indicator value cache record for a given SEB connection and indicator
-     * connection identifier and of given IndicatorType.
+     * if it not already exists and returns the PK for the specified distributed indicator value cache record
-     * If there is already such an indicator for the specified SEB client connection identifier and type,
-     * this returns the id of the existing one.
      *
-     * @param connectionId SEB client connection identifier
+     * @param connectionId the client connection identifier
-     * @param type indicator type
+     * @param type the indicator type
-     * @param value the initial indicator value
+     * @param value the initialization value
-     * @return SEB client indicator value identifier (PK) */
+     * @return the PK of the created or existing distributed indicator value cache record or null when a unexpected
+     *         error happened */
     @Transactional
-    public Long initIndicatorForConnection(
+    public Long createIndicatorForConnection(
             final Long connectionId,
             final IndicatorType type,
-            final Long value) {
+            final long initValue) {
 
-        try {
+        if (!this.webserviceInfo.isDistributed()) {
-
+            log.warn("No distributed setup, skip createIndicatorForConnection");
-            if (log.isDebugEnabled()) {
-                log.trace("*** Initialize indicator value record for SEB connection: {}", connectionId);
-            }
-
-            synchronized (this) {
-
-                Long recordId = null;
-
-                try {
-                    recordId = this.clientIndicatorValueMapper
-                            .indicatorRecordIdByConnectionId(connectionId, type);
-                } catch (final Exception e) {
-                    // There is already more then one indicator record entry!!!
-                    // delete the second one and work on with the first one
-
-                    log.warn("Duplicate indicator entry detected for connectionId: {}, type: {}  --> try to recover",
-                            connectionId, type);
-
-                    try {
-                        final List<ClientIndicatorRecord> records = this.clientIndicatorRecordMapper.selectByExample()
-                                .where(ClientIndicatorRecordDynamicSqlSupport.clientConnectionId,
-                                        isEqualTo(connectionId))
-                                .and(ClientIndicatorRecordDynamicSqlSupport.type, isEqualTo(type.id))
-                                .build()
-                                .execute();
-                        if (records.size() > 1) {
-                            this.clientIndicatorRecordMapper.deleteByPrimaryKey(records.get(1).getId());
-                        }
-
-                        return records.get(0).getId();
-                    } catch (final Exception ee) {
-                        log.error("Failed to recover from duplicate indicator entry: ", ee);
-                        return null;
-                    }
-                }
-
-                if (recordId == null) {
-                    if (!this.webserviceInfo.isMaster()) {
-                        if (log.isDebugEnabled()) {
-                            log.debug("Skip indicator record init because this is no master instance");
-                        }
             return null;
         }
 
+        try {
+
+            // first check if the record already exists
+            final Long recId = this.clientIndicatorValueMapper.indicatorRecordIdByConnectionId(
+                    connectionId,
+                    type);
+            if (recId != null) {
+                log.debug("Distributed indicator value cache already exists for: {}, {}", connectionId, type);
+                return recId;
+            }
+
+            // if not, create new one and return PK
             final ClientIndicatorRecord clientEventRecord = new ClientIndicatorRecord(
-                            null, connectionId, type.id, value);
+                    null, connectionId, type.id, initValue);
-
             this.clientIndicatorRecordMapper.insert(clientEventRecord);
 
             try {
@@ -211,18 +180,38 @@ public class DistributedIndicatorValueService implements DisposableBean {
                 TransactionInterceptor.currentTransactionStatus().setRollbackOnly();
                 throw new RuntimeException("Detected multiple client indicator value entries");
             }
+        } catch (final Exception e) {
+            log.error(
+                    "Failed to initialize distributed indicator value cache in persistent store. connectionId: {} type: {}",
+                    connectionId, type, e);
+
+            return null;
+        }
     }
 
-                return recordId;
+    /** Get the distributed indicator value cache record PK for a given SEB connection and indicator if available.
+     * If not existing for the specified connection and indicator this return null
+     *
+     * @param connectionId the client connection identifier
+     * @param type the indicator type
+     * @return the indicator value cache record PK or null of not defined */
+    @Transactional(readOnly = true)
+    public Long getIndicatorForConnection(final Long connectionId, final IndicatorType type) {
+        try {
+
+            return this.clientIndicatorValueMapper
+                    .indicatorRecordIdByConnectionId(connectionId, type);
 
-            }
         } catch (final Exception e) {
 
-            log.error("Failed to initialize indicator value for connection -> {}", connectionId, e);
+            if (log.isDebugEnabled()) {
+                log.debug("Failed to get indicator PK for connection: {} type: {} cause: {}",
+                        connectionId,
+                        type,
+                        e.getMessage());
+            }
 
-            // force rollback
+            return null;
-            TransactionInterceptor.currentTransactionStatus().setRollbackOnly();
-            throw new RuntimeException("Failed to initialize indicator value for connection -> " + connectionId, e);
         }
     }
 
@@ -235,7 +224,7 @@ public class DistributedIndicatorValueService implements DisposableBean {
         try {
 
             if (log.isDebugEnabled()) {
-                log.debug("*** Delete indicator value record for SEB connection: {}", connectionId);
+                log.debug("Delete indicator value record for SEB connection: {}", connectionId);
             }
 
             final Collection<ClientIndicatorValueRecord> records = this.clientIndicatorValueMapper

src/main/java/ch/ethz/seb/sebserver/webservice/servicelayer/session/impl/indicator/PingIntervalClientIndicator.java

@@ -20,7 +20,6 @@ import ch.ethz.seb.sebserver.gbl.Constants;
 import ch.ethz.seb.sebserver.gbl.model.exam.Indicator;
 import ch.ethz.seb.sebserver.gbl.model.exam.Indicator.IndicatorType;
 import ch.ethz.seb.sebserver.gbl.model.session.ClientEvent;
-import ch.ethz.seb.sebserver.gbl.util.Utils;
 import ch.ethz.seb.sebserver.webservice.datalayer.batis.model.ClientEventRecord;
 
 @Lazy
@@ -40,11 +39,6 @@ public final class PingIntervalClientIndicator extends AbstractPingIndicator {
         this.cachingEnabled = true;
     }
 
-    @Override
-    protected long initValue() {
-        return Utils.getMillisecondsNow();
-    }
-
     @Override
     public void init(
             final Indicator indicatorDefinition,