[grid] Guarding session stop in case the request fails
Should fix
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.RemovalListener;
+import com.google.common.cache.RemovalNotification;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
@@ -183,31 +184,7 @@ private LocalNode(
this.currentSessions = CacheBuilder.newBuilder()
.expireAfterAccess(sessionTimeout)
.ticker(ticker)
- .removalListener((RemovalListener<SessionId, SessionSlot>) notification -> {
- if (notification.getKey() != null && notification.getValue() != null) {
- SessionSlot slot = notification.getValue();
- SessionId id = notification.getKey();
- if (notification.wasEvicted()) {
- // Session is timing out, stopping it by sending a DELETE
- LOG.log(Level.INFO, () -> String.format("Session id %s timed out, stopping...", id));
- slot.execute(new HttpRequest(DELETE, "/session/" + id));
- }
- // Attempt to stop the session
- slot.stop();
- // Invalidate temp file system
- this.tempFileSystems.invalidate(id);
- // Decrement pending sessions if Node is draining
- if (this.isDraining()) {
- int done = pendingSessions.decrementAndGet();
- if (done <= 0) {
- LOG.info("Node draining complete!");
- bus.fire(new NodeDrainComplete(this.getId()));
- }
- }
- } else {
- LOG.log(Debug.getDebugLogLevel(), "Received stop session notification with null values");
- }
- })
+ .removalListener(this::stopTimedOutSession)
.build();
ScheduledExecutorService sessionCleanupNodeService =
@@ -250,6 +227,36 @@ private LocalNode(
new JMXHelper().register(this);
}
+ private void stopTimedOutSession(RemovalNotification<SessionId, SessionSlot> notification) {
+ if (notification.getKey() != null && notification.getValue() != null) {
+ SessionSlot slot = notification.getValue();
+ SessionId id = notification.getKey();
+ if (notification.wasEvicted()) {
+ // Session is timing out, stopping it by sending a DELETE
+ LOG.log(Level.INFO, () -> String.format("Session id %s timed out, stopping...", id));
+ try {
+ slot.execute(new HttpRequest(DELETE, "/session/" + id));
+ } catch (Exception e) {
+ LOG.log(Level.WARNING, String.format("Exception while trying to stop session %s", id), e);
+ }
+ }
+ // Attempt to stop the session
+ slot.stop();
+ // Invalidate temp file system
+ this.tempFileSystems.invalidate(id);
+ // Decrement pending sessions if Node is draining
+ if (this.isDraining()) {
+ int done = pendingSessions.decrementAndGet();
+ if (done <= 0) {
+ LOG.info("Node draining complete!");
+ bus.fire(new NodeDrainComplete(this.getId()));
+ }
+ }
+ } else {
+ LOG.log(Debug.getDebugLogLevel(), "Received stop session notification with null values");
+ }
+ }
+
public static Builder builder(
Tracer tracer,
EventBus bus,
