From 346e626f80e956db4a623209f43d0d1780235e81 Mon Sep 17 00:00:00 2001
From: mikereiche <michael.reiche@couchbase.com>
Date: Wed, 24 Mar 2021 10:52:29 -0400
Subject: [PATCH] Use queryScanConsistency on reactive deleteAll().

It was present on non-Reactive, but missing from reactive.

Closes #1096.
Original pull request #1108.
---
 .../SimpleReactiveCouchbaseRepository.java    |  6 +++---
 .../domain/ReactiveAirportRepository.java     |  4 ++++
 ...chbaseRepositoryQueryIntegrationTests.java | 19 +++++++++++++++++++
 3 files changed, 26 insertions(+), 3 deletions(-)

diff --git a/src/main/java/org/springframework/data/couchbase/repository/support/SimpleReactiveCouchbaseRepository.java b/src/main/java/org/springframework/data/couchbase/repository/support/SimpleReactiveCouchbaseRepository.java
index d6c69ef71..edb8ab0f4 100644
--- a/src/main/java/org/springframework/data/couchbase/repository/support/SimpleReactiveCouchbaseRepository.java
+++ b/src/main/java/org/springframework/data/couchbase/repository/support/SimpleReactiveCouchbaseRepository.java
@@ -16,7 +16,7 @@
 
 package org.springframework.data.couchbase.repository.support;
 
-import static org.springframework.data.couchbase.repository.support.Util.*;
+import static org.springframework.data.couchbase.repository.support.Util.hasNonZeroVersionProperty;
 
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -26,7 +26,6 @@
 import java.util.stream.Collectors;
 
 import org.reactivestreams.Publisher;
-
 import org.springframework.data.couchbase.core.CouchbaseOperations;
 import org.springframework.data.couchbase.core.ReactiveCouchbaseOperations;
 import org.springframework.data.couchbase.core.query.Query;
@@ -189,7 +188,8 @@ public Mono<Long> count() {
 
 	@Override
 	public Mono<Void> deleteAll() {
-		return operations.removeByQuery(entityInformation.getJavaType()).all().then();
+		return operations.removeByQuery(entityInformation.getJavaType()).withConsistency(buildQueryScanConsistency()).all()
+				.then();
 	}
 
 	/**
diff --git a/src/test/java/org/springframework/data/couchbase/domain/ReactiveAirportRepository.java b/src/test/java/org/springframework/data/couchbase/domain/ReactiveAirportRepository.java
index 63cb3388d..51c4d48cc 100644
--- a/src/test/java/org/springframework/data/couchbase/domain/ReactiveAirportRepository.java
+++ b/src/test/java/org/springframework/data/couchbase/domain/ReactiveAirportRepository.java
@@ -45,6 +45,10 @@ public interface ReactiveAirportRepository extends ReactiveSortingRepository<Air
 	@ScanConsistency(query = QueryScanConsistency.REQUEST_PLUS)
 	Flux<Airport> findAll();
 
+	@Override
+	@ScanConsistency(query = QueryScanConsistency.REQUEST_PLUS)
+	Mono<Void> deleteAll();
+
 	@Override
 	Mono<Airport> save(Airport a);
 
diff --git a/src/test/java/org/springframework/data/couchbase/repository/ReactiveCouchbaseRepositoryQueryIntegrationTests.java b/src/test/java/org/springframework/data/couchbase/repository/ReactiveCouchbaseRepositoryQueryIntegrationTests.java
index d6b582f7a..1c2d0d9bd 100644
--- a/src/test/java/org/springframework/data/couchbase/repository/ReactiveCouchbaseRepositoryQueryIntegrationTests.java
+++ b/src/test/java/org/springframework/data/couchbase/repository/ReactiveCouchbaseRepositoryQueryIntegrationTests.java
@@ -195,6 +195,25 @@ void deleteAllById() {
 		}
 	}
 
+	@Test
+	void deleteAll() {
+
+		Airport vienna = new Airport("airports::vie", "vie", "LOWW");
+		Airport frankfurt = new Airport("airports::fra", "fra", "EDDF");
+		Airport losAngeles = new Airport("airports::lax", "lax", "KLAX");
+
+		try {
+			airportRepository.saveAll(asList(vienna, frankfurt, losAngeles)).as(StepVerifier::create)
+					.expectNext(vienna, frankfurt, losAngeles).verifyComplete();
+
+			airportRepository.deleteAll().as(StepVerifier::create).verifyComplete();
+
+			airportRepository.findAll().as(StepVerifier::create).verifyComplete();
+		} finally {
+			airportRepository.deleteAll().block();
+		}
+	}
+
 	@Configuration
 	@EnableReactiveCouchbaseRepositories("org.springframework.data.couchbase")
 	static class Config extends AbstractCouchbaseConfiguration {