Refine TransactionalOperator.executeAndAwait nullability

Before this commit, TransactionalOperator.executeAndAwait had a rigid
null-safety handling. This commit takes advantage of Kotlin capability
to allow to deal with both non-null and nullable depending on the
nullability of the lambda.

Closes gh-29919

spring-tx/src/main/kotlin/org/springframework/transaction/reactive/TransactionalOperatorExtensions.kt

@@ -1,3 +1,19 @@
+/*
+ * Copyright 2002-2023 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.transaction.reactive
 
 import java.util.Optional
@@ -26,6 +42,6 @@ fun <T : Any> Flow<T>.transactional(operator: TransactionalOperator): Flow<T> =
 * @author Mark Paluch
 * @since 5.2
 */
-suspend fun <T : Any> TransactionalOperator.executeAndAwait(f: suspend (ReactiveTransaction) -> T?): T? =
-		execute { status -> mono(Dispatchers.Unconfined) { f(status) } }.map { value -> Optional.of(value) }
+suspend fun <T> TransactionalOperator.executeAndAwait(f: suspend (ReactiveTransaction) -> T): T =
+		execute { status -> mono(Dispatchers.Unconfined) { f(status) } }.map { value -> Optional.ofNullable(value) }
 				.defaultIfEmpty(Optional.empty()).awaitLast().orElse(null)

spring-tx/src/test/kotlin/org/springframework/transaction/reactive/TransactionalOperatorExtensionsTests.kt

@@ -22,7 +22,6 @@ import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.runBlocking
 import org.assertj.core.api.Assertions.assertThat
 import org.junit.jupiter.api.Test
-import org.junit.jupiter.api.fail
 import org.springframework.transaction.support.DefaultTransactionDefinition
 
 class TransactionalOperatorExtensionsTests {
@@ -30,10 +29,11 @@ class TransactionalOperatorExtensionsTests {
 	private val tm = ReactiveTestTransactionManager(false, true)
 
 	@Test
+	@Suppress("UNUSED_VARIABLE")
 	fun commitWithSuspendingFunction() {
 		val operator = TransactionalOperator.create(tm, DefaultTransactionDefinition())
 		runBlocking {
-			operator.executeAndAwait {
+			val returnValue: Boolean = operator.executeAndAwait {
 				delay(1)
 				true
 			}
@@ -43,10 +43,11 @@ class TransactionalOperatorExtensionsTests {
 	}
 
 	@Test
+	@Suppress("UNUSED_VARIABLE")
 	fun commitWithEmptySuspendingFunction() {
 		val operator = TransactionalOperator.create(tm, DefaultTransactionDefinition())
 		runBlocking {
-			operator.executeAndAwait {
+			val returnValue: Boolean? = operator.executeAndAwait {
 				delay(1)
 				null
 			}
@@ -69,7 +70,6 @@ class TransactionalOperatorExtensionsTests {
 				assertThat(tm.rollback).isTrue()
 				return@runBlocking
 			}
-			fail("")
 		}
 	}