Exception Handling (#467)

* Added consistent and better exception handling,
Added reactive repository integration test.
Fixed few API bugs

* Updated innerException to CosmosClientException

src/main/java/com/microsoft/azure/spring/data/cosmosdb/common/CosmosdbUtils.java

@@ -9,6 +9,7 @@
 import com.azure.data.cosmos.CosmosResponseDiagnostics;
 import com.azure.data.cosmos.FeedResponse;
 import com.azure.data.cosmos.FeedResponseDiagnostics;
+import com.azure.data.cosmos.Resource;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.microsoft.azure.spring.data.cosmosdb.core.ResponseDiagnostics;
 import com.microsoft.azure.spring.data.cosmosdb.core.ResponseDiagnosticsProcessor;
@@ -34,8 +35,9 @@ public static <T> T getCopyFrom(@NonNull T instance) {
         }
     }
 
-    public static void fillAndProcessResponseDiagnostics(ResponseDiagnosticsProcessor responseDiagnosticsProcessor,
-                                                         CosmosResponse cosmosResponse, FeedResponse feedResponse) {
+    public static <T extends Resource> void fillAndProcessResponseDiagnostics(
+        ResponseDiagnosticsProcessor responseDiagnosticsProcessor,
+        CosmosResponse<T> cosmosResponse, FeedResponse<T> feedResponse) {
         if (responseDiagnosticsProcessor == null) {
             return;
         }

src/main/java/com/microsoft/azure/spring/data/cosmosdb/exception/CosmosDBAccessException.java

@@ -5,15 +5,45 @@
  */
 package com.microsoft.azure.spring.data.cosmosdb.exception;
 
+import com.azure.data.cosmos.CosmosClientException;
 import org.springframework.dao.DataAccessException;
 import org.springframework.lang.Nullable;
 
+/**
+ * Public class extending DataAccessException, exposes innerException.
+ * Every API in {@link com.microsoft.azure.spring.data.cosmosdb.repository.CosmosRepository}
+ * and {@link com.microsoft.azure.spring.data.cosmosdb.repository.ReactiveCosmosRepository}
+ * should throw {@link CosmosDBAccessException}.
+ * innerException refers to the exception thrown by CosmosDB SDK. Callers of repository APIs can
+ * rely on innerException for any retriable logic, or for more details on the failure of
+ * the operation.
+ */
 public class CosmosDBAccessException extends DataAccessException {
+
+    protected final CosmosClientException cosmosClientException;
+
     public CosmosDBAccessException(String msg) {
         super(msg);
+        this.cosmosClientException = null;
     }
 
     public CosmosDBAccessException(@Nullable String msg, @Nullable Throwable cause) {
         super(msg, cause);
+        if (cause instanceof CosmosClientException) {
+            this.cosmosClientException = (CosmosClientException) cause;
+        } else {
+            this.cosmosClientException = null;
+        }
+    }
+
+    public CosmosDBAccessException(@Nullable String msg, @Nullable Exception cause) {
+        super(msg, cause);
+        this.cosmosClientException = cause instanceof CosmosClientException
+            ? (CosmosClientException) cause
+            : null;
+    }
+
+    public CosmosClientException getCosmosClientException() {
+        return cosmosClientException;
     }
 }

src/main/java/com/microsoft/azure/spring/data/cosmosdb/exception/CosmosDBExceptionUtils.java

@@ -0,0 +1,36 @@
+/**
+ * Copyright (c) Microsoft Corporation. All rights reserved.
+ * Licensed under the MIT License. See LICENSE in the project root for
+ * license information.
+ */
+package com.microsoft.azure.spring.data.cosmosdb.exception;
+
+import com.azure.data.cosmos.CosmosClientException;
+import com.azure.data.cosmos.internal.HttpConstants;
+import org.springframework.util.StringUtils;
+import reactor.core.Exceptions;
+import reactor.core.publisher.Mono;
+
+public class CosmosDBExceptionUtils {
+
+    public static <T> Mono<T> exceptionHandler(String message, Throwable throwable) {
+        if (StringUtils.isEmpty(message)) {
+            message = "Failed to access cosmosdb database";
+        }
+        //  Unwrap the exception in case if it is a reactive exception
+        final Throwable unwrappedThrowable = Exceptions.unwrap(throwable);
+        throw new CosmosDBAccessException(message, unwrappedThrowable);
+    }
+
+    public static <T> Mono<T> findAPIExceptionHandler(String message, Throwable throwable) {
+        //  Unwrap the exception in case if it is a reactive exception
+        final Throwable unwrappedThrowable = Exceptions.unwrap(throwable);
+        if (unwrappedThrowable instanceof CosmosClientException) {
+            final CosmosClientException cosmosClientException = (CosmosClientException) unwrappedThrowable;
+            if (cosmosClientException.statusCode() == HttpConstants.StatusCodes.NOTFOUND) {
+                return Mono.empty();
+            }
+        }
+        return exceptionHandler(message, unwrappedThrowable);
+    }
+}

src/test/java/com/microsoft/azure/spring/data/cosmosdb/core/CosmosTemplateIT.java

@@ -223,8 +223,8 @@ public void testOptimisticLockWhenUpdatingWithWrongEtag() {
         try {
             cosmosTemplate.upsert(Person.class.getSimpleName(), updated, null);
         } catch (CosmosDBAccessException e) {
-            assertThat(e.getCause()).isNotNull();
-            final Throwable cosmosClientException = e.getCause().getCause();
+            assertThat(e.getCosmosClientException()).isNotNull();
+            final Throwable cosmosClientException = e.getCosmosClientException();
             assertThat(cosmosClientException).isInstanceOf(CosmosClientException.class);
             assertThat(cosmosClientException.getMessage()).contains(PRECONDITION_IS_NOT_MET);
 

src/test/java/com/microsoft/azure/spring/data/cosmosdb/domain/Course.java

@@ -0,0 +1,30 @@
+/**
+ * Copyright (c) Microsoft Corporation. All rights reserved.
+ * Licensed under the MIT License. See LICENSE in the project root for
+ * license information.
+ */
+package com.microsoft.azure.spring.data.cosmosdb.domain;
+
+import com.microsoft.azure.spring.data.cosmosdb.core.mapping.Document;
+import com.microsoft.azure.spring.data.cosmosdb.core.mapping.PartitionKey;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+import org.springframework.data.annotation.Id;
+
+@Data
+@Document
+@AllArgsConstructor
+@NoArgsConstructor
+@Getter
+@Setter
+public class Course {
+
+    @Id
+    private String courseId;
+    private String name;
+    @PartitionKey
+    private String department;
+}

src/test/java/com/microsoft/azure/spring/data/cosmosdb/repository/TestRepositoryConfig.java

@@ -11,6 +11,7 @@
 import com.microsoft.azure.spring.data.cosmosdb.config.AbstractCosmosConfiguration;
 import com.microsoft.azure.spring.data.cosmosdb.config.CosmosDBConfig;
 import com.microsoft.azure.spring.data.cosmosdb.repository.config.EnableCosmosRepositories;
+import com.microsoft.azure.spring.data.cosmosdb.repository.config.EnableReactiveCosmosRepositories;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -20,6 +21,7 @@
 @Configuration
 @PropertySource(value = {"classpath:application.properties"})
 @EnableCosmosRepositories
+@EnableReactiveCosmosRepositories
 public class TestRepositoryConfig extends AbstractCosmosConfiguration {
     @Value("${cosmosdb.uri:}")
     private String cosmosDbUri;

src/test/java/com/microsoft/azure/spring/data/cosmosdb/repository/integration/AddressRepositoryIT.java

@@ -10,7 +10,6 @@
 import com.microsoft.azure.spring.data.cosmosdb.common.TestUtils;
 import com.microsoft.azure.spring.data.cosmosdb.core.CosmosTemplate;
 import com.microsoft.azure.spring.data.cosmosdb.domain.Address;
-import com.microsoft.azure.spring.data.cosmosdb.exception.CosmosDBAccessException;
 import com.microsoft.azure.spring.data.cosmosdb.repository.TestRepositoryConfig;
 import com.microsoft.azure.spring.data.cosmosdb.repository.repository.AddressRepository;
 import com.microsoft.azure.spring.data.cosmosdb.repository.support.CosmosEntityInformation;

src/test/java/com/microsoft/azure/spring/data/cosmosdb/repository/integration/ProjectRepositorySortIT.java

@@ -136,7 +136,7 @@ public void testFindAllSortMoreThanOneOrderException() {
         this.repository.findAll(sort);
     }
 
-    @Test(expected = CosmosDBAccessException.class)
+    @Test(expected = IllegalArgumentException.class)
     public void testFindAllSortIgnoreCaseException() {
         final Sort.Order order = Sort.Order.by("name").ignoreCase();
         final Sort sort = Sort.by(order);

src/test/java/com/microsoft/azure/spring/data/cosmosdb/repository/integration/ReactiveCourseRepositoryIT.java

@@ -0,0 +1,197 @@
+/**
+ * Copyright (c) Microsoft Corporation. All rights reserved.
+ * Licensed under the MIT License. See LICENSE in the project root for
+ * license information.
+ */
+package com.microsoft.azure.spring.data.cosmosdb.repository.integration;
+
+import com.azure.data.cosmos.PartitionKey;
+import com.microsoft.azure.spring.data.cosmosdb.core.ReactiveCosmosTemplate;
+import com.microsoft.azure.spring.data.cosmosdb.domain.Course;
+import com.microsoft.azure.spring.data.cosmosdb.exception.CosmosDBAccessException;
+import com.microsoft.azure.spring.data.cosmosdb.repository.TestRepositoryConfig;
+import com.microsoft.azure.spring.data.cosmosdb.repository.repository.ReactiveCourseRepository;
+import com.microsoft.azure.spring.data.cosmosdb.repository.support.CosmosEntityInformation;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.domain.Sort;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+import javax.annotation.PreDestroy;
+import java.util.Arrays;
+import java.util.Collections;
+
+@RunWith(SpringJUnit4ClassRunner.class)
+@ContextConfiguration(classes = TestRepositoryConfig.class)
+public class ReactiveCourseRepositoryIT {
+
+    private static final String COURSE_ID_1 = "1";
+    private static final String COURSE_ID_2 = "2";
+    private static final String COURSE_ID_3 = "3";
+    private static final String COURSE_ID_4 = "4";
+    private static final String COURSE_ID_5 = "5";
+
+    private static final String COURSE_NAME_1 = "Course1";
+    private static final String COURSE_NAME_2 = "Course2";
+    private static final String COURSE_NAME_3 = "Course3";
+    private static final String COURSE_NAME_4 = "Course4";
+    private static final String COURSE_NAME_5 = "Course5";
+
+    private static final String DEPARTMENT_NAME_1 = "Department1";
+    private static final String DEPARTMENT_NAME_2 = "Department2";
+    private static final String DEPARTMENT_NAME_3 = "Department3";
+
+    private static final Course COURSE_1 = new Course(COURSE_ID_1, COURSE_NAME_1, DEPARTMENT_NAME_3);
+    private static final Course COURSE_2 = new Course(COURSE_ID_2, COURSE_NAME_2, DEPARTMENT_NAME_2);
+    private static final Course COURSE_3 = new Course(COURSE_ID_3, COURSE_NAME_3, DEPARTMENT_NAME_2);
+    private static final Course COURSE_4 = new Course(COURSE_ID_4, COURSE_NAME_4, DEPARTMENT_NAME_1);
+    private static final Course COURSE_5 = new Course(COURSE_ID_5, COURSE_NAME_5, DEPARTMENT_NAME_1);
+
+    private final CosmosEntityInformation<Course, String> entityInformation =
+        new CosmosEntityInformation<>(Course.class);
+
+    @Autowired
+    private ReactiveCosmosTemplate template;
+
+    @Autowired
+    private ReactiveCourseRepository repository;
+
+    @PreDestroy
+    public void cleanUpCollection() {
+        template.deleteContainer(entityInformation.getCollectionName());
+    }
+
+    @Before
+    public void setup() {
+        final Flux<Course> savedFlux = repository.saveAll(Arrays.asList(COURSE_1, COURSE_2,
+            COURSE_3, COURSE_4));
+        StepVerifier.create(savedFlux).thenConsumeWhile(course -> true).expectComplete().verify();
+    }
+
+    @After
+    public void cleanup() {
+        final Mono<Void> deletedMono = repository.deleteAll();
+        StepVerifier.create(deletedMono).thenAwait().verifyComplete();
+    }
+
+    @Test
+    public void testFindById() {
+        final Mono<Course> idMono = repository.findById(COURSE_ID_4);
+        StepVerifier.create(idMono).expectNext(COURSE_4).expectComplete().verify();
+    }
+
+    @Test
+    public void testFindByIdAndPartitionKey() {
+        final Mono<Course> idMono = repository.findById(COURSE_ID_4,
+            new PartitionKey(entityInformation.getPartitionKeyFieldValue(COURSE_4)));
+        StepVerifier.create(idMono).expectNext(COURSE_4).expectComplete().verify();
+    }
+
+    @Test
+    public void testFindByIdAsPublisher() {
+        final Mono<Course> byId = repository.findById(Mono.just(COURSE_ID_1));
+        StepVerifier.create(byId).expectNext(COURSE_1).verifyComplete();
+    }
+
+    @Test
+    public void testFindAllWithSort() {
+        final Flux<Course> sortAll = repository.findAll(Sort.by(Sort.Order.desc("name")));
+        StepVerifier.create(sortAll).expectNext(COURSE_4, COURSE_3, COURSE_2, COURSE_1).verifyComplete();
+    }
+
+    @Test
+    public void testFindByIdNotFound() {
+        final Mono<Course> idMono = repository.findById("10");
+        //  Expect an empty mono as return value
+        StepVerifier.create(idMono).expectComplete().verify();
+    }
+
+    @Test
+    public void testFindByIdAndPartitionKeyNotFound() {
+        final Mono<Course> idMono = repository.findById("10",
+            new PartitionKey(entityInformation.getPartitionKeyFieldValue(COURSE_1)));
+        //  Expect an empty mono as return value
+        StepVerifier.create(idMono).expectComplete().verify();
+    }
+
+    @Test
+    public void testFindAll() {
+        final Flux<Course> allFlux = repository.findAll();
+        StepVerifier.create(allFlux).expectNextCount(4).verifyComplete();
+    }
+
+    @Test
+    public void testInsert() {
+        final Mono<Course> save = repository.save(COURSE_5);
+        StepVerifier.create(save).expectNext(COURSE_5).verifyComplete();
+    }
+
+    @Test
+    public void testUpsert() {
+        Mono<Course> save = repository.save(COURSE_1);
+        StepVerifier.create(save).expectNext(COURSE_1).expectComplete().verify();
+
+        save = repository.save(COURSE_1);
+        StepVerifier.create(save).expectNext(COURSE_1).expectComplete().verify();
+    }
+
+    @Test
+    public void testDeleteByIdWithoutPartitionKey() {
+        final Mono<Void> deleteMono = repository.deleteById(COURSE_1.getCourseId());
+        StepVerifier.create(deleteMono).expectError(CosmosDBAccessException.class).verify();
+    }
+
+    @Test
+    public void testDeleteByIdAndPartitionKey() {
+        final Mono<Void> deleteMono = repository.deleteById(COURSE_1.getCourseId(),
+            new PartitionKey(entityInformation.getPartitionKeyFieldValue(COURSE_1)));
+        StepVerifier.create(deleteMono).verifyComplete();
+
+        final Mono<Course> byId = repository.findById(COURSE_ID_1,
+            new PartitionKey(entityInformation.getPartitionKeyFieldValue(COURSE_1)));
+        //  Expect an empty mono as return value
+        StepVerifier.create(byId).verifyComplete();
+    }
+
+    @Test
+    public void testDeleteByEntity() {
+        final Mono<Void> deleteMono = repository.delete(COURSE_4);
+        StepVerifier.create(deleteMono).verifyComplete();
+
+        final Mono<Course> byId = repository.findById(COURSE_ID_4);
+        //  Expect an empty mono as return value
+        StepVerifier.create(byId).expectComplete().verify();
+    }
+
+    @Test
+    public void testDeleteByIdNotFound() {
+        final Mono<Void> deleteMono = repository.deleteById(COURSE_ID_5);
+        StepVerifier.create(deleteMono).expectError(CosmosDBAccessException.class).verify();
+    }
+
+    @Test
+    public void testDeleteByEntityNotFound() {
+        final Mono<Void> deleteMono = repository.delete(COURSE_5);
+        StepVerifier.create(deleteMono).expectError(CosmosDBAccessException.class).verify();
+    }
+
+    @Test
+    public void testCountAll() {
+        final Mono<Long> countMono = repository.count();
+        StepVerifier.create(countMono).expectNext(4L).verifyComplete();
+    }
+
+    @Test
+    public void testFindByDepartmentIn() {
+        final Flux<Course> byDepartmentIn =
+            repository.findByDepartmentIn(Collections.singletonList(DEPARTMENT_NAME_2));
+        StepVerifier.create(byDepartmentIn).expectNextCount(2).verifyComplete();
+    }
+}

src/test/java/com/microsoft/azure/spring/data/cosmosdb/repository/repository/ReactiveCourseRepository.java

@@ -0,0 +1,17 @@
+/**
+ * Copyright (c) Microsoft Corporation. All rights reserved.
+ * Licensed under the MIT License. See LICENSE in the project root for
+ * license information.
+ */
+package com.microsoft.azure.spring.data.cosmosdb.repository.repository;
+
+import com.microsoft.azure.spring.data.cosmosdb.domain.Course;
+import com.microsoft.azure.spring.data.cosmosdb.repository.ReactiveCosmosRepository;
+import reactor.core.publisher.Flux;
+
+import java.util.Collection;
+
+public interface ReactiveCourseRepository extends ReactiveCosmosRepository<Course, String> {
+
+    Flux<Course> findByDepartmentIn(Collection<String> departments);
+}