Specifying withExpiry(Duration) on findById() uses getTouchAndRead(). (#1196)

Closes #982.

src/main/java/org/springframework/data/couchbase/core/ExecutableFindByIdOperation.java

@@ -15,6 +15,7 @@
  */
 package org.springframework.data.couchbase.core;
 
+import java.time.Duration;
 import java.util.Collection;
 
 import org.springframework.data.couchbase.core.support.OneAndAllId;
@@ -24,6 +25,7 @@ import org.springframework.data.couchbase.core.support.WithProjectionId;
 import org.springframework.data.couchbase.core.support.InScope;
 
 import com.couchbase.client.java.kv.GetOptions;
+import org.springframework.data.couchbase.core.support.WithExpiry;
 
 /**
  * Get Operations
@@ -120,11 +122,21 @@ public interface ExecutableFindByIdOperation {
 		FindByIdInScope<T> project(String... fields);
 	}
 
+	interface FindByIdWithExpiry<T> extends FindByIdWithProjection<T>, WithExpiry<T> {
+		/**
+		 * Load only certain fields for the document.
+		 *
+		 * @param expiry the projected fields to load.
+		 */
+		@Override
+		FindByIdWithProjection<T> withExpiry(Duration expiry);
+	}
+
 	/**
 	 * Provides methods for constructing query operations in a fluent way.
 	 *
 	 * @param <T> the entity type to use for the results
 	 */
-	interface ExecutableFindById<T> extends FindByIdWithProjection<T> {}
+	interface ExecutableFindById<T> extends FindByIdWithExpiry<T> {}
 
 }

src/main/java/org/springframework/data/couchbase/core/ExecutableFindByIdOperationSupport.java

@@ -15,6 +15,7 @@
  */
 package org.springframework.data.couchbase.core;
 
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
@@ -34,7 +35,7 @@ public class ExecutableFindByIdOperationSupport implements ExecutableFindByIdOpe
 
 	@Override
 	public <T> ExecutableFindById<T> findById(Class<T> domainType) {
-		return new ExecutableFindByIdSupport<>(template, domainType, null, null, null, null);
+		return new ExecutableFindByIdSupport<>(template, domainType, null, null, null, null, null);
 	}
 
 	static class ExecutableFindByIdSupport<T> implements ExecutableFindById<T> {
@@ -45,18 +46,20 @@ public class ExecutableFindByIdOperationSupport implements ExecutableFindByIdOpe
 		private final String collection;
 		private final GetOptions options;
 		private final List<String> fields;
+		private final Duration expiry;
 		private final ReactiveFindByIdSupport<T> reactiveSupport;
 
 		ExecutableFindByIdSupport(CouchbaseTemplate template, Class<T> domainType, String scope, String collection,
-				GetOptions options, List<String> fields) {
+				GetOptions options, List<String> fields, Duration expiry) {
 			this.template = template;
 			this.domainType = domainType;
 			this.scope = scope;
 			this.collection = collection;
 			this.options = options;
 			this.fields = fields;
+			this.expiry = expiry;
 			this.reactiveSupport = new ReactiveFindByIdSupport<>(template.reactive(), domainType, scope, collection, options,
-					fields, new NonReactiveSupportWrapper(template.support()));
+					fields, expiry, new NonReactiveSupportWrapper(template.support()));
 		}
 
 		@Override
@@ -72,23 +75,29 @@ public class ExecutableFindByIdOperationSupport implements ExecutableFindByIdOpe
 		@Override
 		public TerminatingFindById<T> withOptions(final GetOptions options) {
 			Assert.notNull(options, "Options must not be null.");
-			return new ExecutableFindByIdSupport<>(template, domainType, scope, collection, options, fields);
+			return new ExecutableFindByIdSupport<>(template, domainType, scope, collection, options, fields, expiry);
 		}
 
 		@Override
 		public FindByIdWithOptions<T> inCollection(final String collection) {
-			return new ExecutableFindByIdSupport<>(template, domainType, scope, collection, options, fields);
+			return new ExecutableFindByIdSupport<>(template, domainType, scope, collection, options, fields, expiry);
 		}
 
 		@Override
 		public FindByIdInCollection<T> inScope(final String scope) {
-			return new ExecutableFindByIdSupport<>(template, domainType, scope, collection, options, fields);
+			return new ExecutableFindByIdSupport<>(template, domainType, scope, collection, options, fields, expiry);
 		}
 
 		@Override
 		public FindByIdInScope<T> project(String... fields) {
 			Assert.notEmpty(fields, "Fields must not be null.");
-			return new ExecutableFindByIdSupport<>(template, domainType, scope, collection, options, Arrays.asList(fields));
+			return new ExecutableFindByIdSupport<>(template, domainType, scope, collection, options, Arrays.asList(fields), expiry);
+		}
+
+		@Override
+		public FindByIdWithProjection<T> withExpiry(final Duration expiry) {
+			return new ExecutableFindByIdSupport<>(template, domainType, scope, collection, options, fields,
+					expiry);
 		}
 
 	}

