Pulsar: JSONSchema should works with generic (parameterized type) POJO/DTO.

[ducquoc]

Expected behavior

JSONSchema should works with generic (parameterized type) POJO/DTO. No exception at runtime, or at least update documentation/errorMessage how to work-around/bypass the issue.

Actual behavior

Exception when running producer/consumer with JSONSchema.of(clazz) (or Schema.JSON(clazz) ) wiith generic/parameterized class:

Exception in thread "main" java.lang.NoSuchFieldError: KEY_VALUE
    at org.apache.pulsar.client.impl.schema.KeyValueSchema.<init>(KeyValueSchema.java:50)
    at org.apache.pulsar.client.api.Schema.<clinit>(Schema.java:128)

Steps to reproduce

Create Producer or Consumer with client API, using JSONSchema.of(clazz) (or Schema.JSON(clazz) ) wiith generic/parameterized class.
Compiled OK, but when running the client producer/consumer it throws exception.

Example (https://github.com/ducquoc/pulsar/commit/a0649dc8):
https://github.com/ducquoc/pulsar/blob/fix-client-schema_json-schema-generic-parameterized-type/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/pojo/BrokerMessage.java

** Work-around: use

private Object payload;

and respective getter/setter instead of

private T payload;

System configuration

Pulsar version: 2.2.0 (but seems to be also affected on latest master branch)

[jerrypeng]

So we use Avro underneath to generate a schema for the class. Parameterized/generics in Java are currently not support for schema generation in Avro. Though there is discussion and a PR to add that:

https://issues.apache.org/jira/browse/AVRO-2248

https://github.com/apache/avro/pull/356

Though not sure if that is ever going to get merged in.

[ducquoc]

Thank you for your quick response.

I hope the Avro fix will be merged soon (and not much performance overhead). Until then I will apply the work-around java.lang.Object - sufficient so far :)

[vamsi360]

Hi, I worked on AVRO-2248 and we stabilised it and use it in production in our company. We will resume the discussion and make the required changes to get it merged to master.

For now, if generic types are required, you can use the fork https://github.com/vamsi360/avro

[rohts-patil]

@vamsi360 ,

I cloned your fork and built it to version 1.8.3-ppe-9.10.
I still get the errror.

Caused by: org.apache.avro.AvroTypeException: Unknown type: T
    at org.apache.avro.specific.SpecificData.createSchema(SpecificData.java:312)
    at org.apache.avro.reflect.ReflectData.createSchema(ReflectData.java:720)
    at org.apache.avro.reflect.ReflectData.createFieldSchema(ReflectData.java:870)
    at org.apache.avro.reflect.ReflectData.createSchema(ReflectData.java:686)
    at org.apache.avro.specific.SpecificData$2.load(SpecificData.java:238)
    at org.apache.avro.specific.SpecificData$2.load(SpecificData.java:235)
    at avro.shaded.com.google.common.cache.LocalCache$LoadingValueReference.loadFuture(LocalCache.java:3568)
    at avro.shaded.com.google.common.cache.LocalCache$Segment.loadSync(LocalCache.java:2350)
    at avro.shaded.com.google.common.cache.LocalCache$Segment.lockedGetOrLoad(LocalCache.java:2313)
    at avro.shaded.com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2228)
    ... 6 more

[shan-96]

@vamsi360 ,
I cloned your fork and built it to version 1.8.3-ppe-9.10.
I still get the errror.
Caused by: org.apache.avro.AvroTypeException: Unknown type: T
at org.apache.avro.specific.SpecificData.createSchema(SpecificData.java:312)
at org.apache.avro.reflect.ReflectData.createSchema(ReflectData.java:720)
at org.apache.avro.reflect.ReflectData.createFieldSchema(ReflectData.java:870)
at org.apache.avro.reflect.ReflectData.createSchema(ReflectData.java:686)
at org.apache.avro.specific.SpecificData$2.load(SpecificData.java:238)
at org.apache.avro.specific.SpecificData$2.load(SpecificData.java:235)
at avro.shaded.com.google.common.cache.LocalCache$LoadingValueReference.loadFuture(LocalCache.java:3568)
at avro.shaded.com.google.common.cache.LocalCache$Segment.loadSync(LocalCache.java:2350)
at avro.shaded.com.google.common.cache.LocalCache$Segment.lockedGetOrLoad(LocalCache.java:2313)
at avro.shaded.com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2228)
... 6 more

I am facing the same problem. It seems we are creating the build in a wrong way

[vamsi360]

Hi @rohts-patil @shan-96
Could you tell me how to replicate the issue you are facing with a test project? I will debug this and replicate/fix

[shan-96]

@vamsi360

We are trying to write to Avro Files using version 1.8.3-ppe-9.10

We are writing a java object from memory to a specified file using DataFileWriter.
While building the schema of the POJO at runtime, we call...

