The pulsar client generics have an exception

[livk-cloud]

The code is shown below

public class Main {
    public static void main(String[] args) throws PulsarClientException {
        try (PulsarClient pulsarClient = PulsarClient.builder()
                .serviceUrl("pulsar://livk.com:6650")
                .build()) {
            SchemaDefinition<PulsarMessage<String>> schemaDefinition = SchemaDefinition.<PulsarMessage<String>>builder()
                    .withPojo(PulsarMessage.class)
                    .build();
            try (Producer<PulsarMessage<String>> producer = pulsarClient.<PulsarMessage<String>>newProducer(Schema.JSON(schemaDefinition))
                    .topic("livk-topic")
                    .compressionType(CompressionType.LZ4)
                    .sendTimeout(0, TimeUnit.SECONDS)
                    .enableBatching(true)
                    .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS)
                    .batchingMaxMessages(1000)
                    .maxPendingMessages(1000)
                    .blockIfQueueFull(true)
                    .roundRobinRouterBatchingPartitionSwitchFrequency(10)
                    .batcherBuilder(BatcherBuilder.DEFAULT)
                    .create()) {
                PulsarMessage<String> message = new PulsarMessage<>(LocalDateTime.now(), UUID.randomUUID().toString());
                MessageId messageId = producer.newMessage()
                        .value(message)
                        .send();
                System.out.println(messageId);
            }
        }
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class PulsarMessage<T> {
        private LocalDateTime time;

        private T data;
    }
}

Exception in thread "main" org.apache.pulsar.shade.org.apache.avro.AvroTypeException: Unknown type: T
	at org.apache.pulsar.shade.org.apache.avro.specific.SpecificData.createSchema(SpecificData.java:403)
	at org.apache.pulsar.shade.org.apache.avro.reflect.ReflectData.createSchema(ReflectData.java:780)
	at org.apache.pulsar.shade.org.apache.avro.reflect.ReflectData.createFieldSchema(ReflectData.java:873)
	at org.apache.pulsar.shade.org.apache.avro.reflect.ReflectData$AllowNull.createFieldSchema(ReflectData.java:92)
	at org.apache.pulsar.shade.org.apache.avro.reflect.ReflectData.createSchema(ReflectData.java:736)
	at org.apache.pulsar.shade.org.apache.avro.specific.SpecificData$3.computeValue(SpecificData.java:328)
	at org.apache.pulsar.shade.org.apache.avro.specific.SpecificData$3.computeValue(SpecificData.java:325)
	at java.base/java.lang.ClassValue.getFromHashMap(ClassValue.java:228)
	at java.base/java.lang.ClassValue.getFromBackup(ClassValue.java:210)
	at java.base/java.lang.ClassValue.get(ClassValue.java:116)
	at org.apache.pulsar.shade.org.apache.avro.specific.SpecificData.getSchema(SpecificData.java:339)
	at org.apache.pulsar.client.impl.schema.util.SchemaUtil.extractAvroSchema(SchemaUtil.java:95)
	at org.apache.pulsar.client.impl.schema.util.SchemaUtil.createAvroSchema(SchemaUtil.java:78)
	at org.apache.pulsar.client.impl.schema.util.SchemaUtil.parseSchemaInfo(SchemaUtil.java:51)
	at org.apache.pulsar.client.impl.schema.JSONSchema.of(JSONSchema.java:91)
	at org.apache.pulsar.client.impl.PulsarClientImplementationBindingImpl.newJSONSchema(PulsarClientImplementationBindingImpl.java:222)
	at org.apache.pulsar.client.api.Schema.JSON(Schema.java:362)

When I use a class with generics as the message body, the pulsar client cannot parse it

jdk: 17 pulsar-client:2.11.0

[jak78]

Hi @livk-cloud!
Because of the Java compiler's generics type erasure, the type information of the data field of the PulsarMessage class is lost at compile time.
But the Pulsar client needs to know the type of the data field of the PulsarMessage class. This happens at runtime after the code is compiled.
Because of the type erasure, the Pulsar client is unable to get the type.
To solve this problem, you can create classes with no generic type:

public static class StringPulsarMessage extends PulsarMessage<String> {
    }

SchemaDefinition<StringPulsarMessage> schemaDefinition = SchemaDefinition.<StringPulsarMessage>builder()
                    .withPojo(StringPulsarMessage.class)
                    .build();

Please let me know if this works for you.

[livk-cloud]

public class Main {
    public static void main(String[] args) throws PulsarClientException {
        try (PulsarClient pulsarClient = PulsarClient.builder()
                .serviceUrl("pulsar://livk.com:6650")
                .build()) {
            SchemaDefinition<StrMessage> schemaDefinition = SchemaDefinition.<StrMessage>builder()
                    .withPojo(StrMessage.class)
                    .build();
            try (Producer<StrMessage> producer = pulsarClient.<StrMessage>newProducer(Schema.JSON(schemaDefinition))
                    .topic("livk-topic")
                    .compressionType(CompressionType.LZ4)
                    .sendTimeout(0, TimeUnit.SECONDS)
                    .enableBatching(true)
                    .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS)
                    .batchingMaxMessages(1000)
                    .maxPendingMessages(1000)
                    .blockIfQueueFull(true)
                    .roundRobinRouterBatchingPartitionSwitchFrequency(10)
                    .batcherBuilder(BatcherBuilder.DEFAULT)
                    .create()) {
                StrMessage message = new StrMessage(LocalDateTime.now(), UUID.randomUUID().toString());
                MessageId messageId = producer.newMessage()
                        .value(message)
                        .send();
                System.out.println(messageId);
            }
        }
    }

    @EqualsAndHashCode(callSuper = true)
    @Data
    public static class StrMessage extends PulsarMessage<String> {
        public StrMessage(LocalDateTime time, String data) {
            super(time, data);
        }
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class PulsarMessage<T> {
        private LocalDateTime time;

        private T data;
    }
}

Thank you for your answer
When I modify the code as shown above, I still get the same error

[tisonkun]

Hi @livk-cloud!

Cross refer to:

• JSONSchema should works with generic (parameterized type) POJO/DTO. #18849
• AVRO-2248 Added support for Java parameterized types avro#356
• https://issues.apache.org/jira/browse/AVRO-2248

Quick answer: you may de-parameterize PulsarMessage<T> to PulsarMessage and inline the type or use Object. This seems an Avro upstream issue.

[tisonkun]

Another approach is to annotate the field manually:

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class PulsarMessage<T> {
        private LocalDateTime time;

        @AvroSchema("{\"type\":\"string\"}")
        private T data;
    }

[livk-cloud]

Thank you for your answer, currently using Object should be the optimal solution