Re: pojo warning when using auto generated protobuf class

Posted by Prashant Deva on
URL: http://deprecated-apache-flink-user-mailing-list-archive.369.s1.nabble.com/pojo-warning-when-using-auto-generated-protobuf-class-tp43189p43194.html

so i did  register the type with Kryo and the ProtobufSerializer. However I am still continuing to see the warnings. is this a bug in Flink?

env.config.registerTypeWithKryoSerializer(Trace.APITrace::class.java, ProtobufSerializer::class.java)

 val stream: DataStreamSource<Trace.APITrace> = env.addSource(FlinkKafkaConsumer(Config.TOPIC_SPANS, ApiTraceSchema(), props))

On Sat, Apr 24, 2021 at 8:48 AM, Yun Gao <[hidden email]> wrote:

Hi Prashant,

I think the warn is given when calling 

return TypeInformation.of(Trace.APITrace::class.java)

Currently flink does not have the native support 

for the protobuf types yet[1], thus it would use a

generic serializer created by kryo. 

This should not affect the rightness of the program

and should only affect its performance. One possible

solution might be register custom serializer into the kryo 

serializer framework for protobuf classes, like the example in [2].

Best,

Yun

[1] https://issues.apache.org/jira/browse/FLINK-11333

[2] https://ci.apache.org/projects/flink/flink-docs-stable/dev/custom_serializers.html

------------------Original Mail ------------------

Sender:Prashant Deva <[hidden email]>

Send Date:Sat Apr 24 11:00:17 2021

Recipients:User <[hidden email]>

Subject:pojo warning when using auto generated protobuf class

I am seeing this warning msg when trying to use a custom protobuf de/serializer with kafka source with auto generated java protobuf class:

18:41:31.164 [main] INFO org.apache.flink.api.java.typeutils.TypeExtractor - Class class com.xx.APITrace cannot be used as a POJO type because not all fields are valid POJO fields, and must be processed as GenericType. Please read the Flink documentation on "Data Types & Serialization" for details of the effect on performance.

here is my serializer. What am i doing wrong?

class ApiTraceSchema: DeserializationSchema<Trace.APITrace>, SerializationSchema<Trace.APITrace> {

    override fun getProducedType(): TypeInformation<Trace.APITrace> {
        return TypeInformation.of(Trace.APITrace::class.java)
    }

    override fun deserialize(message: ByteArray): Trace.APITrace {
        return Trace.APITrace.parseFrom(message)
    }

    override fun isEndOfStream(nextElement: Trace.APITrace): Boolean {
        return false
    }

    override fun serialize(element: Trace.APITrace): ByteArray {
        return element.toByteArray()
    }
}