diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/InsertOptions.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/InsertOptions.java index d7334be2d..a4d735ed7 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/InsertOptions.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/InsertOptions.java @@ -16,6 +16,7 @@ package org.springframework.data.cassandra.core; import java.time.Duration; +import java.time.Instant; import java.util.concurrent.TimeUnit; import lombok.EqualsAndHashCode; @@ -40,9 +41,9 @@ public class InsertOptions extends WriteOptions { private boolean ifNotExists; private InsertOptions(@Nullable ConsistencyLevel consistencyLevel, @Nullable RetryPolicy retryPolicy, - @Nullable Boolean tracing, @Nullable Integer fetchSize, Duration readTimeout, Duration ttl, boolean ifNotExists) { + @Nullable Boolean tracing, @Nullable Integer fetchSize, Duration readTimeout, Duration ttl, Long timestamp, boolean ifNotExists) { - super(consistencyLevel, retryPolicy, tracing, fetchSize, readTimeout, ttl); + super(consistencyLevel, retryPolicy, tracing, fetchSize, readTimeout, ttl, timestamp); this.ifNotExists = ifNotExists; } @@ -152,6 +153,16 @@ public class InsertOptions extends WriteOptions { return (InsertOptionsBuilder) super.ttl(ttl); } + @Override + public InsertOptionsBuilder timestamp(long timestamp) { + return (InsertOptionsBuilder) super.timestamp(timestamp); + } + + @Override + public InsertOptionsBuilder timestamp(Instant timestamp) { + return (InsertOptionsBuilder) super.timestamp(timestamp); + } + /** * Use light-weight transactions by applying {@code IF NOT EXISTS}. * @@ -181,7 +192,7 @@ public class InsertOptions extends WriteOptions { */ public InsertOptions build() { return new InsertOptions(this.consistencyLevel, this.retryPolicy, this.tracing, - this.fetchSize, this.readTimeout, this.ttl, this.ifNotExists); + this.fetchSize, this.readTimeout, this.ttl, this.timestamp, this.ifNotExists); } } } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/UpdateOptions.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/UpdateOptions.java index 623e1300f..d5c53b47c 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/UpdateOptions.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/UpdateOptions.java @@ -16,6 +16,7 @@ package org.springframework.data.cassandra.core; import java.time.Duration; +import java.time.Instant; import java.util.concurrent.TimeUnit; import lombok.EqualsAndHashCode; @@ -40,9 +41,9 @@ public class UpdateOptions extends WriteOptions { private boolean ifExists; private UpdateOptions(@Nullable ConsistencyLevel consistencyLevel, @Nullable RetryPolicy retryPolicy, - @Nullable Boolean tracing, @Nullable Integer fetchSize, Duration readTimeout, Duration ttl, boolean ifExists) { + @Nullable Boolean tracing, @Nullable Integer fetchSize, Duration readTimeout, Duration ttl, Long timestamp, boolean ifExists) { - super(consistencyLevel, retryPolicy, tracing, fetchSize, readTimeout, ttl); + super(consistencyLevel, retryPolicy, tracing, fetchSize, readTimeout, ttl, timestamp); this.ifExists = ifExists; } @@ -152,6 +153,16 @@ public class UpdateOptions extends WriteOptions { return (UpdateOptionsBuilder) super.ttl(ttl); } + @Override + public UpdateOptionsBuilder timestamp(long timestamp) { + return (UpdateOptionsBuilder) super.timestamp(timestamp); + } + + @Override + public UpdateOptionsBuilder timestamp(Instant timestamp) { + return (UpdateOptionsBuilder) super.timestamp(timestamp); + } + /** * Use light-weight transactions by applying {@code IF EXISTS}. * @@ -181,7 +192,7 @@ public class UpdateOptions extends WriteOptions { */ public UpdateOptions build() { return new UpdateOptions(this.consistencyLevel, this.retryPolicy, this.tracing, - this.fetchSize, this.readTimeout, this.ttl, this.ifExists); + this.fetchSize, this.readTimeout, this.ttl, this.timestamp, this.ifExists); } } } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/QueryOptionsUtil.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/QueryOptionsUtil.java index 0f57723aa..567837e7e 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/QueryOptionsUtil.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/QueryOptionsUtil.java @@ -106,6 +106,9 @@ public abstract class QueryOptionsUtil { if (!writeOptions.getTtl().isNegative()) { insert.using(QueryBuilder.ttl(Math.toIntExact(writeOptions.getTtl().getSeconds()))); } + if (writeOptions.getTimestamp() != null) { + insert.using(QueryBuilder.timestamp(writeOptions.getTimestamp())); + } return insert; } @@ -126,6 +129,9 @@ public abstract class QueryOptionsUtil { if (!writeOptions.getTtl().isNegative()) { update.using(QueryBuilder.ttl(Math.toIntExact(writeOptions.getTtl().getSeconds()))); } + if (writeOptions.getTimestamp() != null) { + update.using(QueryBuilder.timestamp(writeOptions.getTimestamp())); + } return update; } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/WriteOptions.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/WriteOptions.java index 09be89126..a2ba691dc 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/WriteOptions.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/WriteOptions.java @@ -16,6 +16,7 @@ package org.springframework.data.cassandra.core.cql; import java.time.Duration; +import java.time.Instant; import java.util.concurrent.TimeUnit; import lombok.EqualsAndHashCode; @@ -40,6 +41,7 @@ public class WriteOptions extends QueryOptions { private static final WriteOptions EMPTY = new WriteOptionsBuilder().build(); private final Duration ttl; + private final @Nullable Long timestamp; /** * Creates new {@link WriteOptions} for the given {@link ConsistencyLevel} and {@link RetryPolicy}. @@ -68,14 +70,16 @@ public class WriteOptions extends QueryOptions { super(consistencyLevel, retryPolicy); this.ttl = ttl == null ? Duration.ofMillis(-1) : Duration.ofSeconds(ttl); + this.timestamp = null; } protected WriteOptions(@Nullable ConsistencyLevel consistencyLevel, @Nullable RetryPolicy retryPolicy, - @Nullable Boolean tracing, @Nullable Integer fetchSize, Duration readTimeout, Duration ttl) { + @Nullable Boolean tracing, @Nullable Integer fetchSize, Duration readTimeout, Duration ttl, Long timestamp) { super(consistencyLevel, retryPolicy, tracing, fetchSize, readTimeout); this.ttl = ttl; + this.timestamp = timestamp; } /** @@ -116,6 +120,13 @@ public class WriteOptions extends QueryOptions { return this.ttl; } + /** + * @return mutation timestamp in microseconds. + */ + public Long getTimestamp() { + return this.timestamp; + } + /** * Builder for {@link WriteOptions}. * @@ -125,6 +136,7 @@ public class WriteOptions extends QueryOptions { public static class WriteOptionsBuilder extends QueryOptionsBuilder { protected Duration ttl = Duration.ofMillis(-1); + protected Long timestamp = null; protected WriteOptionsBuilder() {} @@ -133,6 +145,7 @@ public class WriteOptions extends QueryOptions { super(writeOptions); this.ttl = writeOptions.ttl; + this.timestamp = writeOptions.timestamp; } /* @@ -236,6 +249,30 @@ public class WriteOptions extends QueryOptions { return this; } + /** + * Sets the timestamp of write operations. + * + * @param timestamp mutation timestamp in microseconds. + * @return {@code this} {@link WriteOptionsBuilder} + */ + public WriteOptionsBuilder timestamp(long timestamp) { + Assert.isTrue(timestamp >= 0, "Timestamp must be greater than equal to zero"); + this.timestamp = timestamp; + return this; + } + + /** + * Sets the timestamp of write operations. + * + * @param timestamp mutation date time. + * @return {@code this} {@link WriteOptionsBuilder} + */ + public WriteOptionsBuilder timestamp(Instant timestamp) { + Assert.notNull(timestamp, "Timestamp must not be null"); + this.timestamp = TimeUnit.MILLISECONDS.toMicros(timestamp.toEpochMilli()); + return this; + } + /** * Builds a new {@link WriteOptions} with the configured values. * @@ -243,7 +280,7 @@ public class WriteOptions extends QueryOptions { */ public WriteOptions build() { return new WriteOptions(this.consistencyLevel, this.retryPolicy, this.tracing, - this.fetchSize, this.readTimeout, this.ttl); + this.fetchSize, this.readTimeout, this.ttl, this.timestamp); } } } diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/CassandraTemplateIntegrationTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/CassandraTemplateIntegrationTests.java index f9fda3f41..4735a484b 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/CassandraTemplateIntegrationTests.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/CassandraTemplateIntegrationTests.java @@ -19,6 +19,9 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.junit.Assume.assumeTrue; import static org.springframework.data.cassandra.core.query.Criteria.where; +import java.time.Instant; +import java.time.LocalDateTime; +import java.time.ZoneId; import java.util.Arrays; import java.util.Collections; import java.util.HashSet; @@ -191,6 +194,24 @@ public class CassandraTemplateIntegrationTests extends AbstractKeyspaceCreatingI assertThat(count).isEqualTo(1L); } + @Test // DATACASS-155 + public void shouldNotOverrideLaterMutation() { + Instant now = LocalDateTime.now().atZone( ZoneId.systemDefault() ).toInstant(); + User user = new User("heisenberg", "Walter", "White"); + template.insert(user); + + // more recent mutation + user.setFirstname("John"); + template.update(user); + + // previous mutation + user.setFirstname("Greg"); + template.update(user, UpdateOptions.builder().timestamp(now.minusSeconds(10)).build()); + + User loaded = template.selectOneById(user.getId(), User.class); + assertThat(loaded.getFirstname()).isEqualTo("John"); + } + @Test // DATACASS-512 public void shouldInsertEntityAndCountByQuery() { diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/InsertOptionsUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/InsertOptionsUnitTests.java index cf8831045..78f787ec8 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/InsertOptionsUnitTests.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/InsertOptionsUnitTests.java @@ -18,6 +18,9 @@ package org.springframework.data.cassandra.core; import static org.assertj.core.api.Assertions.assertThat; import java.time.Duration; +import java.time.Instant; +import java.time.LocalDateTime; +import java.time.ZoneOffset; import org.junit.Test; @@ -30,13 +33,16 @@ public class InsertOptionsUnitTests { @Test // DATACASS-250 public void shouldConfigureInsertOptions() { + Instant now = LocalDateTime.now().toInstant(ZoneOffset.UTC); InsertOptions insertOptions = InsertOptions.builder() .ttl(10) + .timestamp(now) .withIfNotExists() .build(); assertThat(insertOptions.getTtl()).isEqualTo(Duration.ofSeconds(10)); + assertThat(insertOptions.getTimestamp()).isEqualTo(now.toEpochMilli() * 1000); assertThat(insertOptions.isIfNotExists()).isTrue(); } @@ -45,14 +51,16 @@ public class InsertOptionsUnitTests { InsertOptions insertOptions = InsertOptions.builder() .ttl(10) + .timestamp(1519222753) .withIfNotExists() .build(); - InsertOptions mutated = insertOptions.mutate().ttl(Duration.ofSeconds(5)).build(); + InsertOptions mutated = insertOptions.mutate().ttl(Duration.ofSeconds(5)).timestamp(1519200753).build(); assertThat(mutated).isNotNull(); assertThat(mutated).isNotSameAs(insertOptions); assertThat(mutated.getTtl()).isEqualTo(Duration.ofSeconds(5)); + assertThat(mutated.getTimestamp()).isEqualTo(1519200753); assertThat(mutated.isIfNotExists()).isTrue(); } } diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/UpdateOptionsUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/UpdateOptionsUnitTests.java index b6e23023c..003942bdb 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/UpdateOptionsUnitTests.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/UpdateOptionsUnitTests.java @@ -18,6 +18,9 @@ package org.springframework.data.cassandra.core; import static org.assertj.core.api.Assertions.assertThat; import java.time.Duration; +import java.time.Instant; +import java.time.LocalDateTime; +import java.time.ZoneOffset; import org.junit.Test; @@ -30,13 +33,16 @@ public class UpdateOptionsUnitTests { @Test // DATACASS-250 public void shouldConfigureUpdateOptions() { + Instant now = LocalDateTime.now().toInstant(ZoneOffset.UTC); UpdateOptions updateOptions = UpdateOptions.builder() .ttl(10) + .timestamp(now) .withIfExists() .build(); assertThat(updateOptions.getTtl()).isEqualTo(Duration.ofSeconds(10)); + assertThat(updateOptions.getTimestamp()).isEqualTo(now.toEpochMilli() * 1000); assertThat(updateOptions.isIfExists()).isTrue(); } @@ -45,14 +51,16 @@ public class UpdateOptionsUnitTests { UpdateOptions updateOptions = UpdateOptions.builder() .ttl(10) + .timestamp(1519222753) .withIfExists() .build(); - UpdateOptions mutated = updateOptions.mutate().ttl(20).build(); + UpdateOptions mutated = updateOptions.mutate().ttl(20).timestamp(1519000753).build(); assertThat(mutated).isNotNull(); assertThat(mutated).isNotSameAs(updateOptions); assertThat(mutated.getTtl()).isEqualTo(Duration.ofSeconds(20)); + assertThat(mutated.getTimestamp()).isEqualTo(1519000753); assertThat(mutated.isIfExists()).isTrue(); } } diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/WriteOptionsUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/WriteOptionsUnitTests.java index 16271b8d1..0e970b34a 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/WriteOptionsUnitTests.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/WriteOptionsUnitTests.java @@ -18,6 +18,9 @@ package org.springframework.data.cassandra.core.cql; import static org.assertj.core.api.Assertions.assertThat; import java.time.Duration; +import java.time.Instant; +import java.time.LocalDateTime; +import java.time.ZoneOffset; import org.junit.Test; @@ -38,6 +41,7 @@ public class WriteOptionsUnitTests { WriteOptions writeOptions = WriteOptions.builder() .consistencyLevel(com.datastax.driver.core.ConsistencyLevel.ANY) .ttl(123) + .timestamp(1519000753) .retryPolicy(FallthroughRetryPolicy.INSTANCE) .readTimeout(1) .fetchSize(10) @@ -45,6 +49,7 @@ public class WriteOptionsUnitTests { .build(); assertThat(writeOptions.getTtl()).isEqualTo(Duration.ofSeconds(123)); + assertThat(writeOptions.getTimestamp()).isEqualTo(1519000753); assertThat(writeOptions.getRetryPolicy()).isEqualTo(FallthroughRetryPolicy.INSTANCE); assertThat(writeOptions.getConsistencyLevel()).isEqualTo(ConsistencyLevel.ANY); assertThat(writeOptions.getReadTimeout()).isEqualTo(Duration.ofMillis(1)); @@ -80,10 +85,12 @@ public class WriteOptionsUnitTests { @Test // DATACASS-56 public void buildWriteOptionsMutate() { + Instant now = LocalDateTime.now().toInstant(ZoneOffset.UTC); WriteOptions writeOptions = WriteOptions.builder() .consistencyLevel(com.datastax.driver.core.ConsistencyLevel.ANY) .ttl(123) + .timestamp(now) .retryPolicy(FallthroughRetryPolicy.INSTANCE) .readTimeout(1) .fetchSize(10) @@ -95,6 +102,7 @@ public class WriteOptionsUnitTests { assertThat(mutated).isNotNull(); assertThat(mutated).isNotSameAs(writeOptions); assertThat(mutated.getTtl()).isEqualTo(Duration.ofSeconds(123)); + assertThat(mutated.getTimestamp()).isEqualTo(now.toEpochMilli() * 1000); assertThat(mutated.getRetryPolicy()).isEqualTo(DowngradingConsistencyRetryPolicy.INSTANCE); assertThat(mutated.getConsistencyLevel()).isEqualTo(ConsistencyLevel.ANY); assertThat(mutated.getReadTimeout()).isEqualTo(Duration.ofMillis(1));