DATAMONGO-2113 - Fix resumeTimestamp conversion for change streams.

We now use the first 32 bits of the timestamp to create the instant and ignore the ordinal value.

Original pull request: spring-projects#615.

Author: christophstrobl

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ChangeStreamEvent.java

@@ -83,7 +83,9 @@ public ChangeStreamDocument<Document> getRaw() {
 	 */
 	@Nullable
 	public Instant getTimestamp() {
-		return raw != null && raw.getClusterTime() != null ? Instant.ofEpochMilli(raw.getClusterTime().getValue()) : null;
+
+		return raw != null && raw.getClusterTime() != null
+				? converter.getConversionService().convert(raw.getClusterTime(), Instant.class) : null;
 	}
 
 	/**

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java

@@ -1915,7 +1915,7 @@ public <T> Flux<ChangeStreamEvent<T>> changeStream(@Nullable String database, @N
 
 		publisher = options.getResumeToken().map(BsonValue::asDocument).map(publisher::resumeAfter).orElse(publisher);
 		publisher = options.getCollation().map(Collation::toMongoCollation).map(publisher::collation).orElse(publisher);
-		publisher = options.getResumeTimestamp().map(it -> new BsonTimestamp(it.toEpochMilli()))
+		publisher = options.getResumeTimestamp().map(it -> new BsonTimestamp((int) it.getEpochSecond(), 0))
 				.map(publisher::startAtOperationTime).orElse(publisher);
 		publisher = publisher.fullDocument(options.getFullDocumentLookup().orElse(fullDocument));
 

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/convert/MongoConverters.java

@@ -19,13 +19,15 @@
 import java.math.BigInteger;
 import java.net.MalformedURLException;
 import java.net.URL;
+import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Currency;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 
+import org.bson.BsonTimestamp;
 import org.bson.Document;
 import org.bson.types.Binary;
 import org.bson.types.Code;
@@ -86,6 +88,7 @@ static Collection<Object> getConvertersToRegister() {
 		converters.add(LongToAtomicLongConverter.INSTANCE);
 		converters.add(IntegerToAtomicIntegerConverter.INSTANCE);
 		converters.add(BinaryToByteArrayConverter.INSTANCE);
+		converters.add(BsonTimestampToInstantConverter.INSTANCE);
 
 		return converters;
 	}
@@ -465,4 +468,22 @@ public byte[] convert(Binary source) {
 			return source.getData();
 		}
 	}
+
+	/**
+	 * {@link Converter} implementation converting {@link BsonTimestamp} into {@link Instant}.
+	 *
+	 * @author Christoph Strobl
+	 * @since 2.1.2
+	 */
+	@ReadingConverter
+	enum BsonTimestampToInstantConverter implements Converter<BsonTimestamp, Instant> {
+
+		INSTANCE;
+
+		@Nullable
+		@Override
+		public Instant convert(BsonTimestamp source) {
+			return Instant.ofEpochSecond(source.getTime(), 0);
+		}
+	}
 }

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/messaging/ChangeStreamTask.java

@@ -115,7 +115,7 @@ protected MongoCursor<ChangeStreamDocument<Document>> initCursor(MongoTemplate t
 					.orElseGet(() -> ClassUtils.isAssignable(Document.class, targetType) ? FullDocument.DEFAULT
 							: FullDocument.UPDATE_LOOKUP);
 
-			startAt = changeStreamOptions.getResumeTimestamp().map(Instant::toEpochMilli).map(BsonTimestamp::new)
+			startAt = changeStreamOptions.getResumeTimestamp().map(it -> new BsonTimestamp((int) it.getEpochSecond(), 0))
 					.orElse(null);
 		}
 

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateTests.java

@@ -29,6 +29,7 @@
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
+import java.time.Duration;
 import java.time.Instant;
 import java.util.Arrays;
 import java.util.Collections;
@@ -1355,7 +1356,7 @@ public void watchesDatabaseCorrectly() throws InterruptedException {
 		}
 	}
 
