Throw proper DecodingException in Kotlin Serialization decoders

Closes gh-33138

spring-web/src/main/java/org/springframework/http/codec/KotlinSerializationBinaryDecoder.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2024 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.
@@ -109,7 +109,15 @@ public abstract class KotlinSerializationBinaryDecoder<T extends BinaryFormat> e
 			}
 			return this.byteArrayDecoder
 					.decodeToMono(inputStream, elementType, mimeType, hints)
-					.map(byteArray -> format().decodeFromByteArray(serializer, byteArray));
+					.handle((byteArray, sink) -> {
+						try {
+							sink.next(format().decodeFromByteArray(serializer, byteArray));
+							sink.complete();
+						}
+						catch (IllegalArgumentException ex) {
+							sink.error(new DecodingException("Decoding error: " + ex.getMessage(), ex));
+						}
+					});
 		});
 	}
 

spring-web/src/main/java/org/springframework/http/codec/KotlinSerializationStringDecoder.java

@@ -26,6 +26,7 @@ import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import org.springframework.core.ResolvableType;
+import org.springframework.core.codec.CodecException;
 import org.springframework.core.codec.Decoder;
 import org.springframework.core.codec.DecodingException;
 import org.springframework.core.codec.StringDecoder;
@@ -101,7 +102,14 @@ public abstract class KotlinSerializationStringDecoder<T extends StringFormat> e
 			}
 			return this.stringDecoder
 					.decode(inputStream, elementType, mimeType, hints)
-					.map(string -> format().decodeFromString(serializer, string));
+					.handle((string, sink) -> {
+						try {
+							sink.next(format().decodeFromString(serializer, string));
+						}
+						catch (IllegalArgumentException ex) {
+							sink.error(processException(ex));
+						}
+					});
 		});
 	}
 
@@ -115,8 +123,20 @@ public abstract class KotlinSerializationStringDecoder<T extends StringFormat> e
 			}
 			return this.stringDecoder
 					.decodeToMono(inputStream, elementType, mimeType, hints)
-					.map(string -> format().decodeFromString(serializer, string));
+					.handle((string, sink) -> {
+						try {
+							sink.next(format().decodeFromString(serializer, string));
+							sink.complete();
+						}
+						catch (IllegalArgumentException ex) {
+							sink.error(processException(ex));
+						}
+					});
 		});
 	}
 
+	private CodecException processException(IllegalArgumentException ex) {
+		return new DecodingException("Decoding error: " + ex.getMessage(), ex);
+	}
+
 }

spring-web/src/test/kotlin/org/springframework/http/codec/json/KotlinSerializationJsonDecoderTests.kt

@@ -21,14 +21,13 @@ import org.assertj.core.api.Assertions.assertThat
 import org.junit.jupiter.api.Test
 import org.springframework.core.Ordered
 import org.springframework.core.ResolvableType
+import org.springframework.core.codec.DecodingException
 import org.springframework.core.io.buffer.DataBuffer
 import org.springframework.core.testfixture.codec.AbstractDecoderTests
 import org.springframework.http.MediaType
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
-import reactor.test.StepVerifier
 import reactor.test.StepVerifier.FirstStep
-import java.lang.UnsupportedOperationException
 import java.math.BigDecimal
 import java.nio.charset.Charset
 import java.nio.charset.StandardCharsets
@@ -82,6 +81,29 @@ class KotlinSerializationJsonDecoderTests : AbstractDecoderTests<KotlinSerializa
 		}, null, null)
 	}
 
+	@Test
+	fun decodeWithUnexpectedFormat() {
+		val input = Flux.concat(
+			stringBuffer("{\"ba\":\"b1\",\"fo\":\"f1\"}\n"),
+		)
+
+		testDecode(input, ResolvableType.forClass(Pojo::class.java), { step: FirstStep<Pojo> ->
+			step
+				.expectError(DecodingException::class.java)
+				.verify() }, null, null)
+	}
+
+	@Test
+	fun decodeToMonoWithUnexpectedFormat() {
+		val input = Flux.concat(
+			stringBuffer("{\"ba\":\"b1\",\"fo\":\"f1\"}\n"),
+		)
+
+		testDecodeToMono(input, ResolvableType.forClass(Pojo::class.java), { step: FirstStep<Pojo> ->
+			step.expectError(DecodingException::class.java)
+				.verify() }, null, null)
+	}
+
 	@Test
 	fun decodeStreamWithSingleBuffer() {
 		val input = Flux.concat(

spring-web/src/test/kotlin/org/springframework/http/codec/protobuf/KotlinSerializationProtobufDecoderTests.kt

@@ -24,6 +24,7 @@ import org.assertj.core.api.Assertions.assertThat
 import org.junit.jupiter.api.Test
 import org.springframework.core.Ordered
 import org.springframework.core.ResolvableType
+import org.springframework.core.codec.DecodingException
 import org.springframework.core.io.buffer.DataBuffer
 import org.springframework.core.testfixture.codec.AbstractDecoderTests
 import org.springframework.http.MediaType
@@ -86,6 +87,21 @@ class KotlinSerializationProtobufDecoderTests : AbstractDecoderTests<KotlinSeria
 		}, null, null)
 	}
 
+	@Test
+	fun decodeToMonoWithUnexpectedFormat() {
+		val input = Mono.just(
+			bufferFactory.allocateBuffer(0),
+		)
+
+		val elementType = ResolvableType.forClass(Pojo::class.java)
+
+		testDecodeToMono(input, elementType, { step: FirstStep<Any> ->
+			step
+				.expectError(DecodingException::class.java)
+				.verify()
+		}, null, null)
+	}
+
 	private fun byteBuffer(value: Any): Mono<DataBuffer> {
 		return Mono.defer {
 			val bytes = ProtoBuf.Default.encodeToByteArray(serializer(Pojo::class.java), value)