Issue with large json objects returned as RestMediaType.APPLICATION_NDJSON

[magnus-gustafsson]

Describe the bug

I'm trying to use REST APIs that return a Multi<Message> that produces RestMediaType.APPLICATION_NDJSON.

@GET
@Path("/pojo")
@Produces(RestMediaType.APPLICATION_NDJSON)
@RestStreamElementType(MediaType.APPLICATION_JSON)
public Multi<Message> readPojo() {
   //
}

Quarkus works as intended if the Message is smaller than approximately 1000 bytes. However, if it is larger, org.jboss.resteasy.reactive.client.impl.MultiInvoker feeds incomplete chunks into Jackson for deserialization, which of course fails.

MultiInvoker does not actually check for a newline delimiter in the provided chunk. If there is no newline, it defaults to using the size of the chunk.

Expected behavior

Quarkus should buffer read data until the delimiter is read, then parse the buffered content as a json object.

Actual behavior

Jackson fails to deserialize because the json content is not complete.
Caused by: jakarta.ws.rs.ProcessingException: com.fasterxml.jackson.databind.JsonMappingException: Unexpected end-of-input in VALUE_STRING

How to Reproduce?

Test based on MultiNdjsonTest :

public class LargeNDJsonTest {
    @RegisterExtension
    static final QuarkusUnitTest TEST = new QuarkusUnitTest();

    @TestHTTPResource
    URI uri;
   
    @Test
    void readLargePojo() throws InterruptedException {
        var client = createClient(uri);
        var collected = new CopyOnWriteArrayList<Message>();
        var completionLatch = new CountDownLatch(1);
        client.readPojo(1000).onCompletion().invoke(completionLatch::countDown)
                .subscribe().with(collected::add);

        if (!completionLatch.await(5, TimeUnit.SECONDS)) {
            fail("Streaming did not complete in time");
        }
        assertThat(collected.size(), org.hamcrest.Matchers.is(4));

    }


    private Client createClient(URI uri) {
        return QuarkusRestClientBuilder.newBuilder().baseUri(uri).build(Client.class);
    }


    @Path("/stream")
    public interface Client {
        @GET
        @Path("/pojo/{size}")
        @Produces(RestMediaType.APPLICATION_NDJSON)
        @RestStreamElementType(MediaType.APPLICATION_JSON)
        Multi<Message> readPojo(int size);

    }


    @Path("/stream")
    public static class StreamingResource {
        @Inject
        Vertx vertx;


        @GET
        @Path("/pojo/{size}")
        @Produces(RestMediaType.APPLICATION_NDJSON)
        @RestStreamElementType(MediaType.APPLICATION_JSON)
        public Multi<Message> readPojo(int size) {
            return Multi.createFrom().emitter(
                    em -> {
                        Random random = new Random();
                        byte[] bytes = new byte[size];
                        random.nextBytes(bytes);

                        em.emit(Message.of("one", java.util.Base64.getEncoder().encodeToString(bytes)));
                        em.emit(Message.of("two", java.util.Base64.getEncoder().encodeToString(bytes)));
                        em.emit(Message.of("three", java.util.Base64.getEncoder().encodeToString(bytes)));

                        vertx.setTimer(100, id -> {
                            em.emit(Message.of("four", java.util.Base64.getEncoder().encodeToString(bytes)));
                            em.complete();
                        });
                    });
        }

    }

    public record Message(String name, String value) {
        public static Message of(String name, String value) {
            return new Message(name, value);
        }
    }

}

If client.readPojo(1000) is changed to client.readPojo(25) then the test passes.

Output of uname -a or ver

Linux magnusg-laptop 6.11.0-21-generic #21-Ubuntu SMP PREEMPT_DYNAMIC Wed Feb 19 16:50:40 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux

Output of java -version

openjdk 23.0.2 2025-01-21

Quarkus version or git rev

3.21.1

Build tool (ie. output of mvnw --version or gradlew --version)

Gradle 8.12.1

Additional information

No response

[quarkus-bot]

/cc @FroMage (rest), @geoand (rest)

[geoand]

