Hi everyone, I'm trying to use the

ConfluentAvroRegistryDeserializationSchema.forSpecific

. I am using the POJO generated from the

avro-maven-plugin

but it seems Flink does not understand the POJO.

org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: Expecting type to be a PojoTypeInfo
	at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:372)
	at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222)
	at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:98)
	at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:846)
	at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:240)
	at org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1090)
	at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1168)
	at org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:28)
	at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1168)
Caused by: java.lang.IllegalStateException: Expecting type to be a PojoTypeInfo
	at org.apache.flink.formats.avro.typeutils.AvroTypeInfo.generateFieldsFromAvroSchema(AvroTypeInfo.java:72)
	at org.apache.flink.formats.avro.typeutils.AvroTypeInfo.<init>(AvroTypeInfo.java:55)
	at org.apache.flink.formats.avro.AvroDeserializationSchema.getProducedType(AvroDeserializationSchema.java:177)
	at org.apache.flink.connector.kafka.source.reader.deserializer.KafkaValueOnlyDeserializationSchemaWrapper.getProducedType(KafkaValueOnlyDeserializationSchemaWrapper.java:56)
	at org.apache.flink.connector.kafka.source.KafkaSource.getProducedType(KafkaSource.java:216)
	at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.getTypeInfo(StreamExecutionEnvironment.java:2634)
	at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.fromSource(StreamExecutionEnvironment.java:2006)
	at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.fromSource(StreamExecutionEnvironment.java:1977)
	at org.flink.test.DataStreamJob.main(DataStreamJob.java:73)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
	at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355)

Does anyone have any ideas on how to fix this?

Can you confirm your class actually is a POJO? E.g. has an empty public constructor, a setter for every field, etc.

Yes it is. I have checked the file, it is also autogenerated by the maven plugin.

We've seen before that the Maven plugin actually doesn't generate valid POJOs. I would recommend creating a test to validate the POJO https://github.com/immerok/recipes/blob/main/kafka-json-to-pojo/src/test/java/com/immerok/cookbook/KafkaJSONToPOJOTest.java#L69-L73

I think the POJOs generated from Avro types are supposed to be different. I found this in the documentation. "POJOs are generally represented with a

PojoTypeInfo

and serialized with the

PojoSerializer

(using Kryo as configurable fallback). The exception is when the POJOs are actually Avro types (Avro Specific Records) or produced as “Avro Reflect Types”. In that case the POJO’s are represented by an

AvroTypeInfo

and serialized with the

AvroSerializer

. You can also register your own custom serializer if required; see Serialization for further information."

Hey did you use an IDL or AVSC to generate the pojos?

I added the

avro-maven-plugin

to my

pom.xml

and built the project. The schema files were avsc files. What did you mean by IDL?

KafkaSource.<Redacted>builder()
        .setBootstrapServers(configuration.getBootstrap())
        .setTopics(configuration.RedactedTopic())
        .setGroupId(configuration.getGroupId())
        .setStartingOffsets(OffsetsInitializer.earliest())
        .setValueOnlyDeserializer(forSpecific(Redacted.class, configuration.getSchemaRegistryUrl()))
        .build();

This is how I build my source and I made an avro protocol to generate my pojos which looks something like protocol.advl

@namespace("my.cool.namespace")
protocol MyProtocol{
   record Event {
      string @javaAnnotation("jakarta.validation.constraints.NotNull") id;

      @java-class("java.lang.Long")
      long @javaAnnotation("jakarta.validation.constraints.NotNull") eventTimestamp;

      @java-class("java.lang.Integer")
      int @javaAnnotation([
            "jakarta.validation.constraints.Min(0)",
            "jakarta.validation.constraints.Max(999)",
            "jakarta.validation.constraints.NotNull"])
            redacted;
   }
}

Interesting, so you didn't need

ConfluentAvroRegistryDeserializationSchema

? My setup looks like this

KafkaSource<MyPOJO> history_ledger_source = KafkaSource.<MyPOJO>builder()
				.setBootstrapServers("xxxx")
				.setTopics("topic_name")
				.setStartingOffsets(OffsetsInitializer.earliest())
				.setGroupId("test-001")
				.setProperty("security.protocol", "SASL_SSL")
				.setProperty("sasl.mechanism", "PLAIN")
				.setProperty("client.dns.lookup","use_all_dns_ips")
				.setValueOnlyDeserializer(ConfluentRegistryAvroDeserializationSchema.forSpecific(MyPOJO.class ,registryUrl, registryConfig))
				.build();

I am using a static import, which version of java are you running? Also could I please see an example of your generated avro pojo?

Java 11

Can you create a random schema to try on?

yes I can do that. Might take some minutes

This seems correct, do you still get the same issue when using it with the source?

@Kosta Sovaridis which class are you using for this method

forSpecific()

?

.setValueOnlyDeserializer(forSpecific(Redacted.class, configuration.getSchemaRegistryUrl()))

I am using org.apache.flink.formats.avro.registry.confluent.ConfluentRegistryAvroDeserializationSchema

@Martijn Visser I did the test you recommended and the POJOs failed. However, they failed because there are timestamp fields which are logical types. I did not have this issue with POJOs created by the

avro-maven-plugin

that did not have timestamps. And the addition of a timestamp field made them fail the test as well. I saw that there is an open issue you commented on where

ConfluentAvroRegistryDeserializationSchema

does not process logical types. Do you know about any solutions to this specific logical types issue?

I'm afraid I don't