Fix error handling in Jackson2JsonDecoder

Issue: SPR-15112

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

@@ -118,7 +118,7 @@ public class Jackson2JsonDecoder extends AbstractJackson2Codec implements Decode
 						return value;
 					}
 					catch (IOException ex) {
-						return Flux.error(new CodecException("Error while reading the data", ex));
+						throw new CodecException("Error while reading the data", ex);
 					}
 				});
 	}

spring-web/src/test/java/org/springframework/http/codec/json/Jackson2JsonDecoderTests.java

@@ -28,6 +28,7 @@ import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
 import org.springframework.core.ResolvableType;
+import org.springframework.core.codec.CodecException;
 import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.http.MediaType;
@@ -67,6 +68,16 @@ public class Jackson2JsonDecoderTests extends AbstractDataBufferAllocatingTestCa
 				.verify();
 	}
 
+	@Test
+	public void decodePojoWithError() throws Exception {
+		Flux<DataBuffer> source = Flux.just(stringBuffer("{\"foo\":}"));
+		ResolvableType elementType = ResolvableType.forClass(Pojo.class);
+		Flux<Object> flux = new Jackson2JsonDecoder().decode(source, elementType, null,
+				Collections.emptyMap());
+
+		StepVerifier.create(flux).verifyError(CodecException.class);
+	}
+
 	@Test
 	public void decodeToList() throws Exception {
 		Flux<DataBuffer> source = Flux.just(stringBuffer(