Improve resilience of authentication chain.

- Synchronous or asynchronously thrown exceptions no longer abort
  the authentication chain.

- Fixed concurrent modification of authentication result.

Signed-off-by: Yufei Cai <yufei.cai@bosch.io>

gateway/service/src/main/java/org/eclipse/ditto/gateway/service/security/authentication/AuthenticationChain.java

@@ -17,10 +17,13 @@
 
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executor;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import javax.annotation.Nullable;
 import javax.annotation.concurrent.Immutable;
@@ -122,7 +125,7 @@ private AuthResultAccumulator(
                 final RequestContext requestContext,
                 final DittoHeaders dittoHeaders) {
             this.successResult = successResult;
-            this.failureResults = failureResults;
+            this.failureResults = Collections.unmodifiableList(failureResults);
             this.requestContext = requestContext;
             this.dittoHeaders = dittoHeaders;
         }
@@ -135,11 +138,18 @@ private AuthResultAccumulator appendResult(final AuthenticationProvider<?> authe
                 return new AuthResultAccumulator(nextResult, failureResults, requestContext, dittoHeaders);
             } else {
                 logFailure(authenticationProvider, nextResult);
-                failureResults.add(nextResult);
-                return new AuthResultAccumulator(successResult, failureResults, requestContext, dittoHeaders);
+                final var newFailureResults =
+                        Stream.concat(failureResults.stream(), Stream.of(nextResult)).collect(Collectors.toList());
+                return new AuthResultAccumulator(successResult, newFailureResults, requestContext, dittoHeaders);
             }
         }
 
