Data Format : Unable marshal LinkedHashMap to Protobuf

[weimeilin79]

Making a simple demo on format convert, but was getting error msg:
This works with java main, so I am assuming it's an Quarkus problem?

from("platform-http:/transfer?httpMethodRestrict=POST&consumes=application/json")
          .marshal().protobuf("demo.camel.TransactionProtos$Transaction")
          .toD("kafka:weborder?brokers=localhost:9092&groupId=producergroup&serializerClass=org.apache.kafka.common.serialization.ByteBufferSerializer&key=${header.sender}")

org.apache.camel.NoTypeConversionAvailableException: No type converter available to convert from type: byte[] to the required type: com.google.protobuf.Message with value [B@2eeab517
        at org.apache.camel.impl.converter.CoreTypeConverterRegistry.mandatoryConvertTo(CoreTypeConverterRegistry.java:220)
        at org.apache.camel.dataformat.protobuf.ProtobufDataFormat.convertGraphToMessage(ProtobufDataFormat.java:160)
        at org.apache.camel.dataformat.protobuf.ProtobufDataFormat.marshal(ProtobufDataFormat.java:129)
        at org.apache.camel.support.processor.MarshalProcessor.process(MarshalProcessor.java:64)
        at org.apache.camel.processor.errorhandler.RedeliveryErrorHandler$SimpleTask.run(RedeliveryErrorHandler.java:404)
        at org.apache.camel.impl.engine.DefaultReactiveExecutor$Worker.schedule(DefaultReactiveExecutor.java:148)
        at org.apache.camel.impl.engine.DefaultReactiveExecutor.scheduleMain(DefaultReactiveExecutor.java:60)
        at org.apache.camel.processor.Pipeline.process(Pipeline.java:147)
        at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:288)
        at org.apache.camel.quarkus.component.platform.http.runtime.QuarkusPlatformHttpConsumer.lambda$handleRequest$2(QuarkusPlatformHttpConsumer.java:397)
        at io.vertx.core.impl.ContextImpl.lambda$executeBlocking$2(ContextImpl.java:313)
        at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
        at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
        at java.base/java.lang.Thread.run(Thread.java:834)

Project repo:
Java Main:
https://github.com/weimeilin79/avroprotobuf/tree/master/camel-demo-restprotobuf

Quarkus:
https://github.com/weimeilin79/avroprotobuf/tree/master/camel-quarkus-restprotobuf

[jamesnetherton]

Shouldn't this be unmarshal?

[weimeilin79]

I want to turn it into TransactionProtos$Transaction, so should be marshal right?
The one on Java main as type LinkHashMap gets marshal into TransactionProtos$Transaction, and worked.
Not the one with Quarkus.

[jamesnetherton]

But your app is trying to convert a JSON string to a Map:

.convertBodyTo(String.class)
.marshal().json(JsonLibrary.Jackson,java.util.LinkedHashMap.class)

So it should be using unmarshal().json(...). Then the Map can be passed on to the protobuf marshal() step.

[weimeilin79]

oh, that was just a test, I was playing with it. And accidentally pushed to github.
This is one code I am talk about (I will push a correct version)

from("platform-http:/transfer?httpMethodRestrict=POST&consumes=application/json")
          .marshal().protobuf("demo.camel.TransactionProtos$Transaction")
          .toD("kafka:weborder?brokers=localhost:9092&groupId=producergroup&serializerClass=org.apache.kafka.common.serialization.ByteBufferSerializer&key=${header.sender}")

[jamesnetherton]

I think I see the issue. It's not really specific to camel-quarkus, it depends on which camel HTTP consumer you choose to use.

They each set different types as the message body for the incoming HTTP payload:

• netty-http = InputStream
• jetty = InputStream
• undertow = byte[]
• platform-http = byte[]

ProtobufDataFormat can deal with InputStream but not byte[]. So for your example to work, you'd need something like:

from("platform-http:/transfer?httpMethodRestrict=POST&consumes=application/json").streamCaching()
    .convertBodyTo(InputStream.class)
    .setHeader("receiverid", jsonpath("$.receiverid"))
    .setHeader("sender", jsonpath("$.sender.userid"))
    .marshal().protobuf("demo.camel.TransactionProtos$Transaction")
    .toD("kafka:webtrans?brokers=localhost:9092&groupId=producergroup&serializerClass=org.apache.kafka.common.serialization.ByteBufferSerializer&key=${header.sender}");

[weimeilin79]

Ah... Got it. Thanks! @jamesnetherton

[weimeilin79]

For people that might bump into the same problem:

So directly casting the byte[] to InputStream doesn't cut it. Although everything looks ok, no error during runtime, I end up with an empty object. Meaning the byte was not correctly marshal into the object.
I then change the endpoint to netty-http it worked like a charm. Clearly not a Quarkus issue.
Will checkin with Camel and ask the difference between the two.
Thanks!

Working example

restConfiguration().component("netty-http").host("localhost").port("8080").bindingMode(RestBindingMode.auto);
        
        rest("/").post("transfer").consumes("application/json").to("direct:transfer");

from("direct:transfer")   
        .setHeader("sender",jsonpath("$.sender.userid"))
        .marshal().protobuf("demo.camel.TransactionProtos$Transaction")
        .log("Sender: ${header.sender}")
        .toD("kafka:webtrans-quarkus?brokers=localhost:9092&key=${header.sender}&serializerClass=org.apache.kafka.common.serialization.ByteArraySerializer")
        ;