public Optional<Session> login(@Nullable String currentSessionId, String host,
                                   ActorAwareAuthenticationToken authToken) throws AuthenticationServiceUnavailableException {

        final String previousSessionId = StringUtils.defaultIfBlank(currentSessionId, null);
        final Subject subject = new Subject.Builder().sessionId(previousSessionId).host(host).buildSubject();

        ThreadContext.bind(subject);

        try {
            final Session session = subject.getSession();

            subject.login(authToken);

            return createSession(subject, session, host);
        } catch (AuthenticationServiceUnavailableException e) {
            log.info("Session creation failed due to authentication service being unavailable. Actor: \"{}\"",
                    authToken.getActor().urn());
            final Map<String, Object> auditEventContext = ImmutableMap.of(
                    "remote_address", host,
                    "message", "Authentication service unavailable: " + e.getMessage()
            );
            auditEventSender.failure(authToken.getActor(), SESSION_CREATE, auditEventContext);
            throw e;
        } catch (AuthenticationException e) {
            log.info("Invalid credentials in session create request. Actor: \"{}\"", authToken.getActor().urn());
            final Map<String, Object> auditEventContext = ImmutableMap.of(
                    "remote_address", host
            );
            auditEventSender.failure(authToken.getActor(), SESSION_CREATE, auditEventContext);
            return Optional.empty();
        }
    }
@Test
    public void serviceUnavailableStateIsCleared() {
        setUpUserMock();

        assertFalse(SecurityUtils.getSubject().isAuthenticated());

        final AtomicBoolean doThrow = new AtomicBoolean(true);
        final SimpleAccountRealm switchableRealm = new SimpleAccountRealm() {
            @Override
            protected AuthenticationInfo doGetAuthenticationInfo(
                    AuthenticationToken token) throws AuthenticationException {
                if (doThrow.get()) {
                    throw new AuthenticationServiceUnavailableException("not available");
                } else {
                    return super.doGetAuthenticationInfo(token);
                }
            }
        };

        securityManager.setRealms(ImmutableList.of(switchableRealm, new SimpleAccountRealm()));

        // realm will throw an exception on auth attempt
        assertThatThrownBy(() -> sessionCreator.login(null, "host", validToken)).isInstanceOf(
                AuthenticationServiceUnavailableException.class);
        assertThat(SecurityUtils.getSubject().isAuthenticated()).isFalse();

        // switch realm to not throw an exception but simply reject the credentials
        doThrow.set(false);

        sessionCreator.login(null, "host", validToken);
        assertThat(SecurityUtils.getSubject().isAuthenticated()).isFalse();
    }