+        private AuthResultAccumulator appendFailure(final AuthenticationProvider<?> provider,
+                final Throwable throwable) {
+
+            return appendResult(provider, DefaultAuthenticationResult.failed(dittoHeaders, throwable));
+        }
+
         private void logSuccess(final AuthenticationProvider<?> provider) {
             if (LOGGER.isDebugEnabled()) {
                 LOGGER.withCorrelationId(dittoHeaders)
@@ -164,18 +174,26 @@ private CompletableFuture<AuthResultAccumulator> andThen(final CompletableFuture
                 return CompletableFuture.completedFuture(this);
             } else {
                 return other.thenApplyAsync(that -> {
-                    failureResults.addAll(that.failureResults);
-                    return new AuthResultAccumulator(that.successResult, failureResults, requestContext, dittoHeaders);
+                    final var newFailureResults =
+                            Stream.concat(failureResults.stream(), that.failureResults.stream())
+                                    .collect(Collectors.toList());
+                    return new AuthResultAccumulator(that.successResult, newFailureResults, requestContext,
+                            dittoHeaders);
                 }, authenticationDispatcher);
             }
         }
 
         private CompletableFuture<AuthResultAccumulator> andThen(
                 final AuthenticationProvider<?> authenticationProvider) {
             if (successResult == null && authenticationProvider.isApplicable(requestContext)) {
-                return authenticationProvider.authenticate(requestContext, dittoHeaders)
-                        .thenApplyAsync(result -> appendResult(authenticationProvider, result),
-                                authenticationDispatcher);
+                try {
+                    return authenticationProvider.authenticate(requestContext, dittoHeaders)
+                            .thenApplyAsync(result -> appendResult(authenticationProvider, result),
+                                    authenticationDispatcher)
+                            .exceptionally(e -> appendFailure(authenticationProvider, e));
+                } catch (final Throwable e) {
+                    return CompletableFuture.completedFuture(appendFailure(authenticationProvider, e));
+                }
             } else {
                 return CompletableFuture.completedFuture(this);
             }

gateway/service/src/main/java/org/eclipse/ditto/gateway/service/security/authentication/DefaultAuthenticationFailureAggregator.java

@@ -53,7 +53,8 @@ static DefaultAuthenticationFailureAggregator getInstance() {
 
     @Override
     public DittoRuntimeException aggregateAuthenticationFailures(final List<AuthenticationResult> failedAuthResults) {
-        final List<DittoRuntimeException> reasonsOfFailure = getDittoRuntimeExceptionReasonsWithDescription(failedAuthResults);
+        final List<DittoRuntimeException> reasonsOfFailure =
+                getDittoRuntimeExceptionReasonsWithDescription(failedAuthResults);
 
         if (reasonsOfFailure.isEmpty()) {
             final String msgPattern = "The failed authentication results did not contain any failure reason of type " +

gateway/service/src/test/java/org/eclipse/ditto/gateway/service/security/authentication/AuthenticationChainTest.java

@@ -12,7 +12,10 @@
  */
 package org.eclipse.ditto.gateway.service.security.authentication;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
@@ -33,6 +36,7 @@
 import org.eclipse.ditto.base.model.exceptions.DittoRuntimeException;
 import org.eclipse.ditto.base.model.headers.DittoHeaders;
 import org.eclipse.ditto.base.model.signals.commands.exceptions.GatewayAuthenticationFailedException;
+import org.eclipse.ditto.base.model.signals.commands.exceptions.GatewayServiceUnavailableException;
 import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.Rule;
@@ -186,6 +190,59 @@ public void authenticateReturnsFailedAuthenticationResult() throws ExecutionExce
         softly.assertThat(authenticationResult).isEqualTo(expectedAuthenticationResult);
     }
 
+    @Test
+    public void synchronousExceptionDoesNotAbortAuthenticationChain() {
+        final RequestContext requestContextMock = mock(RequestContext.class);
+        final DittoRuntimeException error = GatewayServiceUnavailableException.newBuilder().build();
+        final var expectedAuthenticationResult =
+                DefaultAuthenticationResult.successful(dittoHeaders, knownAuthorizationContext);
+        when(authenticationProviderA.isApplicable(any())).thenReturn(true);
+        when(authenticationProviderA.authenticate(any(), any())).thenThrow(error);
+        when(authenticationProviderB.isApplicable(any())).thenReturn(true);
+        when(authenticationProviderB.authenticate(any(), any()))
+                .thenReturn(CompletableFuture.completedFuture(expectedAuthenticationResult));
+        final AuthenticationChain underTest =
+                AuthenticationChain.getInstance(Arrays.asList(authenticationProviderA, authenticationProviderB),
+                        authenticationFailureAggregator,
+                        messageDispatcher);
+
+        final AuthenticationResult authenticationResult =
+                underTest.authenticate(requestContextMock, dittoHeaders).join();
+
+        assertThat(authenticationResult).isEqualTo(expectedAuthenticationResult);
+        verify(authenticationProviderA).isApplicable(eq(requestContextMock));
+        verify(authenticationProviderB).isApplicable(eq(requestContextMock));
+        verify(authenticationProviderA).authenticate(eq(requestContextMock), eq(dittoHeaders));
+        verify(authenticationProviderB).authenticate(eq(requestContextMock), eq(dittoHeaders));
+    }
+
+    @Test
+    public void asynchronousExceptionDoesNotAbortAuthenticationChain() {
+        final RequestContext requestContextMock = mock(RequestContext.class);
+        final DittoRuntimeException error = GatewayServiceUnavailableException.newBuilder().build();
+        final var expectedAuthenticationResult =
+                DefaultAuthenticationResult.successful(dittoHeaders, knownAuthorizationContext);
+        when(authenticationProviderA.isApplicable(any())).thenReturn(true);
+        when(authenticationProviderA.authenticate(any(), any()))
+                .thenReturn(CompletableFuture.failedFuture(error));
+        when(authenticationProviderB.isApplicable(any())).thenReturn(true);
+        when(authenticationProviderB.authenticate(any(), any()))
+                .thenReturn(CompletableFuture.completedFuture(expectedAuthenticationResult));
+        final AuthenticationChain underTest =
+                AuthenticationChain.getInstance(Arrays.asList(authenticationProviderA, authenticationProviderB),
+                        authenticationFailureAggregator,
+                        messageDispatcher);
+
+        final AuthenticationResult authenticationResult =
+                underTest.authenticate(requestContextMock, dittoHeaders).join();
+
+        assertThat(authenticationResult).isEqualTo(expectedAuthenticationResult);
+        verify(authenticationProviderA).isApplicable(eq(requestContextMock));
+        verify(authenticationProviderB).isApplicable(eq(requestContextMock));
+        verify(authenticationProviderA).authenticate(eq(requestContextMock), eq(dittoHeaders));
+        verify(authenticationProviderB).authenticate(eq(requestContextMock), eq(dittoHeaders));
+    }
+
     @Test
     public void authenticateReturnsSuccessfulAuthenticationResultIfOneProviderSucceeds()
             throws ExecutionException, InterruptedException {