I'll check this out soon

[geoand]

Can you please attach the exact test file you used as a tried the one you mention in the description and could not get the test to work at all.

[magnus-gustafsson]

Sure,
Here is an modified version of io.quarkus.rest.client.reactive.MultiNdjsonTest:

package io.quarkus.rest.client.reactive;

import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.fail;

import java.net.URI;
import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.Random;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import jakarta.inject.Inject;
import jakarta.ws.rs.GET;
import jakarta.ws.rs.Path;
import jakarta.ws.rs.Produces;
import jakarta.ws.rs.core.MediaType;

import org.jboss.resteasy.reactive.RestStreamElementType;
import org.jboss.resteasy.reactive.common.util.RestMediaType;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectWriter;

import io.quarkus.test.QuarkusUnitTest;
import io.quarkus.test.common.http.TestHTTPResource;
import io.quarkus.vertx.web.ReactiveRoutes;
import io.quarkus.vertx.web.Route;
import io.smallrye.mutiny.Multi;
import io.vertx.core.Vertx;
import io.vertx.ext.web.RoutingContext;
import wiremock.org.apache.hc.client5.http.utils.Base64;

public class MultiNdjsonTest {
    @RegisterExtension
    static final QuarkusUnitTest TEST = new QuarkusUnitTest()
            .withApplicationRoot((jar) -> jar.addClasses(TestJacksonBasicMessageBodyReader.class));

    @TestHTTPResource
    URI uri;

    @Test
    void shouldReadNdjsonStringAsMulti() throws InterruptedException {
        var client = createClient(uri);
        var collected = new CopyOnWriteArrayList<String>();
        var completionLatch = new CountDownLatch(1);
        client.readString().onCompletion().invoke(completionLatch::countDown)
                .subscribe().with(collected::add);

        if (!completionLatch.await(5, TimeUnit.SECONDS)) {
            fail("Streaming did not complete in time");
        }
        assertThat(collected).hasSize(4)
                .contains("\"one\"", "\"two\"", "\"three\"", "\"four\"");
    }

    @Test
    void shouldReadNdjsonPojoAsMulti() throws InterruptedException {
        var client = createClient(uri);
        var collected = new CopyOnWriteArrayList<Message>();
        var completionLatch = new CountDownLatch(1);
        client.readPojo().onCompletion().invoke(completionLatch::countDown)
                .subscribe().with(collected::add);

        if (!completionLatch.await(5, TimeUnit.SECONDS)) {
            fail("Streaming did not complete in time");
        }
        var expected = Arrays.asList(Message.of("one", "1"),
                Message.of("two", "2"), Message.of("three", "3"),
                Message.of("four", "4"));
        assertThat(collected).hasSize(4).containsAll(expected);
    }

    @Test
    void shouldReadNdjsonPojoFromReactiveRoutes() throws InterruptedException {
        URI reactiveRoutesBaseUri = URI.create(uri.toString() + "/rr");
        var client = createClient(reactiveRoutesBaseUri);
        var collected = new CopyOnWriteArrayList<Message>();
        var completionLatch = new CountDownLatch(1);
        client.readPojo().onCompletion().invoke(completionLatch::countDown)
                .subscribe().with(collected::add);

        if (!completionLatch.await(5, TimeUnit.SECONDS)) {
            fail("Streaming did not complete in time");
        }
        var expected = Arrays.asList(Message.of("superman", "1"),
                Message.of("batman", "2"), Message.of("spiderman", "3"));
        assertThat(collected).hasSize(3).containsAll(expected);
    }

    @Test
    void shouldReadNdjsonFromSingleMessage() throws InterruptedException {
        var client = createClient(uri);
        var collected = new CopyOnWriteArrayList<Message>();
        var completionLatch = new CountDownLatch(1);
        client.readPojoSingle().onCompletion().invoke(completionLatch::countDown)
                .subscribe().with(collected::add);

        if (!completionLatch.await(5, TimeUnit.SECONDS)) {
            fail("Streaming did not complete in time");
        }
        var expected = Arrays.asList(
                Message.of("zero", "0"), Message.of("one", "1"),
                Message.of("two", "2"), Message.of("three", "3"));
        assertThat(collected).hasSize(4).containsAll(expected);
    }

