DATACASS-791 - Suppress results for suspended query methods returning kotlin.Unit.

We now discard results for suspended query methods if the return type is kotlin.Unit.

Related ticket: DATACMNS-1779

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraQueryExecution.java

@@ -35,6 +35,7 @@ import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.mapping.model.EntityInstantiators;
 import org.springframework.data.repository.query.ResultProcessor;
 import org.springframework.data.repository.query.ReturnedType;
+import org.springframework.data.util.ReflectionUtils;
 import org.springframework.util.ClassUtils;
 
 import com.datastax.oss.driver.api.core.cql.Row;
@@ -256,7 +257,7 @@ interface ReactiveCassandraQueryExecution {
 
 			ReturnedType returnedType = processor.getReturnedType();
 
-			if (returnedType.getReturnedType().equals(Void.class)) {
+			if (ReflectionUtils.isVoid(returnedType.getReturnedType())) {
 
 				if (source instanceof Mono) {
 					return ((Mono<?>) source).then();

spring-data-cassandra/src/test/kotlin/org/springframework/data/cassandra/repository/CoroutineRepositoryUnitTests.kt

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2020 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.cassandra.repository
+
+import com.datastax.oss.driver.api.core.cql.Statement
+import io.mockk.every
+import io.mockk.mockk
+import kotlinx.coroutines.runBlocking
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+import org.springframework.data.annotation.Id
+import org.springframework.data.cassandra.ReactiveResultSet
+import org.springframework.data.cassandra.core.ReactiveCassandraOperations
+import org.springframework.data.cassandra.core.convert.MappingCassandraConverter
+import org.springframework.data.cassandra.core.cql.ReactiveCqlOperations
+import org.springframework.data.cassandra.core.mapping.CassandraMappingContext
+import org.springframework.data.cassandra.repository.support.ReactiveCassandraRepositoryFactory
+import org.springframework.data.repository.kotlin.CoroutineCrudRepository
+import reactor.core.publisher.Mono
+
+/**
+ * Unit tests for Kotlin Coroutine repositories.
+ *
+ * @author Mark Paluch
+ */
+class CoroutineRepositoryUnitTests {
+
+	val operations = mockk<ReactiveCassandraOperations>(relaxed = true)
+	val cqlOperations = mockk<ReactiveCqlOperations>(relaxed = true)
+	val resultSet = mockk<ReactiveResultSet>(relaxed = true)
+	lateinit var repositoryFactory: ReactiveCassandraRepositoryFactory
+
+	@BeforeEach
+	fun before() {
+
+		every { operations.getConverter() } returns MappingCassandraConverter(CassandraMappingContext())
+		every { operations.reactiveCqlOperations } returns cqlOperations
+		repositoryFactory = ReactiveCassandraRepositoryFactory(operations)
+	}
+
+	@Test // DATACASS-791
+	fun `should discard result of suspended query method without result`() {
+
+		every { resultSet.wasApplied() } returns true
+		every { cqlOperations.queryForResultSet(any<Statement<*>>()) } returns Mono.just(resultSet)
+
+		val repository = repositoryFactory.getRepository(PersonRepository::class.java)
+
+		runBlocking {
+			repository.deleteAllByName("foo")
+		}
+	}
+
+	interface PersonRepository : CoroutineCrudRepository<Person, Long> {
+
+		suspend fun deleteAllByName(name: String)
+	}
+
+	data class Person(@Id var id: Long, var name: String)
+}