Fix ordering of releasing resources in JSON Encoder

Prior to this commit, the Jackson 2.x encoders, in case of encoding a
stream of data, would first release the `ByteArrayBuilder` and then the
`JsonGenerator`. This order is inconsistent with the single value
variant (see `o.s.h.codec.json.AbstractJackson2Encoder#encodeValue`) and
invalid since the `JsonGenerator` uses internally the
`ByteArrayBuilder`.

In case of a CSV Encoder, the codec can buffer data to write the column
names of the CSV file. Writing an empty Flux with this Encoder would not
fail but still log a NullPointerException ignored by the reactive
pipeline.

This commit fixes the order and avoid such issues at runtime.

Fixes gh-30493

spring-web/spring-web.gradle

@@ -78,6 +78,7 @@ dependencies {
 	testImplementation("com.fasterxml.jackson.datatype:jackson-datatype-jsr310")
 	testImplementation("com.fasterxml.jackson.module:jackson-module-kotlin")
 	testImplementation("com.fasterxml.jackson.module:jackson-module-parameter-names")
+	testImplementation("com.fasterxml.jackson.dataformat:jackson-dataformat-csv")
 	testImplementation("com.squareup.okhttp3:mockwebserver")
 	testImplementation("io.micrometer:micrometer-observation-test")
 	testImplementation("io.projectreactor:reactor-test")

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

@@ -205,8 +205,8 @@ public Flux<DataBuffer> encode(Publisher<?> inputStream, DataBufferFactory buffe
 						.doOnNext(dataBuffer -> Hints.touchDataBuffer(dataBuffer, hintsToUse, logger))
 						.doAfterTerminate(() -> {
 							try {
-								byteBuilder.release();
 								generator.close();
+								byteBuilder.release();
 							}
 							catch (IOException ex) {
 								logger.error("Could not close Encoder resources", ex);

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

@@ -0,0 +1,101 @@
+/*
+ * 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.http.codec.json;
+
+import java.util.List;
+import java.util.Map;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectWriter;
+import com.fasterxml.jackson.dataformat.csv.CsvMapper;
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Flux;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.core.testfixture.codec.AbstractEncoderTests;
+import org.springframework.http.MediaType;
+import org.springframework.util.Assert;
+import org.springframework.util.MimeType;
+import org.springframework.web.testfixture.xml.Pojo;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for {@link AbstractJackson2Encoder} for the CSV variant and how resources are managed.
+ * @author Brian Clozel
+ */
+class JacksonCsvEncoderTests extends AbstractEncoderTests<org.springframework.http.codec.json.JacksonCsvEncoderTests.JacksonCsvEncoder> {
+
+	public JacksonCsvEncoderTests() {
+		super(new JacksonCsvEncoder());
+	}
+
+	@Test
+	@Override
+	public void canEncode() throws Exception {
+		ResolvableType pojoType = ResolvableType.forClass(Pojo.class);
+		assertThat(this.encoder.canEncode(pojoType, JacksonCsvEncoder.TEXT_CSV)).isTrue();
+	}
+
+	@Test
+	@Override
+	public void encode() throws Exception {
+		Flux<Object> input = Flux.just(new Pojo("spring", "framework"),
+				new Pojo("spring", "data"),
+				new Pojo("spring", "boot"));
+
+		testEncode(input, Pojo.class, step -> step
+				.consumeNextWith(expectString("bar,foo\nframework,spring\n"))
+				.consumeNextWith(expectString("data,spring\n"))
+				.consumeNextWith(expectString("boot,spring\n"))
+				.verifyComplete());
+	}
+
+	@Test
+	// See gh-30493
+	// this test did not fail directly but logged a NullPointerException dropped by the reactive pipeline
+	void encodeEmptyFlux() {
+		Flux<Object> input = Flux.empty();
+		testEncode(input, Pojo.class, step -> step.verifyComplete());
+	}
+
+	static class JacksonCsvEncoder extends AbstractJackson2Encoder {
+		public static final MediaType TEXT_CSV = new MediaType("text", "csv");
+
+		public JacksonCsvEncoder() {
+			this(CsvMapper.builder().build(), TEXT_CSV);
+		}
+
+		@Override
+		protected byte[] getStreamingMediaTypeSeparator(MimeType mimeType) {
+			// CsvMapper emits newlines
+			return new byte[0];
+		}
+
+		public JacksonCsvEncoder(ObjectMapper mapper, MimeType... mimeTypes) {
+			super(mapper, mimeTypes);
+			Assert.isInstanceOf(CsvMapper.class, mapper);
+			setStreamingMediaTypes(List.of(TEXT_CSV));
+		}
+
+		@Override
+		protected ObjectWriter customizeWriter(ObjectWriter writer, MimeType mimeType, ResolvableType elementType, Map<String, Object> hints) {
+			var mapper = (CsvMapper) getObjectMapper();
+			return writer.with(mapper.schemaFor(elementType.toClass()).withHeader());
+		}
+	}
+}