    @Test
    void shouldReadLargeNdjsonPojoAsMulti() throws InterruptedException {
        var client = createClient(uri);
        var collected = new CopyOnWriteArrayList<Message>();
        var completionLatch = new CountDownLatch(1);
        client.readLargePojo().onCompletion().invoke(completionLatch::countDown)
                .subscribe().with(collected::add);

        if (!completionLatch.await(5, TimeUnit.SECONDS)) {
            fail("Streaming did not complete in time");
        }

        assertThat(collected).hasSize(4);
    }

    private Client createClient(URI uri) {
        return QuarkusRestClientBuilder.newBuilder().baseUri(uri).register(new TestJacksonBasicMessageBodyReader())
                .build(Client.class);
    }

    @Path("/stream")
    public interface Client {

        @GET
        @Path("/string")
        @Produces(RestMediaType.APPLICATION_NDJSON)
        @RestStreamElementType(MediaType.APPLICATION_JSON)
        Multi<String> readString();

        @GET
        @Path("/pojo")
        @Produces(RestMediaType.APPLICATION_NDJSON)
        @RestStreamElementType(MediaType.APPLICATION_JSON)
        Multi<Message> readPojo();

        @GET
        @Path("/single-pojo")
        @Produces(RestMediaType.APPLICATION_NDJSON)
        @RestStreamElementType(MediaType.APPLICATION_JSON)
        Multi<Message> readPojoSingle();

        @GET
        @Path("/large-pojo")
        @Produces(RestMediaType.APPLICATION_NDJSON)
        @RestStreamElementType(MediaType.APPLICATION_JSON)
        Multi<Message> readLargePojo();

    }

    public static class ReactiveRoutesResource {

        @Route(path = "/rr/stream/pojo", produces = ReactiveRoutes.ND_JSON)
        Multi<Message> people(RoutingContext context) {
            return Multi.createFrom().items(
                    Message.of("superman", "1"),
                    Message.of("batman", "2"),
                    Message.of("spiderman", "3"));
        }
    }

    @Path("/stream")
    public static class StreamingResource {
        @Inject
        Vertx vertx;

        @GET
        @Path("/string")
        @Produces(RestMediaType.APPLICATION_NDJSON)
        @RestStreamElementType(MediaType.APPLICATION_JSON)
        public Multi<String> readString() {
            return Multi.createFrom().emitter(
                    em -> {
                        em.emit("one");
                        em.emit("two");
                        em.emit("three");
                        em.emit("four");
                        em.complete();
                    });
        }

        @GET
        @Path("/pojo")
        @Produces(RestMediaType.APPLICATION_NDJSON)
        @RestStreamElementType(MediaType.APPLICATION_JSON)
        public Multi<Message> readPojo() {
            return Multi.createFrom().emitter(
                    em -> {
                        em.emit(Message.of("one", "1"));
                        em.emit(Message.of("two", "2"));
                        em.emit(Message.of("three", "3"));
                        vertx.setTimer(100, id -> {
                            em.emit(Message.of("four", "4"));
                            em.complete();
                        });
                    });
        }

        @GET
        @Path("/single-pojo")
        @Produces(RestMediaType.APPLICATION_NDJSON)
        @RestStreamElementType(MediaType.APPLICATION_JSON)
        public String getPojosAsString() throws JsonProcessingException {
            ObjectMapper mapper = new ObjectMapper();
            StringBuilder result = new StringBuilder();
            ObjectWriter objectWriter = mapper.writerFor(Message.class);
            for (var msg : List.of(Message.of("zero", "0"),
                    Message.of("one", "1"),
                    Message.of("two", "2"),
                    Message.of("three", "3"))) {
                result.append(objectWriter.writeValueAsString(msg));
                result.append("\n");
            }
            return result.toString();
        }

