From c86396a60f31aa01adadd122178878afe134dbe5 Mon Sep 17 00:00:00 2001 From: jgjesdal Date: Tue, 15 Sep 2026 13:46:46 +0200 Subject: [PATCH 1/3] fix(api): a failure below the body cache is not a 200 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CachingBodyFilter caught ServletException/IOException from the chain, logged it and returned. Nothing then set a status, so the response went out at Tomcat's default 200 with an empty body — a request the server dropped, reported to the caller as a success. DispatcherServlet wraps everything a handler throws, Errors included, into "Handler dispatch failed", so this masked every unhandled failure on every non-streaming endpoint. Seen in the e2e run as POST /timeseries/data answering 200 instead of 204 while the api was throwing OutOfMemoryError. copyBodyToResponse() moves into a finally so a body written before the failure still reaches the caller instead of being discarded. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: jgjesdal --- .../api/filters/CachingBodyFilter.java | 26 +++-- .../api/filters/CachingBodyFilterTest.java | 103 ++++++++++++++++++ 2 files changed, 119 insertions(+), 10 deletions(-) create mode 100644 datahub-api/src/test/java/ai/intellistream/datahub/api/filters/CachingBodyFilterTest.java diff --git a/datahub-api/src/main/java/ai/intellistream/datahub/api/filters/CachingBodyFilter.java b/datahub-api/src/main/java/ai/intellistream/datahub/api/filters/CachingBodyFilter.java index 5ddf6023..d322f422 100644 --- a/datahub-api/src/main/java/ai/intellistream/datahub/api/filters/CachingBodyFilter.java +++ b/datahub-api/src/main/java/ai/intellistream/datahub/api/filters/CachingBodyFilter.java @@ -4,19 +4,26 @@ import jakarta.servlet.*; import jakarta.servlet.http.HttpServletRequest; import jakarta.servlet.http.HttpServletResponse; -import lombok.extern.slf4j.Slf4j; import org.springframework.web.util.ContentCachingRequestWrapper; import org.springframework.web.util.ContentCachingResponseWrapper; import java.io.IOException; -@Slf4j public class CachingBodyFilter implements Filter { // https://stackoverflow.com/questions/39935190/contentcachingresponsewrapper-produces-empty-response + /** + * Nothing is caught here, deliberately. This filter exists to make bodies loggable, so a + * failure below it is never its business to handle: swallowing one would leave the response + * at Tomcat's default 200 with an empty body, reporting a request that was dropped on the + * floor as a success. DispatcherServlet wraps everything a handler throws — {@code Error}s + * included — into {@code ServletException: Handler dispatch failed}, so a catch here would + * mask every unhandled failure on every endpoint, not just the odd I/O fault. + */ @Override - public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) { + public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) + throws IOException, ServletException { HttpServletRequest httpRequest = (HttpServletRequest) request; // Streaming endpoints must NOT be wrapped: a file download or graph export response can be @@ -26,11 +33,7 @@ public void doFilter(ServletRequest request, ServletResponse response, FilterCha // wrappers are present, so passing the raw request/response through simply skips body // logging here. if (StreamingEndpoints.matches(httpRequest)) { - try { - chain.doFilter(request, response); - } catch (IOException | ServletException e) { - log.error("Error in streaming request", e); - } + chain.doFilter(request, response); return; } @@ -38,9 +41,12 @@ public void doFilter(ServletRequest request, ServletResponse response, FilterCha ContentCachingResponseWrapper resWrapper = new ContentCachingResponseWrapper((HttpServletResponse) response); try { chain.doFilter(reqWrapper, resWrapper); + } finally { + // In a finally, not after the call: whatever was buffered before a failure has to reach + // the real response either way, or an error body written further down is discarded and + // the caller gets an empty one. Copying an empty buffer is a no-op, so the container is + // still free to write its own error page over an uncommitted response. resWrapper.copyBodyToResponse(); - } catch (IOException | ServletException e) { - log.error("Error extracting body", e); } } diff --git a/datahub-api/src/test/java/ai/intellistream/datahub/api/filters/CachingBodyFilterTest.java b/datahub-api/src/test/java/ai/intellistream/datahub/api/filters/CachingBodyFilterTest.java new file mode 100644 index 00000000..8783f203 --- /dev/null +++ b/datahub-api/src/test/java/ai/intellistream/datahub/api/filters/CachingBodyFilterTest.java @@ -0,0 +1,103 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later +package ai.intellistream.datahub.api.filters; + +import jakarta.servlet.FilterChain; +import jakarta.servlet.ServletException; +import org.junit.jupiter.api.Test; +import org.springframework.http.HttpStatus; +import org.springframework.http.MediaType; +import org.springframework.mock.web.MockFilterChain; +import org.springframework.mock.web.MockHttpServletRequest; +import org.springframework.mock.web.MockHttpServletResponse; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Body caching must not change what the caller is told happened. + * + *

The filter used to log a failure below it and return, which left the response at Tomcat's + * default 200 with an empty body — a request the server dropped, reported to the client as a + * success. DispatcherServlet wraps everything a handler throws into {@code ServletException}, so + * that applied to every unhandled failure on every non-streaming endpoint. + */ +class CachingBodyFilterTest { + + private final CachingBodyFilter filter = new CachingBodyFilter(); + private final MockHttpServletResponse response = new MockHttpServletResponse(); + + private static MockHttpServletRequest post(String uri) { + MockHttpServletRequest request = new MockHttpServletRequest("POST", uri); + request.setContent("{\"items\":[]}".getBytes(StandardCharsets.UTF_8)); + request.setContentType(MediaType.APPLICATION_JSON_VALUE); + return request; + } + + @Test + void aFailureBelowTheFilterIsNotReportedAsSuccess() { + FilterChain exploding = (req, res) -> { + throw new ServletException("Handler dispatch failed", new OutOfMemoryError("Java heap space")); + }; + + assertThatThrownBy(() -> filter.doFilter(post("/timeseries/data"), response, exploding)) + .isInstanceOf(ServletException.class) + .hasMessage("Handler dispatch failed"); + + // Left uncommitted, so the container is still free to turn this into a 500. + assertThat(response.isCommitted()).isFalse(); + assertThat(response.getContentAsByteArray()).isEmpty(); + } + + @Test + void anIoFailureBelowTheFilterAlsoPropagates() { + FilterChain exploding = (req, res) -> { + throw new IOException("broken pipe"); + }; + + assertThatThrownBy(() -> filter.doFilter(post("/events/create"), response, exploding)) + .isInstanceOf(IOException.class); + } + + @Test + void aFailureOnAStreamingEndpointAlsoPropagates() { + FilterChain exploding = (req, res) -> { + throw new ServletException("Handler dispatch failed"); + }; + + assertThatThrownBy(() -> filter.doFilter(post("/resources/import"), response, exploding)) + .isInstanceOf(ServletException.class); + } + + @Test + void aBodyWrittenBeforeAFailureStillReachesTheCaller() throws Exception { + FilterChain writesThenFails = (req, res) -> { + res.getWriter().write("{\"error\":\"boom\"}"); + throw new ServletException("Handler dispatch failed"); + }; + + assertThatThrownBy(() -> filter.doFilter(post("/events/create"), response, writesThenFails)) + .isInstanceOf(ServletException.class); + + assertThat(response.getContentAsString()).isEqualTo("{\"error\":\"boom\"}"); + } + + @Test + void aSuccessfulResponseIsCopiedThroughExactlyOnce() throws Exception { + MockFilterChain chain = new MockFilterChain() { + @Override + public void doFilter(jakarta.servlet.ServletRequest req, jakarta.servlet.ServletResponse res) + throws IOException, ServletException { + res.getWriter().write("{\"items\":[]}"); + super.doFilter(req, res); + } + }; + + filter.doFilter(post("/events/create"), response, chain); + + assertThat(response.getStatus()).isEqualTo(HttpStatus.OK.value()); + assertThat(response.getContentAsString()).isEqualTo("{\"items\":[]}"); + } +} From 5fa1aee55845a3e49e90243f06fd9c3a64b12979 Mon Sep 17 00:00:00 2001 From: olav Date: Tue, 15 Sep 2026 13:45:34 +0200 Subject: [PATCH 2/3] fix(api): a request that fails is never answered 200 CachingBodyFilter caught IOException and ServletException from the rest of the chain and only logged them. DispatcherServlet wraps anything no handler answered in a ServletException, an Error included, so every such failure left the filter with no status set and went out as 200 with an empty body. The e2e's Rust `test_datapoints` hit it: 52 concurrent 100 000-point JSON inserts against the CI api's 384 MiB heap (one such body costs about 20 MiB parsed), and the SDK got 200 where the contract says 204. Every SDK treats a 2xx as stored. The filter now lets the failure propagate on both of its paths, so the container answers it with a 500 the SDKs retry. The body is copied to the response only on success; copying a half-written one would commit the response before the 500 could be sent. DatapointInsertHttpTest drives POST /timeseries/data over real HTTP through the production filter chain with the service throwing OutOfMemoryError: 500 now, and `expected: 500 but was: 200` against the old filter. CachingBodyFilterTest pins propagation on the wrapped and streaming paths. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: olav --- .../api/filters/CachingBodyFilter.java | 30 ++-- .../controllers/DatapointInsertHttpTest.java | 151 ++++++++++++++++++ .../api/filters/CachingBodyFilterTest.java | 79 +++++++++ 3 files changed, 246 insertions(+), 14 deletions(-) create mode 100644 datahub-api/src/test/java/ai/intellistream/datahub/api/controllers/DatapointInsertHttpTest.java create mode 100644 datahub-api/src/test/java/ai/intellistream/datahub/api/filters/CachingBodyFilterTest.java diff --git a/datahub-api/src/main/java/ai/intellistream/datahub/api/filters/CachingBodyFilter.java b/datahub-api/src/main/java/ai/intellistream/datahub/api/filters/CachingBodyFilter.java index 5ddf6023..4995950f 100644 --- a/datahub-api/src/main/java/ai/intellistream/datahub/api/filters/CachingBodyFilter.java +++ b/datahub-api/src/main/java/ai/intellistream/datahub/api/filters/CachingBodyFilter.java @@ -4,19 +4,27 @@ import jakarta.servlet.*; import jakarta.servlet.http.HttpServletRequest; import jakarta.servlet.http.HttpServletResponse; -import lombok.extern.slf4j.Slf4j; import org.springframework.web.util.ContentCachingRequestWrapper; import org.springframework.web.util.ContentCachingResponseWrapper; import java.io.IOException; -@Slf4j +/** + * Wraps the request and response so {@code ReqLogService} can log their bodies. + * + *

A failure from further down the chain always propagates. Nothing here may catch it: an + * exception no handler answered has left the status unset, so swallowing it sends the caller a + * {@code 200} with an empty body however much of the request was lost, which is how a datapoint + * insert that failed on the server came back to the SDK as stored. Propagated, the container + * answers it with a {@code 500} the SDKs retry. + */ public class CachingBodyFilter implements Filter { // https://stackoverflow.com/questions/39935190/contentcachingresponsewrapper-produces-empty-response @Override - public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) { + public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) + throws IOException, ServletException { HttpServletRequest httpRequest = (HttpServletRequest) request; // Streaming endpoints must NOT be wrapped: a file download or graph export response can be @@ -26,22 +34,16 @@ public void doFilter(ServletRequest request, ServletResponse response, FilterCha // wrappers are present, so passing the raw request/response through simply skips body // logging here. if (StreamingEndpoints.matches(httpRequest)) { - try { - chain.doFilter(request, response); - } catch (IOException | ServletException e) { - log.error("Error in streaming request", e); - } + chain.doFilter(request, response); return; } ContentCachingRequestWrapper reqWrapper = new ContentCachingRequestWrapper(httpRequest, 1024 * 1024 * 20); ContentCachingResponseWrapper resWrapper = new ContentCachingResponseWrapper((HttpServletResponse) response); - try { - chain.doFilter(reqWrapper, resWrapper); - resWrapper.copyBodyToResponse(); - } catch (IOException | ServletException e) { - log.error("Error extracting body", e); - } + chain.doFilter(reqWrapper, resWrapper); + // Only on success, never in a finally: copying a half-written body would commit the + // response and leave the container unable to answer the failure with its 500. + resWrapper.copyBodyToResponse(); } } diff --git a/datahub-api/src/test/java/ai/intellistream/datahub/api/controllers/DatapointInsertHttpTest.java b/datahub-api/src/test/java/ai/intellistream/datahub/api/controllers/DatapointInsertHttpTest.java new file mode 100644 index 00000000..5840ef7c --- /dev/null +++ b/datahub-api/src/test/java/ai/intellistream/datahub/api/controllers/DatapointInsertHttpTest.java @@ -0,0 +1,151 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later +package ai.intellistream.datahub.api.controllers; + +import ai.intellistream.datahub.api.ApiDatahubApplication; +import ai.intellistream.datahub.api.datasecurity.DataSecurity; +import ai.intellistream.datahub.api.init.pulsar.SubscriptionTopicProvisioner; +import ai.intellistream.datahub.api.services.IngestQuotaService; +import ai.intellistream.datahub.api.services.TimeseriesService; +import ai.intellistream.datahub.clickhouse.ClickHouseClientPool; +import ai.intellistream.datahub.config.InstanceLock; +import ai.intellistream.datahub.tenant.Tenant; +import ai.intellistream.datahub.tenant.TenantConfigService; +import org.apache.pulsar.client.admin.PulsarAdmin; +import org.apache.pulsar.client.api.Producer; +import org.apache.pulsar.client.api.PulsarClient; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.security.oauth2.jwt.Jwt; +import org.springframework.security.oauth2.jwt.JwtDecoder; +import org.springframework.test.context.ActiveProfiles; +import org.springframework.test.context.bean.override.mockito.MockitoBean; + +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * {@code POST /timeseries/data} over real HTTP, in the booted application: the production filter + * chain and the message converters Spring Boot actually assembled, which together decide the status + * a caller sees. The mocks are the start-up network clients, as in {@code SecurityFilterChainTest}. + */ +@SpringBootTest(classes = ApiDatahubApplication.class, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT) +@ActiveProfiles("ctxtest") +class DatapointInsertHttpTest { + + /** Any string does: the JwtDecoder is a mock, and this is the value it is told to accept. */ + private static final String FAKE_BEARER = "dummy-bearer-insert-http"; + private static final String TENANT = "tenant-insert-http"; + + @Value("${local.server.port}") + private int port; + + @Autowired + private TenantConfigService tenantConfigService; + + @MockitoBean + private JwtDecoder jwtDecoder; + + @MockitoBean + private PulsarClient pulsarClient; + + @MockitoBean + private PulsarAdmin pulsarAdmin; + + @MockitoBean + private ClickHouseClientPool clickHouseClientPool; + + @MockitoBean + private SubscriptionTopicProvisioner subscriptionTopicProvisioner; + + @MockitoBean + private InstanceLock instanceLock; + + @MockitoBean(name = "eventMessageProducer") + private Producer eventMessageProducer; + + @MockitoBean(name = "subscriptionNotifyProducer") + private Producer subscriptionNotifyProducer; + + @MockitoBean(name = "allDatapointProducer") + private Producer allDatapointProducer; + + @MockitoBean(name = "httpMessageProducer") + private Producer httpMessageProducer; + + @MockitoBean + private DataSecurity dataSecurity; + + @MockitoBean + private IngestQuotaService ingestQuota; + + @MockitoBean + private TimeseriesService timeseriesService; + + @BeforeEach + void setUp() { + when(jwtDecoder.decode(FAKE_BEARER)).thenReturn(Jwt.withTokenValue(FAKE_BEARER) + .header("alg", "none") + .claim("organization", Map.of("org", Map.of("id", TENANT))) + .claim("realm_access", Map.of("roles", List.of("DATAHUB_ACCESS"))) + .subject("insert-http-test") + .build()); + tenantConfigService.cachedTenants.put(TENANT, new Tenant()); + } + + /** + * The controller answers every {@link Exception} itself, so what escapes it is an {@link Error}, + * and the likeliest one under a large concurrent ingest is running out of heap. The body-cache + * filter used to swallow it: nothing had set a status, so the caller was told {@code 200} and + * every SDK counted the datapoints as stored. + */ + @Test + void anInsertThatRunsOutOfMemoryIsAServerErrorNotA200() throws Exception { + when(timeseriesService.insertDatapoints(any())).thenThrow(new OutOfMemoryError("Java heap space")); + + HttpResponse response = post(""" + {"items":[{"externalId":"sensor_a", + "datapoints":[{"timestamp":1745328000000,"value":"22.4"}]}]}"""); + + // A status the SDKs retry. A 200 here tells the caller the datapoints were stored. + assertThat(response.statusCode()).as(response.body()).isEqualTo(500); + } + + /** + * Strictness is attached by {@code StrictRequestBodyConfig} to the list Spring Boot builds, so + * only a booted application shows it survived: a unit test of the config cannot tell whether + * Boot's converter assembly still calls it. + */ + @Test + void anUnknownFieldInTheBodyIsA400NamingIt() throws Exception { + HttpResponse response = post(""" + {"items":[{"externalId":"sensor_a", "tableEngine":"MERGETREE", + "datapoints":[{"timestamp":1745328000000,"value":"22.4"}]}]}"""); + + assertThat(response.statusCode()).as(response.body()).isEqualTo(400); + assertThat(response.body()).contains("tableEngine"); + verify(timeseriesService, never()).insertDatapoints(any()); + } + + private HttpResponse post(String json) throws Exception { + return HttpClient.newHttpClient().send(HttpRequest.newBuilder() + .uri(URI.create("http://localhost:" + port + "/timeseries/data")) + .header("Authorization", "Bearer " + FAKE_BEARER) + .header("Content-Type", "application/json") + .POST(HttpRequest.BodyPublishers.ofString(json)) + .build(), + HttpResponse.BodyHandlers.ofString()); + } +} diff --git a/datahub-api/src/test/java/ai/intellistream/datahub/api/filters/CachingBodyFilterTest.java b/datahub-api/src/test/java/ai/intellistream/datahub/api/filters/CachingBodyFilterTest.java new file mode 100644 index 00000000..0e5d4650 --- /dev/null +++ b/datahub-api/src/test/java/ai/intellistream/datahub/api/filters/CachingBodyFilterTest.java @@ -0,0 +1,79 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later +package ai.intellistream.datahub.api.filters; + +import jakarta.servlet.FilterChain; +import jakarta.servlet.ServletException; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; +import org.springframework.mock.web.MockHttpServletRequest; +import org.springframework.mock.web.MockHttpServletResponse; +import org.springframework.web.util.ContentCachingRequestWrapper; +import org.springframework.web.util.ContentCachingResponseWrapper; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * A failure below the body cache must reach the container. Swallowed, it left the status unset and + * the caller saw {@code 200} with an empty body for a request the api never finished, which is how a + * datapoint insert that failed on the server was reported as stored. + */ +class CachingBodyFilterTest { + + /** Both branches the filter takes: wrapped, and streaming. */ + @ParameterizedTest + @CsvSource({ + "POST, /timeseries/data", + "PUT, /files", + }) + void aFailureDownTheChainPropagates(String method, String uri) { + // What DispatcherServlet raises for anything no handler answered, an Error included. + ServletException failure = new ServletException("Handler dispatch failed", + new OutOfMemoryError("Java heap space")); + FilterChain chain = (req, res) -> { + throw failure; + }; + MockHttpServletResponse response = new MockHttpServletResponse(); + + assertThatThrownBy(() -> new CachingBodyFilter() + .doFilter(new MockHttpServletRequest(method, uri), response, chain)) + .isSameAs(failure); + assertThat(response.isCommitted()) + .as("left uncommitted, so the container can still answer with a 500") + .isFalse(); + } + + @ParameterizedTest + @CsvSource({ + "POST, /timeseries/data", + "PUT, /files", + }) + void anIoFailureDownTheChainPropagates(String method, String uri) { + IOException failure = new IOException("connection reset"); + FilterChain chain = (req, res) -> { + throw failure; + }; + + assertThatThrownBy(() -> new CachingBodyFilter() + .doFilter(new MockHttpServletRequest(method, uri), new MockHttpServletResponse(), chain)) + .isSameAs(failure); + } + + @Test + void aSuccessfulResponseIsWrappedAndItsBodyReachesTheClient() throws Exception { + MockHttpServletResponse response = new MockHttpServletResponse(); + FilterChain chain = (req, res) -> { + assertThat(req).isInstanceOf(ContentCachingRequestWrapper.class); + assertThat(res).isInstanceOf(ContentCachingResponseWrapper.class); + res.getOutputStream().write("{\"items\":[]}".getBytes(StandardCharsets.UTF_8)); + }; + + new CachingBodyFilter().doFilter(new MockHttpServletRequest("POST", "/timeseries/data/list"), response, chain); + + assertThat(response.getContentAsString()).isEqualTo("{\"items\":[]}"); + } +} From ad7aa776370a73345a67d85f958b43468c393374 Mon Sep 17 00:00:00 2001 From: olav Date: Tue, 15 Sep 2026 13:45:34 +0200 Subject: [PATCH 3/3] refactor(api): attach the strict request-body converter through the converter builder `WebMvcConfigurer.extendMessageConverters(List)` is deprecated for removal in Spring Framework 7 ([removal] warning in the e2e build). The replacement is `configureMessageConverters(HttpMessageConverters.ServerBuilder)`. The swap goes through `configureMessageConvertersList` rather than `withJsonConverter`: Spring Boot names its own JSON converter on the same builder, so whichever configurer ran last would win. A list configurer runs inside `build()` after all of them, sees the converter Boot settled on, and replaces it in place as before. StrictRequestBodyConfigTest now also drives the real builder with the JSON converter named after this config has run. The booted-application check that an unknown field is still a 400 naming it is in DatapointInsertHttpTest, added in the previous commit; with this hook disabled that request reaches the controller instead. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: olav --- .../config/StrictRequestBodyConfig.java | 15 +++++++++--- .../config/StrictRequestBodyConfigTest.java | 24 ++++++++++++++++--- 2 files changed, 33 insertions(+), 6 deletions(-) diff --git a/datahub-api/src/main/java/ai/intellistream/datahub/config/StrictRequestBodyConfig.java b/datahub-api/src/main/java/ai/intellistream/datahub/config/StrictRequestBodyConfig.java index e56cebd8..42bccda2 100644 --- a/datahub-api/src/main/java/ai/intellistream/datahub/config/StrictRequestBodyConfig.java +++ b/datahub-api/src/main/java/ai/intellistream/datahub/config/StrictRequestBodyConfig.java @@ -5,6 +5,7 @@ import org.springframework.context.annotation.Configuration; import org.springframework.http.converter.ByteArrayHttpMessageConverter; import org.springframework.http.converter.HttpMessageConverter; +import org.springframework.http.converter.HttpMessageConverters; import org.springframework.http.converter.StringHttpMessageConverter; import org.springframework.http.converter.json.JacksonJsonHttpMessageConverter; import org.jspecify.annotations.NonNull; @@ -51,11 +52,19 @@ public StrictRequestBodyConfig(JsonMapper configuredMapper) { } /** - * Extend rather than configure: the default converter list is already built, so replacing the - * Jackson one in place leaves ordering and every other converter untouched. + * Swaps the converter in the finished list rather than naming a JSON converter on the builder. + * + *

{@code withJsonConverter} would race Spring Boot, whose own configurer sets the JSON converter + * on the same builder, and whichever of the two runs last would win. A list configurer runs inside + * {@code build()}, after every configurer has had its turn, so it sees the converter Boot settled + * on and replaces it in place, leaving ordering and every other converter untouched. */ @Override - public void extendMessageConverters(@NonNull List> converters) { + public void configureMessageConverters(HttpMessageConverters.@NonNull ServerBuilder builder) { + builder.configureMessageConvertersList(this::replaceJacksonConverter); + } + + void replaceJacksonConverter(List> converters) { int replaced = 0; for (int i = 0; i < converters.size(); i++) { if (converters.get(i) instanceof JacksonJsonHttpMessageConverter) { diff --git a/datahub-api/src/test/java/ai/intellistream/datahub/config/StrictRequestBodyConfigTest.java b/datahub-api/src/test/java/ai/intellistream/datahub/config/StrictRequestBodyConfigTest.java index d74d000e..aa87196f 100644 --- a/datahub-api/src/test/java/ai/intellistream/datahub/config/StrictRequestBodyConfigTest.java +++ b/datahub-api/src/test/java/ai/intellistream/datahub/config/StrictRequestBodyConfigTest.java @@ -5,6 +5,7 @@ import org.junit.jupiter.api.Test; import org.springframework.http.converter.ByteArrayHttpMessageConverter; import org.springframework.http.converter.HttpMessageConverter; +import org.springframework.http.converter.HttpMessageConverters; import org.springframework.http.converter.StringHttpMessageConverter; import org.springframework.http.converter.json.JacksonJsonHttpMessageConverter; import tools.jackson.core.type.TypeReference; @@ -55,17 +56,34 @@ void theRequestBodyConverterIsTheStrictOne() { JsonMapper shared = JsonMapper.builder().build(); List> converters = defaultishConverters(); - new StrictRequestBodyConfig(shared).extendMessageConverters(converters); + new StrictRequestBodyConfig(shared).replaceJacksonConverter(converters); assertThat(converters).anyMatch(StrictJacksonJsonHttpMessageConverter.class::isInstance); } + /** + * Through the real builder, with the JSON converter named after this config has run, as Spring + * Boot's own configurer may: the swap must still land, since it happens when the list is built. + */ + @Test + void theBuiltServerConvertersCarryTheStrictOneWhoeverNamesTheJsonConverterLast() { + HttpMessageConverters.ServerBuilder builder = HttpMessageConverters.forServer().registerDefaults(); + + new StrictRequestBodyConfig(JsonMapper.builder().build()).configureMessageConverters(builder); + builder.withJsonConverter(new JacksonJsonHttpMessageConverter()); + + List> built = new ArrayList<>(); + builder.build().forEach(built::add); + assertThat(built).anyMatch(StrictJacksonJsonHttpMessageConverter.class::isInstance); + assertThat(built).noneMatch(c -> c.getClass() == JacksonJsonHttpMessageConverter.class); + } + /** Replacing the Jackson converter must not disturb the others or their order. */ @Test void leavesTheOtherConvertersAlone() { List> converters = defaultishConverters(); - new StrictRequestBodyConfig(JsonMapper.builder().build()).extendMessageConverters(converters); + new StrictRequestBodyConfig(JsonMapper.builder().build()).replaceJacksonConverter(converters); assertThat(converters).hasSize(3); assertThat(converters.get(0)).isInstanceOf(ByteArrayHttpMessageConverter.class); @@ -79,7 +97,7 @@ void toleratesAConverterListWithoutJackson() { List> converters = new ArrayList<>(List.of(new StringHttpMessageConverter())); assertThatCode(() -> new StrictRequestBodyConfig(JsonMapper.builder().build()) - .extendMessageConverters(converters)).doesNotThrowAnyException(); + .replaceJacksonConverter(converters)).doesNotThrowAnyException(); assertThat(converters).hasSize(1); } }