src/main/java/org/springframework/data/couchbase/core/ExecutableInsertByIdOperation.java

@@ -21,6 +21,8 @@ import java.util.Collection;
 import org.springframework.data.couchbase.core.support.InCollection;
 import org.springframework.data.couchbase.core.support.InScope;
 import org.springframework.data.couchbase.core.support.OneAndAllEntity;
+import org.springframework.data.couchbase.core.support.WithDurability;
+import org.springframework.data.couchbase.core.support.WithExpiry;
 import org.springframework.data.couchbase.core.support.WithInsertOptions;
 
 import com.couchbase.client.core.msg.kv.DurabilityLevel;

src/main/java/org/springframework/data/couchbase/core/ExecutableRemoveByIdOperation.java

@@ -21,6 +21,7 @@ import java.util.List;
 import org.springframework.data.couchbase.core.support.InCollection;
 import org.springframework.data.couchbase.core.support.InScope;
 import org.springframework.data.couchbase.core.support.OneAndAllId;
+import org.springframework.data.couchbase.core.support.WithDurability;
 import org.springframework.data.couchbase.core.support.WithRemoveOptions;
 
 import com.couchbase.client.core.msg.kv.DurabilityLevel;

src/main/java/org/springframework/data/couchbase/core/ExecutableReplaceByIdOperation.java

@@ -21,6 +21,8 @@ import java.util.Collection;
 import org.springframework.data.couchbase.core.support.InCollection;
 import org.springframework.data.couchbase.core.support.InScope;
 import org.springframework.data.couchbase.core.support.OneAndAllEntity;
+import org.springframework.data.couchbase.core.support.WithDurability;
+import org.springframework.data.couchbase.core.support.WithExpiry;
 import org.springframework.data.couchbase.core.support.WithReplaceOptions;
 
 import com.couchbase.client.core.msg.kv.DurabilityLevel;

src/main/java/org/springframework/data/couchbase/core/ExecutableUpsertByIdOperation.java

@@ -21,6 +21,8 @@ import java.util.Collection;
 import org.springframework.data.couchbase.core.support.InCollection;
 import org.springframework.data.couchbase.core.support.InScope;
 import org.springframework.data.couchbase.core.support.OneAndAllEntity;
+import org.springframework.data.couchbase.core.support.WithDurability;
+import org.springframework.data.couchbase.core.support.WithExpiry;
 import org.springframework.data.couchbase.core.support.WithUpsertOptions;
 
 import com.couchbase.client.core.msg.kv.DurabilityLevel;

src/main/java/org/springframework/data/couchbase/core/ReactiveFindByIdOperation.java

@@ -18,11 +18,13 @@ package org.springframework.data.couchbase.core;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import java.time.Duration;
 import java.util.Collection;
 
 import org.springframework.data.couchbase.core.support.InCollection;
 import org.springframework.data.couchbase.core.support.InScope;
 import org.springframework.data.couchbase.core.support.OneAndAllIdReactive;
+import org.springframework.data.couchbase.core.support.WithExpiry;
 import org.springframework.data.couchbase.core.support.WithGetOptions;
 import org.springframework.data.couchbase.core.support.WithProjectionId;
 
@@ -124,11 +126,21 @@ public interface ReactiveFindByIdOperation {
 
 	}
 
+	interface FindByIdWithExpiry<T> extends FindByIdWithProjection<T>, WithExpiry<T> {
+		/**
+		 * Load only certain fields for the document.
+		 *
+		 * @param expiry the projected fields to load.
+		 */
+		@Override
+		FindByIdWithProjection<T> withExpiry(Duration expiry);
+	}
+
 	/**
 	 * Provides methods for constructing query operations in a fluent way.
 	 *
 	 * @param <T> the entity type to use for the results
 	 */
-	interface ReactiveFindById<T> extends FindByIdWithProjection<T> {}
+	interface ReactiveFindById<T> extends FindByIdWithExpiry<T> {}
 
 }

