DATAMONGO-2113 - Fix resumeTimestamp conversion for change streams.

pom.xml

@@ -5,7 +5,7 @@
 
 	<groupId>org.springframework.data</groupId>
 	<artifactId>spring-data-mongodb-parent</artifactId>
-	<version>2.2.0.BUILD-SNAPSHOT</version>
+	<version>2.2.0.DATAMONGO-2113-SNAPSHOT</version>
 	<packaging>pom</packaging>
 
 	<name>Spring Data MongoDB</name>

spring-data-mongodb-benchmarks/pom.xml

@@ -7,7 +7,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>2.2.0.BUILD-SNAPSHOT</version>
+		<version>2.2.0.DATAMONGO-2113-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

spring-data-mongodb-cross-store/pom.xml

@@ -6,7 +6,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>2.2.0.BUILD-SNAPSHOT</version>
+		<version>2.2.0.DATAMONGO-2113-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 
@@ -50,7 +50,7 @@
 		<dependency>
 			<groupId>org.springframework.data</groupId>
 			<artifactId>spring-data-mongodb</artifactId>
-			<version>2.2.0.BUILD-SNAPSHOT</version>
+			<version>2.2.0.DATAMONGO-2113-SNAPSHOT</version>
 		</dependency>
 
 		<!-- reactive -->

spring-data-mongodb-distribution/pom.xml

@@ -13,7 +13,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>2.2.0.BUILD-SNAPSHOT</version>
+		<version>2.2.0.DATAMONGO-2113-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

spring-data-mongodb/pom.xml

@@ -11,7 +11,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>2.2.0.BUILD-SNAPSHOT</version>
+		<version>2.2.0.DATAMONGO-2113-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

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

@@ -15,14 +15,18 @@
  */
 package org.springframework.data.mongodb.core.convert;
 
-import static org.hamcrest.Matchers.*;
-import static org.junit.Assert.*;
+import static org.assertj.core.api.Assertions.*;
 
 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,15 +36,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;
-
 /**
  * Unit tests for {@link MongoConverters}.
  *
@@ -55,10 +58,10 @@ public void convertsBigDecimalToStringAndBackCorrectly() {
 
 		BigDecimal bigDecimal = BigDecimal.valueOf(254, 1);
 		String value = BigDecimalToStringConverter.INSTANCE.convert(bigDecimal);
-		assertThat(value, is("25.4"));
+		assertThat(value).isEqualTo("25.4");
 
 		BigDecimal reference = StringToBigDecimalConverter.INSTANCE.convert(value);
-		assertThat(reference, is(bigDecimal));
+		assertThat(reference).isEqualTo(bigDecimal);
 	}
 
 	@Test // DATAMONGO-858
@@ -69,7 +72,7 @@ public void convertsBoxToDocumentAndBackCorrectly() {
 		Document document = GeoConverters.BoxToDocumentConverter.INSTANCE.convert(box);
 		Shape shape = GeoConverters.DocumentToBoxConverter.INSTANCE.convert(document);
 
-		assertThat(shape, is((org.springframework.data.geo.Shape) box));
+		assertThat(shape).isEqualTo(box);
 	}
 
 	@Test // DATAMONGO-858
@@ -80,7 +83,7 @@ public void convertsCircleToDocumentAndBackCorrectly() {
 		Document document = GeoConverters.CircleToDocumentConverter.INSTANCE.convert(circle);
 		Shape shape = GeoConverters.DocumentToCircleConverter.INSTANCE.convert(document);
 
-		assertThat(shape, is((org.springframework.data.geo.Shape) circle));
+		assertThat(shape).isEqualTo(circle);
 	}
 
 	@Test // DATAMONGO-858
@@ -91,7 +94,7 @@ public void convertsPolygonToDocumentAndBackCorrectly() {
 		Document document = GeoConverters.PolygonToDocumentConverter.INSTANCE.convert(polygon);
 		Shape shape = GeoConverters.DocumentToPolygonConverter.INSTANCE.convert(document);
 
-		assertThat(shape, is((org.springframework.data.geo.Shape) polygon));
+		assertThat(shape).isEqualTo(polygon);
 	}
 
 	@Test // DATAMONGO-858
@@ -102,7 +105,7 @@ public void convertsSphereToDocumentAndBackCorrectly() {
 		Document document = GeoConverters.SphereToDocumentConverter.INSTANCE.convert(sphere);
 		org.springframework.data.geo.Shape shape = GeoConverters.DocumentToSphereConverter.INSTANCE.convert(document);
 
-		assertThat(shape, is((org.springframework.data.geo.Shape) sphere));
+		assertThat(shape).isEqualTo(sphere);
 	}
 
 	@Test // DATAMONGO-858
@@ -113,36 +116,44 @@ public void convertsPointToListAndBackCorrectly() {
 		Document document = GeoConverters.PointToDocumentConverter.INSTANCE.convert(point);
 		org.springframework.data.geo.Point converted = GeoConverters.DocumentToPointConverter.INSTANCE.convert(document);
 
-		assertThat(converted, is((org.springframework.data.geo.Point) point));
+		assertThat(converted).isEqualTo(point);
 	}
 
 	@Test // DATAMONGO-1372
 	public void convertsCurrencyToStringCorrectly() {
-		assertThat(CurrencyToStringConverter.INSTANCE.convert(Currency.getInstance("USD")), is("USD"));
+		assertThat(CurrencyToStringConverter.INSTANCE.convert(Currency.getInstance("USD"))).isEqualTo("USD");
 	}
 
 	@Test // DATAMONGO-1372
 	public void convertsStringToCurrencyCorrectly() {
-		assertThat(StringToCurrencyConverter.INSTANCE.convert("USD"), is(Currency.getInstance("USD")));
+		assertThat(StringToCurrencyConverter.INSTANCE.convert("USD")).isEqualTo(Currency.getInstance("USD"));
 	}
 
 	@Test // DATAMONGO-1416
 	public void convertsAtomicLongToLongCorrectly() {
-		assertThat(AtomicLongToLongConverter.INSTANCE.convert(new AtomicLong(100L)), is(100L));
+		assertThat(AtomicLongToLongConverter.INSTANCE.convert(new AtomicLong(100L))).isEqualTo(100L);
 	}
 
 	@Test // DATAMONGO-1416
 	public void convertsAtomicIntegerToIntegerCorrectly() {
-		assertThat(AtomicIntegerToIntegerConverter.INSTANCE.convert(new AtomicInteger(100)), is(100));
+		assertThat(AtomicIntegerToIntegerConverter.INSTANCE.convert(new AtomicInteger(100))).isEqualTo(100);
 	}
 
 	@Test // DATAMONGO-1416
 	public void convertsLongToAtomicLongCorrectly() {
-		assertThat(LongToAtomicLongConverter.INSTANCE.convert(100L), is(instanceOf(AtomicLong.class)));
+		assertThat(LongToAtomicLongConverter.INSTANCE.convert(100L)).isInstanceOf(AtomicLong.class);
 	}
 
 	@Test // DATAMONGO-1416
 	public void convertsIntegerToAtomicIntegerCorrectly() {
-		assertThat(IntegerToAtomicIntegerConverter.INSTANCE.convert(100), is(instanceOf(AtomicInteger.class)));
+		assertThat(IntegerToAtomicIntegerConverter.INSTANCE.convert(100)).isInstanceOf(AtomicInteger.class);
 	}
+
+	@Test // DATAMONGO-2113
+	public void convertsBsonTimestampToInstantCorrectly() {
+
+		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);