Ensure Jackson generator flushed

Closes gh-22771

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

@@ -172,6 +172,7 @@ public abstract class AbstractJackson2Encoder extends Jackson2CodecSupport imple
 		try {
 			JsonGenerator generator = getObjectMapper().getFactory().createGenerator(outputStream, encoding);
 			writer.writeValue(generator, value);
+			generator.flush();
 			release = false;
 		}
 		catch (InvalidDefinitionException ex) {

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

@@ -17,6 +17,7 @@
 package org.springframework.http.codec.json;
 
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.Map;
@@ -24,17 +25,21 @@ import java.util.Map;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import com.fasterxml.jackson.annotation.JsonTypeName;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
 import org.junit.Test;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
 
 import org.springframework.core.ResolvableType;
 import org.springframework.core.codec.AbstractEncoderTestCase;
+import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.http.MediaType;
 import org.springframework.http.codec.Pojo;
 import org.springframework.http.codec.ServerSentEvent;
 import org.springframework.util.MimeType;
+import org.springframework.util.MimeTypeUtils;
 
 import static java.util.Collections.singletonMap;
 import static org.junit.Assert.*;
@@ -85,8 +90,6 @@ public class Jackson2JsonEncoderTests extends AbstractEncoderTestCase<Jackson2Js
 				.consumeNextWith(expectString("{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}\n"))
 				.verifyComplete(),
 				APPLICATION_STREAM_JSON, null);
-
-
 	}
 
 	@Test // SPR-15866
@@ -200,6 +203,21 @@ public class Jackson2JsonEncoderTests extends AbstractEncoderTestCase<Jackson2Js
 				null, hints);
 	}
 
+	@Test // gh-22771
+	public void encodeWithFlushAfterWriteOff() {
+		ObjectMapper mapper = new ObjectMapper();
+		mapper.configure(SerializationFeature.FLUSH_AFTER_WRITE_VALUE, false);
+		Jackson2JsonEncoder encoder = new Jackson2JsonEncoder(mapper);
+
+		Flux<DataBuffer> result = encoder.encode(Flux.just(new Pojo("foo", "bar")), this.bufferFactory,
+				ResolvableType.forClass(Pojo.class), MimeTypeUtils.APPLICATION_JSON, Collections.emptyMap());
+
+		StepVerifier.create(result)
+				.consumeNextWith(expectString("[{\"foo\":\"foo\",\"bar\":\"bar\"}]"))
+				.expectComplete()
+				.verify(Duration.ofSeconds(5));
+	}
+
 
 	@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
 	private static class ParentClass {