-	@Test // DATAMONGO-2012
+	@Test // DATAMONGO-2012, DATAMONGO-2113
 	public void resumesAtTimestampCorrectly() throws InterruptedException {
 
 		Assumptions.assumeThat(ReplicaSet.required().runsAsReplicaSet()).isTrue();
@@ -1372,7 +1373,7 @@ public void resumesAtTimestampCorrectly() throws InterruptedException {
 		Person person2 = new Person("Data", 37);
 		Person person3 = new Person("MongoDB", 39);
 
-		StepVerifier.create(template.save(person1)).expectNextCount(1).verifyComplete();
+		StepVerifier.create(template.save(person1).delayElement(Duration.ofSeconds(1))).expectNextCount(1).verifyComplete();
 		StepVerifier.create(template.save(person2)).expectNextCount(1).verifyComplete();
 
 		Thread.sleep(500); // just give it some time to link receive all events

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/convert/MongoConvertersUnitTests.java

@@ -19,10 +19,15 @@
 import static org.junit.Assert.*;
 
 import java.math.BigDecimal;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
 import java.util.Currency;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 
+import org.assertj.core.data.TemporalUnitLessThanOffset;
+import org.bson.BsonTimestamp;
+import org.bson.Document;
 import org.junit.Test;
 import org.springframework.data.geo.Box;
 import org.springframework.data.geo.Circle;
@@ -32,14 +37,14 @@
 import org.springframework.data.mongodb.core.convert.MongoConverters.AtomicIntegerToIntegerConverter;
 import org.springframework.data.mongodb.core.convert.MongoConverters.AtomicLongToLongConverter;
 import org.springframework.data.mongodb.core.convert.MongoConverters.BigDecimalToStringConverter;
+import org.springframework.data.mongodb.core.convert.MongoConverters.BsonTimestampToInstantConverter;
 import org.springframework.data.mongodb.core.convert.MongoConverters.CurrencyToStringConverter;
 import org.springframework.data.mongodb.core.convert.MongoConverters.IntegerToAtomicIntegerConverter;
 import org.springframework.data.mongodb.core.convert.MongoConverters.LongToAtomicLongConverter;
 import org.springframework.data.mongodb.core.convert.MongoConverters.StringToBigDecimalConverter;
 import org.springframework.data.mongodb.core.convert.MongoConverters.StringToCurrencyConverter;
 import org.springframework.data.mongodb.core.geo.Sphere;
-
-import org.bson.Document;
+import org.springframework.data.mongodb.test.util.Assertions;
 
 /**
  * Unit tests for {@link MongoConverters}.
@@ -145,4 +150,12 @@ public void convertsLongToAtomicLongCorrectly() {
 	public void convertsIntegerToAtomicIntegerCorrectly() {
 		assertThat(IntegerToAtomicIntegerConverter.INSTANCE.convert(100), is(instanceOf(AtomicInteger.class)));
 	}
+
+	@Test // DATAMONGO-2113
+	public void convertsBsonTimestampToInstantCorrectly() {
+		
+		Assertions.assertThat(BsonTimestampToInstantConverter.INSTANCE.convert(new BsonTimestamp(6615900307735969796L)))
+				.isCloseTo(Instant.ofEpochSecond(1540384327), new TemporalUnitLessThanOffset(100, ChronoUnit.MILLIS));
+	}
+
 }

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/messaging/ChangeStreamTests.java

@@ -405,7 +405,7 @@ public void readsFullDocumentForUpdateWhenNotMappedToDomainTypeButLookupSpecifie
 				.append("user_name", "jellyBelly").append("age", 8).append("_class", User.class.getName()));
 	}
 
-	@Test // DATAMONGO-2012
+	@Test // DATAMONGO-2012, DATAMONGO-2113
 	public void resumeAtTimestampCorrectly() throws InterruptedException {
 
 		CollectingMessageListener<ChangeStreamDocument<Document>, User> messageListener1 = new CollectingMessageListener<>();
@@ -415,6 +415,9 @@ public void resumeAtTimestampCorrectly() throws InterruptedException {
 		awaitSubscription(subscription1);
 
 		template.save(jellyBelly);
+
+		Thread.sleep(1000); // cluster timestamp is in seconds, so we need to wait at least one.
+
 		template.save(sugarSplashy);
 
 		awaitMessages(messageListener1, 12);