Deserialize JSON with Camel Routes-CodePudding

I'm trying to unmarshal json data generated by debezium inside a kafka topic.

My approach is simple, use POJOs and Jackson Library, however, since this json has a root object (initialized inside "{}") it throws an error.

This is the json received, I'm just interested on the payload:

{
    "schema": {
        "type": "struct",
        "fields": [{
            "type": "double",
            "optional": false,
            "field": "codid"
        }, {
            "type": "string",
            "optional": true,
            "field": "__op"
        }, {
            "type": "string",
            "optional": true,
            "field": "__deleted"
        }],
        "optional": false,
        "name": "demo.RESCUE.Value"
    },
    "payload": {
        "codid": 0.0,
        "__op": "r",
        "__deleted": "false"
    }
}

And this is my Route:

public class Routes extends RouteBuilder{

    public static class MySplitter {
        public List<Payload> splitBody(Rescue data) {
            return data.getPayload().stream().collect(toList());
        }
    }

    @Override
    public void configure() throws Exception {

        from("kafka:{{kafka.source.topic.name}}?brokers={{kafka.bootstrap.address}}&autoOffsetReset=earliest")
        .log("Received body: ${body}")
            .unmarshal().json(JsonLibrary.Jackson, Rescue.class)
            .split().method(MySplitter.class, "splitBody")
            .marshal().json(JsonLibrary.Jackson)
            .convertBodyTo(String.class)
            .log("Output: ${body}");
    }
    
}

And the error received:

com.fasterxml.jackson.databind.exc.MismatchedInputException: Cannot deserialize value of type `java.util.ArrayList<org.demo.pojos.rescue.Payload>` from Object value (token `JsonToken.START_OBJECT`)

CodePudding user response：

If you are just interested in payload, you have to extract this object from the whole JSON. For example with JSONPath.

Camel supports JSONPath as expression language. Therefore you can try something like

.log("Received body: ${body}")  // logs the full JSON
.setBody().jsonpathWriteAsString("$.payload")
.log("Reduced body: ${body}")   // should log the new body (only the payload)
...

Notice that you need to add the camel-jsonpath dependency

<dependency>
    <groupId>org.apache.camel</groupId>
    <artifactId>camel-jsonpath</artifactId>
</dependency>

or if you use SpringBoot

<dependency>
    <groupId>org.apache.camel.springboot</groupId>
    <artifactId>camel-jsonpath-starter</artifactId>
</dependency>