        @GET
        @Path("/large-pojo")
        @Produces(RestMediaType.APPLICATION_NDJSON)
        @RestStreamElementType(MediaType.APPLICATION_JSON)
        public Multi<Message> readLargePojo() {
            return Multi.createFrom().emitter(
                    em -> {
                        byte[] bytes = new byte[4*1024];
                        Random random = new Random();
                        random.nextBytes(bytes);
                        String value = Base64.encodeBase64String(bytes);
                        em.emit(Message.of("one", value));
                        em.emit(Message.of("two", value));
                        em.emit(Message.of("three", value));
                        vertx.setTimer(100, id -> {
                            em.emit(Message.of("four", value));
                            em.complete();
                        });
                    });
        }
    }

    public static class Message {
        public String name;
        public String value;

        public static Message of(String name, String value) {
            Message message = new Message();
            message.name = name;
            message.value = value;
            return message;
        }

        @Override
        public boolean equals(Object o) {
            if (this == o)
                return true;
            if (o == null || getClass() != o.getClass())
                return false;
            Message message = (Message) o;
            return Objects.equals(name, message.name) && Objects.equals(value, message.value);
        }

        @Override
        public int hashCode() {
            return Objects.hash(name, value);
        }

        @Override
        public String toString() {
            return "Message{" +
                    "name='" + name + '\'' +
                    ", value='" + value + '\'' +
                    '}';
        }
    }
}

The test shouldReadLargeNdjsonPojoAsMulti() fails with the timeout because jackson throws
Caused by: jakarta.ws.rs.ProcessingException: com.fasterxml.jackson.core.JsonParseException: Unrecognized token 'eVl2B05HisA8BnaFNjwxJTqqdY': was expecting (JSON String, Number, Array, Object or token 'null', 'true' or 'false')

[geoand]

Thanks.

I tried this and it looks like a Vert.x issue to me.
Essentially what happens is that even though the server sends the chunks properly (I have verified this using Wireshark), the handler method of Vert.x's HttpClientResponse is getting called with buffers that don't correspond to the whole chunk (even though I have made sure the max chunk size is larger than what the server sends).
@cescoffier if you have any ideas, I'll gladly hear them, otherwise I think the next step is to create a reproducer using Vert.x only and opening an issue.

[magnus-gustafsson]

Should it not be the MultiInvoker responsibility to buffer the read vert.x chunks until a delimiter is part of the chunk, and only after that, try to parse all buffered chunks?
Similar to how it is done in registerForJsonStream?

[geoand]

I guess yeah, we would probably need a specific method for nd-json instead of hiding that inside registerForChunks. Would you be insterested in contributing that?

[cescoffier]

I agree, the multi invoker should be able to handle that and accumulate the chunk before calling the parser.

[magnus-gustafsson]

It seems that the easiest fix would be to use the same parser (registerForJsonStream) for RestMediaType.APPLICATION_NDJSON_TYPE as is used for RestMediaType.APPLICATION_STREAM_JSON_TYPE. That solves this issue.

However, I believe there is another (semi-related) bug here. MultiInvoker parses both RestMediaType.APPLICATION_STREAM_JSON_TYPE and RestMediaType.APPLICATION_NDJSON_TYPE as newline (\n)-delimited JSON, but that should only be true for RestMediaType.APPLICATION_NDJSON_TYPE.

My understanding is that
APPLICATION_STREAM_JSON_TYPE (application/stream+json) looks like:

{"id":1,"name":"Alice"}{"id":2,"name":"Bob"}

and

APPLICATION_NDJSON_TYPE (application/x-ndjson or application/ndjson) looks like:

{"id":1,"name":"Alice"}\n
{"id":2,"name":"Bob"}\n

Please correct me if I'm wrong here.

I'm not sure why MultiInvoker parses RestMediaType.APPLICATION_NDJSON_TYPE as \n-delimited in one way and RestMediaType.APPLICATION_STREAM_JSON_TYPE as \n-delimited as well, but in a different (buffering) manner.

[geoand]

@magnus-gustafsson I am assuming it's just an oversight. At it seems like you have already done all the hard work, it would be great to have a PR with your fixes, ideally with tests you have used to verify that things work as expected.