Synchronize when adding session
Also confirm upload when adding session and status is NEW
@@ -47,7 +47,6 @@ import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Collection;
-import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -266,11 +265,6 @@ public class SessionRepository {
return getSessionList(curator.getChildren(sessionsPath));
}
- public void addRemoteSession(RemoteSession session) {
- remoteSessionCache.put(session.getSessionId(), session);
- metrics.incAddedSessions();
- }
-
public int deleteExpiredRemoteSessions(Clock clock, Duration expiryTime) {
int deleted = 0;
for (long sessionId : getRemoteSessions()) {
@@ -344,12 +338,18 @@ public class SessionRepository {
*
* @param sessionId session id for the new session
*/
- public void sessionAdded(long sessionId) {
+ public synchronized void sessionAdded(long sessionId) {
SessionZooKeeperClient sessionZKClient = createSessionZooKeeperClient(sessionId);
if (sessionZKClient.readStatus().equals(Session.Status.DELETE)) return;
log.log(Level.FINE, () -> "Adding remote session " + sessionId);
- createRemoteSession(sessionId);
+ RemoteSession session = createRemoteSession(sessionId);
+ if (session.getStatus() == Session.Status.NEW) {
+ log.log(Level.FINE, () -> session.logPre() + "Confirming upload for session " + sessionId);
+ session.confirmUpload();
+ } else {
+ log.log(Level.WARNING, () -> session.logPre() + "Session " + sessionId + " added, but with unexpected status " + session.getStatus());
+ }
if (distributeApplicationPackage())
createLocalSessionUsingDistributedApplicationPackage(sessionId);
}
@@ -376,7 +376,7 @@ public class SessionRepository {
long sessionId = remoteSession.getSessionId();
try (Lock lock = lock(sessionId)) {
// TODO: Change log level to FINE when debugging is finished
- log.log(Level.INFO, "Deactivating and deleting remote session " + sessionId);
+ log.log(Level.INFO, () -> remoteSession.logPre() + "Deactivating and deleting remote session " + sessionId);
remoteSession.deactivate();
remoteSession.delete();
remoteSessionCache.remove(sessionId);
@@ -384,7 +384,7 @@ public class SessionRepository {
LocalSession localSession = getLocalSession(sessionId);
if (localSession != null) {
// TODO: Change log level to FINE when debugging is finished
- log.log(Level.INFO, "Deleting local session " + sessionId);
+ log.log(Level.INFO, () -> localSession.logPre() + "Deleting local session " + sessionId);
deleteLocalSession(localSession);
}
}
@@ -434,26 +434,16 @@ public class SessionRepository {
log.log(Level.FINE, () -> "Got child event: " + event);
switch (event.getType()) {
case CHILD_ADDED:
- sessionsChanged();
- synchronizeOnNew(getSessionListFromDirectoryCache(Collections.singletonList(event.getData())));
- break;
case CHILD_REMOVED:
case CONNECTION_RECONNECTED:
sessionsChanged();
break;
+ default:
+ break;
}
});
}
- private void synchronizeOnNew(List<Long> sessionList) {
- for (long sessionId : sessionList) {
- RemoteSession session = remoteSessionCache.get(sessionId);
- if (session == null) continue; // session might have been deleted after getting session list
- log.log(Level.FINE, () -> session.logPre() + "Confirming upload for session " + sessionId);
- session.confirmUpload();
- }
- }
-
/**
* Creates a new deployment session from an application package.
*
