Fix error handling regression for non-streaming Flux

Closes gh-29038

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

@@ -20,6 +20,7 @@ import java.io.IOException;
 import java.lang.annotation.Annotation;
 import java.nio.charset.Charset;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
@@ -47,6 +48,7 @@ import org.springframework.core.codec.EncodingException;
 import org.springframework.core.codec.Hints;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.core.io.buffer.DefaultDataBufferFactory;
 import org.springframework.core.log.LogFormatUtils;
 import org.springframework.http.MediaType;
 import org.springframework.http.codec.HttpMessageEncoder;
@@ -73,6 +75,8 @@ public abstract class AbstractJackson2Encoder extends Jackson2CodecSupport imple
 
 	private static final byte[] EMPTY_BYTES = new byte[0];
 
+	private static DataBuffer EMPTY_BUFFER = DefaultDataBufferFactory.sharedInstance.wrap(EMPTY_BYTES);
+
 	private static final Map<String, JsonEncoding> ENCODINGS;
 
 	static {
@@ -174,11 +178,21 @@ public abstract class AbstractJackson2Encoder extends Jackson2CodecSupport imple
 			}
 			else {
 				JsonArrayJoinHelper helper = new JsonArrayJoinHelper();
-				return Flux.concat(
-						helper.getPrefix(bufferFactory, hints, logger),
-						Flux.from(inputStream).map(value -> encodeStreamingValue(
-								value, bufferFactory, hints, sequenceWriter, byteBuilder, helper.getDelimiter(), EMPTY_BYTES)),
-						helper.getSuffix(bufferFactory, hints, logger));
+
+				// Do not prepend JSON array prefix until first signal is known, onNext vs onError
+				// Keeps response not committed for error handling
+
+				Flux<DataBuffer> flux1 = helper.getPrefix(bufferFactory, hints, logger)
+						.concatWith(Flux.just(EMPTY_BUFFER).repeat());
+
+				Flux<DataBuffer> flux2 = Flux.from(inputStream).map(value -> encodeStreamingValue(
+						value, bufferFactory, hints, sequenceWriter, byteBuilder, helper.getDelimiter(), EMPTY_BYTES));
+
+				dataBufferFlux = Flux.zip(flux1, flux2, (buffer1, buffer2) ->
+								(buffer1 != EMPTY_BUFFER ?
+										bufferFactory.join(Arrays.asList(buffer1, buffer2)) :
+										buffer2))
+						.concatWith(helper.getSuffix(bufferFactory, hints, logger));
 			}
 
 			return dataBufferFlux

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

@@ -137,21 +137,30 @@ public class Jackson2JsonEncoderTests extends AbstractEncoderTests<Jackson2JsonE
 		);
 
 		testEncode(input, Pojo.class, step -> step
-				.consumeNextWith(expectString("["))
-				.consumeNextWith(expectString("{\"foo\":\"foo\",\"bar\":\"bar\"}"))
+				.consumeNextWith(expectString("[{\"foo\":\"foo\",\"bar\":\"bar\"}"))
 				.consumeNextWith(expectString(",{\"foo\":\"foofoo\",\"bar\":\"barbar\"}"))
 				.consumeNextWith(expectString(",{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}"))
 				.consumeNextWith(expectString("]"))
 				.verifyComplete());
 	}
 
+	@Test // gh-29038
+	void encodeNonStreamWithErrorAsFirstSignal() {
+		String message = "I'm a teapot";
+		Flux<Object> input = Flux.error(new IllegalStateException(message));
+
+		Flux<DataBuffer> output = this.encoder.encode(
+				input, this.bufferFactory, ResolvableType.forClass(Pojo.class), null, null);
+
+		StepVerifier.create(output).expectErrorMessage(message).verify();
+	}
+
 	@Test
 	public void encodeWithType() {
 		Flux<ParentClass> input = Flux.just(new Foo(), new Bar());
 
 		testEncode(input, ParentClass.class, step -> step
-				.consumeNextWith(expectString("["))
-				.consumeNextWith(expectString("{\"type\":\"foo\"}"))
+				.consumeNextWith(expectString("[{\"type\":\"foo\"}"))
 				.consumeNextWith(expectString(",{\"type\":\"bar\"}"))
 				.consumeNextWith(expectString("]"))
 				.verifyComplete());
@@ -159,7 +168,7 @@ public class Jackson2JsonEncoderTests extends AbstractEncoderTests<Jackson2JsonE
 
 
 	@Test  // SPR-15727
-	public void encodeAsStreamWithCustomStreamingType() {
+	public void encodeStreamWithCustomStreamingType() {
 		MediaType fooMediaType = new MediaType("application", "foo");
 		MediaType barMediaType = new MediaType("application", "bar");
 		this.encoder.setStreamingMediaTypes(Arrays.asList(fooMediaType, barMediaType));
@@ -263,8 +272,7 @@ public class Jackson2JsonEncoderTests extends AbstractEncoderTests<Jackson2JsonE
 				ResolvableType.forClass(Pojo.class), MimeTypeUtils.APPLICATION_JSON, Collections.emptyMap());
 
 		StepVerifier.create(result)
-				.consumeNextWith(expectString("["))
-				.consumeNextWith(expectString("{\"foo\":\"foo\",\"bar\":\"bar\"}"))
+				.consumeNextWith(expectString("[{\"foo\":\"foo\",\"bar\":\"bar\"}"))
 				.consumeNextWith(expectString("]"))
 				.expectComplete()
 				.verify(Duration.ofSeconds(5));