Synchronize access to sessionStateWatchers map
@@ -315,7 +315,7 @@ public class SessionRepository {
public void deleteLocalSession(LocalSession session) {
long sessionId = session.getSessionId();
log.log(Level.FINE, () -> "Deleting local session " + sessionId);
- SessionStateWatcher watcher = sessionStateWatchers.remove(sessionId);
+ SessionStateWatcher watcher = removeSessionStateWatcher(sessionId);
if (watcher != null) watcher.close();
localSessionCache.remove(sessionId);
NestedTransaction transaction = new NestedTransaction();
@@ -323,6 +323,12 @@ public class SessionRepository {
transaction.commit();
}
+ private SessionStateWatcher removeSessionStateWatcher(long sessionId) {
+ synchronized (sessionStateWatchers) {
+ return sessionStateWatchers.remove(sessionId);
+ }
+ }
+
private void deleteAllSessions() {
for (LocalSession session : getLocalSessions()) {
deleteLocalSession(session);
@@ -886,6 +892,7 @@ public class SessionRepository {
}
private void updateSessionStateWatcher(long sessionId, RemoteSession remoteSession) {
+ synchronized (sessionStateWatchers) {
SessionStateWatcher sessionStateWatcher = sessionStateWatchers.get(sessionId);
if (sessionStateWatcher == null) {
Curator.FileCache fileCache = curator.createFileCache(getSessionStatePath(sessionId).getAbsolute(), false);
@@ -895,6 +902,7 @@ public class SessionRepository {
sessionStateWatcher.updateRemoteSession(remoteSession);
}
}
+ }
@Override
public String toString() {
@@ -928,7 +936,7 @@ public class SessionRepository {
long sessionId = it.next().sessionId;
if (existingSessions.contains(sessionId)) continue;
- SessionStateWatcher watcher = sessionStateWatchers.remove(sessionId);
+ SessionStateWatcher watcher = removeSessionStateWatcher(sessionId);
if (watcher != null) watcher.close();
it.remove();
metricUpdater.incRemovedSessions();
