Corrupt multipart should not hang SynchronossPartHttpMessageReader

This commit notifies the Synchronoss listener that the buffer stream
has ended.

Closes gh-23768

spring-web/src/test/java/org/springframework/http/codec/multipart/SynchronossPartHttpMessageReaderTests.java

@@ -18,6 +18,7 @@ package org.springframework.http.codec.multipart;
 
 import java.io.File;
 import java.io.IOException;
+import java.nio.channels.ReadableByteChannel;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.Map;
@@ -46,6 +47,7 @@ import org.springframework.mock.http.server.reactive.test.MockServerHttpRequest;
 import org.springframework.util.MultiValueMap;
 
 import static java.util.Collections.emptyMap;
+import static java.util.Collections.singletonMap;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.springframework.core.ResolvableType.forClassWithGenerics;
 import static org.springframework.http.HttpHeaders.CONTENT_TYPE;
@@ -142,6 +144,24 @@ public class SynchronossPartHttpMessageReaderTests extends AbstractLeakCheckingT
 		subscriber.cancel();
 	}
 
+	@Test
+	void gh23768() throws IOException {
+		ReadableByteChannel channel = new ClassPathResource("invalid.multipart", getClass()).readableChannel();
+		Flux<DataBuffer> body = DataBufferUtils.readByteChannel(() -> channel, this.bufferFactory, 1024);
+
+		MediaType contentType = new MediaType("multipart", "form-data",
+				singletonMap("boundary", "NbjrKgjbsaMLdnMxMfDpD6myWomYc0qNX0w"));
+		ServerHttpRequest request = MockServerHttpRequest.post("/")
+				.contentType(contentType)
+				.body(body);
+
+		Mono<MultiValueMap<String, Part>> parts = this.reader.readMono(PARTS_ELEMENT_TYPE, request, emptyMap());
+
+		StepVerifier.create(parts)
+				.assertNext(result -> assertThat(result).isEmpty())
+				.verifyComplete();
+	}
+
 	@Test
 	void readTooManyParts() {
 		testMultipartExceptions(reader -> reader.setMaxParts(1), ex -> {