private static Schema getJsonSchema(Class c) {
    Schema s = ReflectData.AllowNull.get().getSchema(c);
    AvroSchema avroSchema = new AvroSchema(s);
    return avroSchema.getAvroSchema();
  }

However, we still get the same exception as mentioned above which we got through maven released version of Avro (1.8)

Our schema (POJO) looks like this:

public RawData extends Data<String>{
    private static final long serialVersionUID = 1L;
    private long field 1;
    private String field 2;
}

public abstract class Data<T> implements Serializable {
    private static final long serialVersionUID = 1L;
    private String dataType;
    private T id;
}

We want to define AvroSchemafor RawData.class

cc: @rohts-patil

[frankjkelly]

We too have been hit by this issue trying to send generics

public class InteractionEvent<T>

[jiazhai]

@gaoran10 @congbobo184 Which of you have time to take a look at this issue?

[sijie]

@frankjkelly Which pulsar client are you using?

[frankjkelly]

@sijie I am using 2.6.1 Client with 2.6.1 Server

When I use the code as follows

    private Producer<SignalEvent> getSignalProducer(String tenant) throws PulsarClientException {
        if (signalProducer.get(tenant) == null) {
            signalProducer.put(tenant, client.newProducer(Schema.JSON(SignalEvent.class))
                    .topic(createTopicUri(tenant, Topic.SIGNAL))
                    .create());
        }
        return signalProducer.get(tenant);
    }

I get the error

Unknown type: T
org.apache.avro.AvroTypeException: Unknown type: T
    at org.apache.avro.specific.SpecificData.createSchema(SpecificData.java:410)
    at org.apache.avro.reflect.ReflectData.createSchema(ReflectData.java:663)
    at org.apache.avro.reflect.ReflectData.createFieldSchema(ReflectData.java:759)
    at org.apache.avro.reflect.ReflectData$AllowNull.createFieldSchema(ReflectData.java:86)
    at org.apache.avro.reflect.ReflectData.createSchema(ReflectData.java:615)
    at org.apache.avro.specific.SpecificData$3.computeValue(SpecificData.java:335)
    at org.apache.avro.specific.SpecificData$3.computeValue(SpecificData.java:332)
    at java.base/java.lang.ClassValue.getFromHashMap(ClassValue.java:226)
    at java.base/java.lang.ClassValue.getFromBackup(ClassValue.java:208)
    at java.base/java.lang.ClassValue.get(ClassValue.java:114)
    at org.apache.avro.specific.SpecificData.getSchema(SpecificData.java:346)
    at org.apache.pulsar.client.impl.schema.StructSchema.extractAvroSchema(StructSchema.java:174)
    at org.apache.pulsar.client.impl.schema.StructSchema.createAvroSchema(StructSchema.java:161)
    at org.apache.pulsar.client.impl.schema.StructSchema.parseSchemaInfo(StructSchema.java:187)
    at org.apache.pulsar.client.impl.schema.JSONSchema.of(JSONSchema.java:96)
    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.pulsar.client.internal.DefaultImplementation.lambda$newJSONSchema$26(DefaultImplementation.java:242)
    at org.apache.pulsar.client.internal.ReflectionUtils.catchExceptions(ReflectionUtils.java:35)
    at org.apache.pulsar.client.internal.DefaultImplementation.newJSONSchema(DefaultImplementation.java:239)
    at org.apache.pulsar.client.api.Schema.JSON(Schema.java:260)
    at com.cogito.library.eventmanager.PulsarEventBroker.getSignalProducer(PulsarEventBroker.java:106)
    at com.cogito.library.eventmanager.PulsarEventBroker.sendEvent(PulsarEventBroker.java:184)

but this code works fine

    private Producer<String> getSignalProducer(String tenant) throws PulsarClientException {
        if (signalProducer.get(tenant) == null) {
            signalProducer.put(tenant, client.newProducer(Schema.STRING)
                    .topic(createTopicUri(tenant, Topic.SIGNAL))
                    .create());
        }
        return signalProducer.get(tenant);
    }

Thanks!

[sijie]

@frankjkelly I see. So it is related to extracting the schema information using the AVRO library. We will take a look and circle back.

[congbobo184]

@frankjkelly I think, because you use generic type T, so use can't provide the specific type. we can't generate avro schema according to the json schema information.

[frankjkelly]

Yes - as per @jerrypeng comment above
"Parameterized/generics in Java are currently not support for schema generation in Avro. Though there is discussion and a PR to add that:

https://issues.apache.org/jira/browse/AVRO-2248

apache/avro#356

Though not sure if that is ever going to get merged in."

So it appears it's up to the folks on the Avro side?

[sijie]

@frankjkelly Yeah, currently it is blocked by apache/avro#356. I would encourage people to leave a comment in the avro issue and pull request to see if AVRO community can accelerate the process on adopting that change.

[frankjkelly]

Thank you - will do!