Apply zero TTL to modifying statements.

We now apply a zero TTL if it was set. Previously, we omitted zero TTLs and dropped these.

Closes #1535

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/DeleteOptions.java

@@ -51,7 +51,7 @@ private DeleteOptions(@Nullable ConsistencyLevel consistencyLevel, ExecutionProf
 			@Nullable Boolean idempotent, @Nullable CqlIdentifier keyspace, @Nullable Integer pageSize,
 			@Nullable CqlIdentifier routingKeyspace, @Nullable ByteBuffer routingKey,
 			@Nullable ConsistencyLevel serialConsistencyLevel,
-			Duration timeout, Duration ttl, @Nullable Long timestamp, @Nullable Boolean tracing, boolean ifExists,
+			Duration timeout, @Nullable Duration ttl, @Nullable Long timestamp, @Nullable Boolean tracing, boolean ifExists,
 			@Nullable Filter ifCondition) {
 
 		super(consistencyLevel, executionProfileResolver, idempotent, keyspace, pageSize, routingKeyspace, routingKey,

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/InsertOptions.java

@@ -48,7 +48,8 @@ private InsertOptions(@Nullable ConsistencyLevel consistencyLevel, ExecutionProf
 			@Nullable Boolean idempotent, @Nullable CqlIdentifier keyspace, @Nullable Integer pageSize,
 			@Nullable CqlIdentifier routingKeyspace, @Nullable ByteBuffer routingKey,
 			@Nullable ConsistencyLevel serialConsistencyLevel,
-			Duration timeout, Duration ttl, @Nullable Long timestamp, @Nullable Boolean tracing, boolean ifNotExists,
+			Duration timeout, @Nullable Duration ttl, @Nullable Long timestamp, @Nullable Boolean tracing,
+			boolean ifNotExists,
 			boolean insertNulls) {
 
 		super(consistencyLevel, executionProfileResolver, idempotent, keyspace, pageSize, routingKeyspace, routingKey,

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/UpdateOptions.java

@@ -51,7 +51,8 @@ public class UpdateOptions extends WriteOptions {
 	private UpdateOptions(@Nullable ConsistencyLevel consistencyLevel, ExecutionProfileResolver executionProfileResolver,
 			@Nullable Filter ifCondition, boolean ifExists, @Nullable Boolean idempotent, @Nullable CqlIdentifier keyspace,
 			@Nullable Integer pageSize, @Nullable CqlIdentifier routingKeyspace, @Nullable ByteBuffer routingKey,
-			@Nullable ConsistencyLevel serialConsistencyLevel, Duration timeout, Duration ttl, @Nullable Long timestamp,
+			@Nullable ConsistencyLevel serialConsistencyLevel, Duration timeout, @Nullable Duration ttl,
+			@Nullable Long timestamp,
 			@Nullable Boolean tracing) {
 
 		super(consistencyLevel, executionProfileResolver, idempotent, keyspace, pageSize, routingKeyspace, routingKey,

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/QueryOptionsUtil.java

@@ -19,6 +19,7 @@
 import java.util.function.BiFunction;
 
 import org.springframework.data.cassandra.core.cql.util.Bindings;
+import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
 import com.datastax.oss.driver.api.core.cql.BatchStatement;
@@ -205,8 +206,8 @@ private static int getTtlSeconds(Duration ttl) {
 		return Math.toIntExact(ttl.getSeconds());
 	}
 
-	private static boolean hasTtl(Duration ttl) {
-		return !ttl.isZero() && !ttl.isNegative();
+	private static boolean hasTtl(@Nullable Duration ttl) {
+		return ttl != null && !ttl.isNegative();
 	}
 
 	/**

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/WriteOptions.java

@@ -43,14 +43,15 @@ public class WriteOptions extends QueryOptions {
 
 	private static final WriteOptions EMPTY = new WriteOptionsBuilder().build();
 
-	private final Duration ttl;
+	private final @Nullable Duration ttl;
 
 	private final @Nullable Long timestamp;
 
 	protected WriteOptions(@Nullable ConsistencyLevel consistencyLevel, ExecutionProfileResolver executionProfileResolver,
 			@Nullable Boolean idempotent, @Nullable CqlIdentifier keyspace, @Nullable Integer pageSize,
 			@Nullable CqlIdentifier routingKeyspace, @Nullable ByteBuffer routingKey,
-			@Nullable ConsistencyLevel serialConsistencyLevel, Duration timeout, Duration ttl, @Nullable Long timestamp,
+			@Nullable ConsistencyLevel serialConsistencyLevel, Duration timeout, @Nullable Duration ttl,
+			@Nullable Long timestamp,
 			@Nullable Boolean tracing) {
 
 		super(consistencyLevel, executionProfileResolver, idempotent, keyspace, pageSize, routingKeyspace, routingKey,
@@ -92,8 +93,9 @@ public WriteOptionsBuilder mutate() {
 	}
 
 	/**
-	 * @return the time to live, if set.
+	 * @return the time to live, if set, otherwise {@literal null}.
 	 */
+	@Nullable
 	public Duration getTtl() {
 		return this.ttl;
 	}

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/StatementFactoryUnitTests.java

@@ -339,11 +339,12 @@ void insertShouldApplyQueryOptions() {
 		StatementBuilder<RegularInsert> insert = statementFactory.insert(person, queryOptions);
 
 		SimpleStatement statement = insert.build();
+		assertThat(statement.getQuery()).isEqualTo("INSERT INTO person (id) VALUES (?)");
 		assertThat(statement.getExecutionProfileName()).isEqualTo("foo");
 		assertThat(statement.getSerialConsistencyLevel()).isEqualTo(DefaultConsistencyLevel.QUORUM);
 	}
 
-	@Test // GH-1401
+	@Test // GH-1401, GH-1535
 	void insertWithOptionsShouldRenderBindMarkers() {
 
 		Person person = new Person();
@@ -358,6 +359,16 @@ void insertWithOptionsShouldRenderBindMarkers() {
 
 		assertThat(statement.getQuery()).isEqualTo("INSERT INTO person (id) VALUES (?) USING TIMESTAMP ? AND TTL ?");
 		assertThat(statement.getPositionalValues()).containsExactly("foo", 1234L, 10);
+
+		queryOptions = WriteOptions.builder() //
+				.ttl(Duration.ZERO).timestamp(1234).build();
+
+		insert = statementFactory.insert(person, queryOptions);
+
+		statement = insert.build(ParameterHandling.BY_INDEX);
+
+		assertThat(statement.getQuery()).isEqualTo("INSERT INTO person (id) VALUES (?) USING TIMESTAMP ? AND TTL ?");
+		assertThat(statement.getPositionalValues()).containsExactly("foo", 1234L, 0);
 	}
 
 	@Test // DATACASS-656