src/main/java/org/springframework/data/couchbase/core/ReactiveFindByIdOperationSupport.java

@@ -15,22 +15,27 @@
  */
 package org.springframework.data.couchbase.core;
 
-import static com.couchbase.client.java.kv.GetOptions.getOptions;
+import static com.couchbase.client.java.kv.GetAndTouchOptions.getAndTouchOptions;
 
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.data.couchbase.core.mapping.CouchbasePersistentEntity;
 import org.springframework.data.couchbase.core.support.PseudoArgs;
 import org.springframework.util.Assert;
 
 import com.couchbase.client.core.error.DocumentNotFoundException;
+import com.couchbase.client.java.CommonOptions;
+import com.couchbase.client.java.ReactiveCollection;
 import com.couchbase.client.java.codec.RawJsonTranscoder;
+import com.couchbase.client.java.kv.GetAndTouchOptions;
 import com.couchbase.client.java.kv.GetOptions;
 
 public class ReactiveFindByIdOperationSupport implements ReactiveFindByIdOperation {
@@ -44,7 +49,7 @@ public class ReactiveFindByIdOperationSupport implements ReactiveFindByIdOperati
 
 	@Override
 	public <T> ReactiveFindById<T> findById(Class<T> domainType) {
-		return new ReactiveFindByIdSupport<>(template, domainType, null, null, null, null, template.support());
+		return new ReactiveFindByIdSupport<>(template, domainType, null, null, null, null, null, template.support());
 	}
 
 	static class ReactiveFindByIdSupport<T> implements ReactiveFindById<T> {
@@ -53,36 +58,39 @@ public class ReactiveFindByIdOperationSupport implements ReactiveFindByIdOperati
 		private final Class<T> domainType;
 		private final String scope;
 		private final String collection;
-		private final GetOptions options;
+		private final CommonOptions<?> options;
 		private final List<String> fields;
 		private final ReactiveTemplateSupport support;
+		private final Duration expiry;
 
 		ReactiveFindByIdSupport(ReactiveCouchbaseTemplate template, Class<T> domainType, String scope, String collection,
-				GetOptions options, List<String> fields, ReactiveTemplateSupport support) {
+				CommonOptions<?> options, List<String> fields, Duration expiry, ReactiveTemplateSupport support) {
 			this.template = template;
 			this.domainType = domainType;
 			this.scope = scope;
 			this.collection = collection;
 			this.options = options;
 			this.fields = fields;
+			this.expiry = expiry;
 			this.support = support;
 		}
 
 		@Override
 		public Mono<T> one(final String id) {
-			GetOptions gOptions = options != null ? options : getOptions();
-			if (gOptions.build().transcoder() == null) {
-				gOptions.transcoder(RawJsonTranscoder.INSTANCE);
-			}
-			if (fields != null && !fields.isEmpty()) {
-				gOptions.project(fields);
-			}
-			PseudoArgs<GetOptions> pArgs = new PseudoArgs(template, scope, collection, gOptions, domainType);
+
+			CommonOptions<?> gOptions = initGetOptions();
+			PseudoArgs<?> pArgs = new PseudoArgs(template, scope, collection, gOptions, domainType);
 			LOG.trace("findById {}", pArgs);
-			return Mono.just(id)
-					.flatMap(docId -> template.getCouchbaseClientFactory().withScope(pArgs.getScope())
-							.getCollection(pArgs.getCollection()).reactive().get(docId, pArgs.getOptions()))
-					.flatMap(result -> support.decodeEntity(id, result.contentAs(String.class), result.cas(), domainType))
+
+			return Mono.just(id).flatMap(docId -> {
+				ReactiveCollection reactive = template.getCouchbaseClientFactory().withScope(pArgs.getScope())
+						.getCollection(pArgs.getCollection()).reactive();
+				if (pArgs.getOptions() instanceof GetAndTouchOptions) {
+					return reactive.getAndTouch(docId, expiryToUse(), (GetAndTouchOptions) pArgs.getOptions());
+				} else {
+					return reactive.get(docId, (GetOptions) pArgs.getOptions());
+				}
+			}).flatMap(result -> support.decodeEntity(id, result.contentAs(String.class), result.cas(), domainType))
 					.onErrorResume(throwable -> {
 						if (throwable instanceof RuntimeException) {
 							if (throwable instanceof DocumentNotFoundException) {
@@ -107,24 +115,62 @@ public class ReactiveFindByIdOperationSupport implements ReactiveFindByIdOperati
 		@Override
 		public TerminatingFindById<T> withOptions(final GetOptions options) {
 			Assert.notNull(options, "Options must not be null.");
-			return new ReactiveFindByIdSupport<>(template, domainType, scope, collection, options, fields, support);
+			return new ReactiveFindByIdSupport<>(template, domainType, scope, collection, options, fields, expiry, support);
 		}
 
 		@Override
 		public FindByIdWithOptions<T> inCollection(final String collection) {
-			return new ReactiveFindByIdSupport<>(template, domainType, scope, collection, options, fields, support);
+			return new ReactiveFindByIdSupport<>(template, domainType, scope, collection, options, fields, expiry, support);
 		}
 
 		@Override
 		public FindByIdInCollection<T> inScope(final String scope) {
-			return new ReactiveFindByIdSupport<>(template, domainType, scope, collection, options, fields, support);
+			return new ReactiveFindByIdSupport<>(template, domainType, scope, collection, options, fields, expiry, support);
 		}
 
 		@Override
 		public FindByIdInScope<T> project(String... fields) {
 			Assert.notNull(fields, "Fields must not be null");
 			return new ReactiveFindByIdSupport<>(template, domainType, scope, collection, options, Arrays.asList(fields),
-					support);
+					expiry, support);
+		}
+
+		@Override
+		public FindByIdWithProjection<T> withExpiry(final Duration expiry) {
+			return new ReactiveFindByIdSupport<>(template, domainType, scope, collection, options, fields, expiry, support);
+		}
+
+		private CommonOptions<?> initGetOptions() {
+			CommonOptions<?> getOptions;
+			if (expiry != null || options instanceof GetAndTouchOptions) {
+				GetAndTouchOptions gOptions = options != null ? (GetAndTouchOptions) options : getAndTouchOptions();
+				if (gOptions.build().transcoder() == null) {
+					gOptions.transcoder(RawJsonTranscoder.INSTANCE);
+				}
+				getOptions = gOptions;
+			} else {
+				GetOptions gOptions = options != null ? (GetOptions) options : GetOptions.getOptions();
+				if (gOptions.build().transcoder() == null) {
+					gOptions.transcoder(RawJsonTranscoder.INSTANCE);
+				}
+				if (fields != null && !fields.isEmpty()) {
+					gOptions.project(fields);
+				}
+				getOptions = gOptions;
+			}
+			return getOptions;
+		}
+
+		private Duration expiryToUse() {
+			Duration expiryToUse = expiry;
+			if (expiryToUse != null || options instanceof GetAndTouchOptions) {
+				if (expiryToUse == null) { // GetAndTouchOptions without specifying expiry -> get expiry from annoation
+					final CouchbasePersistentEntity<?> entity = template.getConverter().getMappingContext()
+							.getRequiredPersistentEntity(domainType);
+					expiryToUse = Duration.ofSeconds(entity.getExpiry());
+				}
+			}
+			return expiry;
 		}
 	}
 

src/main/java/org/springframework/data/couchbase/core/ReactiveInsertByIdOperation.java

@@ -24,6 +24,8 @@ import java.util.Collection;
 import org.springframework.data.couchbase.core.support.InCollection;
 import org.springframework.data.couchbase.core.support.InScope;
 import org.springframework.data.couchbase.core.support.OneAndAllEntityReactive;
+import org.springframework.data.couchbase.core.support.WithDurability;
+import org.springframework.data.couchbase.core.support.WithExpiry;
 import org.springframework.data.couchbase.core.support.WithInsertOptions;
 
 import com.couchbase.client.core.msg.kv.DurabilityLevel;

src/main/java/org/springframework/data/couchbase/core/ReactiveRemoveByIdOperation.java

@@ -23,6 +23,7 @@ import java.util.Collection;
 import org.springframework.data.couchbase.core.support.InCollection;
 import org.springframework.data.couchbase.core.support.InScope;
 import org.springframework.data.couchbase.core.support.OneAndAllIdReactive;
+import org.springframework.data.couchbase.core.support.WithDurability;
 import org.springframework.data.couchbase.core.support.WithRemoveOptions;
 
 import com.couchbase.client.core.msg.kv.DurabilityLevel;

src/main/java/org/springframework/data/couchbase/core/ReactiveReplaceByIdOperation.java

@@ -24,6 +24,8 @@ import java.util.Collection;
 import org.springframework.data.couchbase.core.support.InCollection;
 import org.springframework.data.couchbase.core.support.InScope;
 import org.springframework.data.couchbase.core.support.OneAndAllEntityReactive;
+import org.springframework.data.couchbase.core.support.WithDurability;
+import org.springframework.data.couchbase.core.support.WithExpiry;
 import org.springframework.data.couchbase.core.support.WithReplaceOptions;
 
 import com.couchbase.client.core.msg.kv.DurabilityLevel;

src/main/java/org/springframework/data/couchbase/core/ReactiveUpsertByIdOperation.java

@@ -24,6 +24,8 @@ import java.util.Collection;
 import org.springframework.data.couchbase.core.support.InCollection;
 import org.springframework.data.couchbase.core.support.InScope;
 import org.springframework.data.couchbase.core.support.OneAndAllEntityReactive;
+import org.springframework.data.couchbase.core.support.WithDurability;
+import org.springframework.data.couchbase.core.support.WithExpiry;
 import org.springframework.data.couchbase.core.support.WithUpsertOptions;
 
 import com.couchbase.client.core.msg.kv.DurabilityLevel;
@@ -39,7 +41,6 @@ import com.couchbase.client.java.kv.UpsertOptions;
  */
 public interface ReactiveUpsertByIdOperation {
 
-
 	/**
 	 * Upsert using the KV service.
 	 *
@@ -118,6 +119,7 @@ public interface ReactiveUpsertByIdOperation {
 	interface UpsertByIdWithDurability<T> extends UpsertByIdInScope<T>, WithDurability<T> {
 		@Override
 		UpsertByIdInCollection<T> withDurability(DurabilityLevel durabilityLevel);
+
 		@Override
 		UpsertByIdInCollection<T> withDurability(PersistTo persistTo, ReplicateTo replicateTo);
 

src/main/java/org/springframework/data/couchbase/core/support/WithDurability.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020 the original author or authors
+ * Copyright 2020-2021 the original author or authors
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.springframework.data.couchbase.core;
+package org.springframework.data.couchbase.core.support;
 
 import com.couchbase.client.core.msg.kv.DurabilityLevel;
 import com.couchbase.client.java.kv.PersistTo;

src/main/java/org/springframework/data/couchbase/core/support/WithExpiry.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020 the original author or authors
+ * Copyright 2020-2021 the original author or authors
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -13,16 +13,17 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.springframework.data.couchbase.core;
+package org.springframework.data.couchbase.core.support;
 
 import java.time.Duration;
 
 /**
- * A common interface for all of Insert, Replace, Upsert that take expiry
+ * A common interface for those that support withExpiry()
  *
  * @author Michael Reiche
- * @param <T> - the entity class
+ * @param <R> - the entity class
  */
-public interface WithExpiry<T> {
+public interface WithExpiry<R> {
 	Object withExpiry(Duration expiry);
+
 }

src/test/java/org/springframework/data/couchbase/core/CouchbaseTemplateKeyValueIntegrationTests.java

@@ -28,6 +28,7 @@ import java.lang.reflect.Constructor;
 import java.lang.reflect.InvocationTargetException;
 import java.time.Duration;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
@@ -38,9 +39,13 @@ import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.springframework.dao.DataIntegrityViolationException;
 import org.springframework.dao.DuplicateKeyException;
+import org.springframework.data.couchbase.core.ExecutableFindByIdOperation.ExecutableFindById;
 import org.springframework.data.couchbase.core.ExecutableRemoveByIdOperation.ExecutableRemoveById;
 import org.springframework.data.couchbase.core.ExecutableReplaceByIdOperation.ExecutableReplaceById;
 import org.springframework.data.couchbase.core.support.OneAndAllEntity;
+import org.springframework.data.couchbase.core.support.OneAndAllId;
+import org.springframework.data.couchbase.core.support.WithDurability;
+import org.springframework.data.couchbase.core.support.WithExpiry;
 import org.springframework.data.couchbase.domain.Address;
 import org.springframework.data.couchbase.domain.Course;
 import org.springframework.data.couchbase.domain.NaiveAuditorAware;
@@ -81,6 +86,30 @@ class CouchbaseTemplateKeyValueIntegrationTests extends JavaIntegrationTests {
 		couchbaseTemplate.removeByQuery(UserAnnotated3.class).all();
 	}
 
+	@Test
+	void findByIdWithExpiry() {
+		try {
+			User user1 = new User(UUID.randomUUID().toString(), "user1", "user1");
+			User user2 = new User(UUID.randomUUID().toString(), "user2", "user2");
+
+			Collection<User> upserts = (Collection<User>) couchbaseTemplate.upsertById(User.class)
+					.all(Arrays.asList(user1, user2));
+
+			User foundUser = couchbaseTemplate.findById(User.class).withExpiry(Duration.ofSeconds(1)).one(user1.getId());
+			user1.setVersion(foundUser.getVersion());// version will have changed
+			assertEquals(user1, foundUser);
+			sleepMs(2000);
+
+			Collection<User> foundUsers = (Collection<User>) couchbaseTemplate.findById(User.class)
+					.all(Arrays.asList(user1.getId(), user2.getId()));
+			assertEquals(1, foundUsers.size(), "should have found exactly 1 user");
+			assertEquals(user2, foundUsers.iterator().next());
+		} finally {
+			couchbaseTemplate.removeByQuery(User.class).withConsistency(QueryScanConsistency.REQUEST_PLUS).all();
+		}
+
+	}
+
 	@Test
 	void upsertAndFindById() {
 		User user = new User(UUID.randomUUID().toString(), "firstname", "lastname");
@@ -219,28 +248,34 @@ class CouchbaseTemplateKeyValueIntegrationTests extends JavaIntegrationTests {
 		// replace, upsert )
 		Set<User> users = new HashSet<>(); // set of all documents we will insert
 		// Entity classes
-		for (Class clazz : new Class[] { User.class, UserAnnotated.class, UserAnnotated2.class, UserAnnotated3.class }) {
+		for (Class<?> clazz : new Class[] { User.class, UserAnnotated.class, UserAnnotated2.class, UserAnnotated3.class }) {
 			// insert, replace, upsert
-			for (OneAndAllEntity<User> operator : new OneAndAllEntity[] { couchbaseTemplate.insertById(clazz),
-					couchbaseTemplate.replaceById(clazz), couchbaseTemplate.upsertById(clazz) }) {
+			for (Object operator : new Object[] { couchbaseTemplate.insertById(clazz), couchbaseTemplate.replaceById(clazz),
+					couchbaseTemplate.upsertById(clazz), couchbaseTemplate.findById(clazz) }) {
 
 				// create an entity of type clazz
-				Constructor cons = clazz.getConstructor(String.class, String.class, String.class);
+				Constructor<?> cons = clazz.getConstructor(String.class, String.class, String.class);
 				User user = (User) cons.newInstance("" + operator.getClass().getSimpleName() + "_" + clazz.getSimpleName(),
 						"firstname", "lastname");
 
 				if (clazz.equals(User.class)) { // User.java doesn't have an expiry annotation
-					operator = (OneAndAllEntity) ((WithExpiry<User>) operator).withExpiry(Duration.ofSeconds(1));
+					operator = ((WithExpiry<User>) operator).withExpiry(Duration.ofSeconds(1));
 				} else if (clazz.equals(UserAnnotated3.class)) { // override the expiry from the annotation with no expiry
-					operator = (OneAndAllEntity) ((WithExpiry<User>) operator).withExpiry(Duration.ofSeconds(0));
+					operator = ((WithExpiry<User>) operator).withExpiry(Duration.ofSeconds(0));
 				}
 
-				// if replace or remove, we need to insert a document to replace
-				if (operator instanceof ExecutableReplaceById || operator instanceof ExecutableRemoveById) {
-					couchbaseTemplate.insertById(User.class).one(user);
+				// if replace, remove or find, we need to insert a document first
+				if (operator instanceof ExecutableReplaceById || operator instanceof ExecutableRemoveById
+						|| operator instanceof ExecutableFindById) {
+					user = couchbaseTemplate.insertById(User.class).one(user);
+				}
+
+				// call to insert/replace/update/find
+				User returned = operator instanceof OneAndAllEntity ? ((OneAndAllEntity<User>) operator).one(user)
+						: ((OneAndAllId<User>) operator).one(user.getId());
+				if (operator instanceof OneAndAllId) { // the user.version won't be updated
+					user.setVersion(returned.getVersion());
 				}
-				// call to insert/replace/update
-				User returned = operator.one(user);
 				assertEquals(user, returned);
 				users.add(user);
 			}
@@ -250,7 +285,7 @@ class CouchbaseTemplateKeyValueIntegrationTests extends JavaIntegrationTests {
 		for (User user : users) {
 			User found = couchbaseTemplate.findById(user.getClass()).one(user.getId());
 			if (found instanceof UserAnnotated3) {
-				assertNotNull(found, "found should be non null as it was set to have no expirty");
+				assertNotNull(found, "found should be non null as it was set to have no expiry");
 			} else {
 				assertNull(found, "found should have been null as document should be expired");
 			}

src/test/java/org/springframework/data/couchbase/core/ReactiveCouchbaseTemplateKeyValueIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2020 the original author or authors
+ * Copyright 2012-2021 the original author or authors
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -27,7 +27,10 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
 import java.lang.reflect.Constructor;
 import java.lang.reflect.InvocationTargetException;
 import java.time.Duration;
+import java.util.Arrays;
+import java.util.Collection;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Set;
 import java.util.UUID;
 
@@ -35,9 +38,13 @@ import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.springframework.dao.DataIntegrityViolationException;
 import org.springframework.dao.DuplicateKeyException;
-import org.springframework.data.couchbase.core.ExecutableRemoveByIdOperation.ExecutableRemoveById;
+import org.springframework.data.couchbase.core.ReactiveFindByIdOperation.ReactiveFindById;
+import org.springframework.data.couchbase.core.ReactiveRemoveByIdOperation.ReactiveRemoveById;
+import org.springframework.data.couchbase.core.ReactiveReplaceByIdOperation.ReactiveReplaceById;
 import org.springframework.data.couchbase.core.support.OneAndAllEntityReactive;
-import org.springframework.data.couchbase.domain.NaiveAuditorAware;
+import org.springframework.data.couchbase.core.support.OneAndAllIdReactive;
+import org.springframework.data.couchbase.core.support.WithDurability;
+import org.springframework.data.couchbase.core.support.WithExpiry;
 import org.springframework.data.couchbase.domain.PersonValue;
 import org.springframework.data.couchbase.domain.ReactiveNaiveAuditorAware;
 import org.springframework.data.couchbase.domain.User;
@@ -50,6 +57,7 @@ import org.springframework.data.couchbase.util.JavaIntegrationTests;
 
 import com.couchbase.client.java.kv.PersistTo;
 import com.couchbase.client.java.kv.ReplicateTo;
+import com.couchbase.client.java.query.QueryScanConsistency;
 
 /**
  * KV tests Theses tests rely on a cb server running.
@@ -64,9 +72,35 @@ class ReactiveCouchbaseTemplateKeyValueIntegrationTests extends JavaIntegrationT
 	@Override
 	public void beforeEach() {
 		super.beforeEach();
-		reactiveCouchbaseTemplate.removeByQuery(User.class).all();
-		reactiveCouchbaseTemplate.removeByQuery(UserAnnotated.class).all();
-		reactiveCouchbaseTemplate.removeByQuery(UserAnnotated2.class).all();
+		List<RemoveResult> r1  = reactiveCouchbaseTemplate.removeByQuery(User.class).all().collectList().block();
+		List<RemoveResult> r2 = reactiveCouchbaseTemplate.removeByQuery(UserAnnotated.class).all().collectList().block();
+		List<RemoveResult> r3 = reactiveCouchbaseTemplate.removeByQuery(UserAnnotated2.class).all().collectList().block();
+	}
+
+	@Test
+	void findByIdWithExpiry() {
+		try {
+			User user1 = new User(UUID.randomUUID().toString(), "user1", "user1");
+			User user2 = new User(UUID.randomUUID().toString(), "user2", "user2");
+
+			Collection<User> upserts = (Collection<User>) reactiveCouchbaseTemplate.upsertById(User.class)
+					.all(Arrays.asList(user1, user2)).collectList().block();
+
+			User foundUser = reactiveCouchbaseTemplate.findById(User.class).withExpiry(Duration.ofSeconds(1))
+					.one(user1.getId()).block();
+			user1.setVersion(foundUser.getVersion());// version will have changed
+			assertEquals(user1, foundUser);
+			sleepMs(2000);
+
+			Collection<User> foundUsers = (Collection<User>) reactiveCouchbaseTemplate.findById(User.class)
+					.all(Arrays.asList(user1.getId(), user2.getId())).collectList().block();
+			assertEquals(1, foundUsers.size(), "should have found exactly 1 user");
+			assertEquals(user2, foundUsers.iterator().next());
+		} finally {
+			reactiveCouchbaseTemplate.removeByQuery(User.class).withConsistency(QueryScanConsistency.REQUEST_PLUS).all()
+					.collectList().block();
+		}
+
 	}
 
 	@Test
@@ -82,8 +116,8 @@ class ReactiveCouchbaseTemplateKeyValueIntegrationTests extends JavaIntegrationT
 		modifying.setVersion(user.getVersion());
 		modified = reactiveCouchbaseTemplate.replaceById(User.class).one(modifying).block();
 		assertEquals(modifying, modified);
-		if(user == modified){
-			throw new RuntimeException ( " user == modified ");
+		if (user == modified) {
+			throw new RuntimeException(" user == modified ");
 		}
 		assertNotEquals(user, modified);
 		assertEquals(ReactiveNaiveAuditorAware.AUDITOR, modified.getCreatedBy());
@@ -124,7 +158,7 @@ class ReactiveCouchbaseTemplateKeyValueIntegrationTests extends JavaIntegrationT
 			}
 
 			// if replace, we need to insert a document to replace
-			if (operator instanceof ReactiveReplaceByIdOperation.ReactiveReplaceById) {
+			if (operator instanceof ReactiveReplaceById) {
 				reactiveCouchbaseTemplate.insertById(User.class).one(user).block();
 			}
 			// call to insert/replace/update
@@ -152,9 +186,9 @@ class ReactiveCouchbaseTemplateKeyValueIntegrationTests extends JavaIntegrationT
 		// Entity classes
 		for (Class<?> clazz : new Class[] { User.class, UserAnnotated.class, UserAnnotated2.class, UserAnnotated3.class }) {
 			// insert, replace, upsert
-			for (OneAndAllEntityReactive<User> operator : new OneAndAllEntityReactive[] {
-					reactiveCouchbaseTemplate.insertById(clazz), reactiveCouchbaseTemplate.replaceById(clazz),
-					reactiveCouchbaseTemplate.upsertById(clazz) }) {
+			for (Object operator : new Object[] { reactiveCouchbaseTemplate.insertById(clazz),
+					reactiveCouchbaseTemplate.replaceById(clazz), reactiveCouchbaseTemplate.upsertById(clazz),
+					reactiveCouchbaseTemplate.findById(clazz) }) {
 
 				// create an entity of type clazz
 				Constructor<?> cons = clazz.getConstructor(String.class, String.class, String.class);
@@ -162,18 +196,23 @@ class ReactiveCouchbaseTemplateKeyValueIntegrationTests extends JavaIntegrationT
 						"firstname", "lastname");
 
 				if (clazz.equals(User.class)) { // User.java doesn't have an expiry annotation
-					operator = (OneAndAllEntityReactive<User>) ((WithExpiry<User>) operator).withExpiry(Duration.ofSeconds(1));
+					operator = ((WithExpiry<User>) operator).withExpiry(Duration.ofSeconds(1));
 				} else if (clazz.equals(UserAnnotated3.class)) { // override the expiry from the annotation with no expiry
-					operator = (OneAndAllEntityReactive<User>) ((WithExpiry<User>) operator).withExpiry(Duration.ofSeconds(0));
+					operator = ((WithExpiry<User>) operator).withExpiry(Duration.ofSeconds(0));
 				}
 
-				// if replace or remove, we need to insert a document to replace
-				if (operator instanceof ReactiveReplaceByIdOperation.ReactiveReplaceById
-						|| operator instanceof ExecutableRemoveById) {
-					reactiveCouchbaseTemplate.insertById(User.class).one(user).block();
+				// if replace, remove or find, we need to insert a document first
+				if (operator instanceof ReactiveReplaceById || operator instanceof ReactiveRemoveById
+						|| operator instanceof ReactiveFindById) {
+					user = reactiveCouchbaseTemplate.insertById(User.class).one(user).block();
+				}
+				// call to insert/replace/update/find
+				User returned = operator instanceof OneAndAllEntityReactive
+						? ((OneAndAllEntityReactive<User>) operator).one(user).block()
+						: ((OneAndAllIdReactive<User>) operator).one(user.getId()).block();
+				if (operator instanceof OneAndAllIdReactive) { // the user.version won't be updated
+					user.setVersion(returned.getVersion());
 				}
-				// call to insert/replace/update
-				User returned = operator.one(user).block();
 				assertEquals(user, returned);
 				users.add(user);
 			}
@@ -183,7 +222,7 @@ class ReactiveCouchbaseTemplateKeyValueIntegrationTests extends JavaIntegrationT
 		for (User user : users) {
 			User found = reactiveCouchbaseTemplate.findById(user.getClass()).one(user.getId()).block();
 			if (found instanceof UserAnnotated3) {
-				assertNotNull(found, "found should be non null as it was set to have no expirty");
+				assertNotNull(found, "found should be non null as it was set to have no expiry");
 			} else {
 				assertNull(found, "found should have been null as document should be expired");
 			}