You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2023/03/28 07:33:40 UTC
[GitHub] [pulsar] livk-cloud created a discussion: The pulsar client generics have an exception
GitHub user livk-cloud created a discussion: The pulsar client generics have an exception
The code is shown below
```java
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;
}
}
```
```shell
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
GitHub link: https://github.com/apache/pulsar/discussions/19946
----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org
[GitHub] [pulsar] livk-cloud added a comment to the discussion: The pulsar client generics have an exception
Posted by GitBox <gi...@apache.org>.
GitHub user livk-cloud added a comment to the discussion: The pulsar client generics have an exception
Thank you for your answer, currently using Object should be the optimal solution
GitHub link: https://github.com/apache/pulsar/discussions/19946#discussioncomment-5452739
----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org
[GitHub] [pulsar] tisonkun added a comment to the discussion: The pulsar client generics have an exception
Posted by GitBox <gi...@apache.org>.
GitHub user tisonkun added a comment to the discussion: The pulsar client generics have an exception
Hi @livk-cloud!
Cross refer to:
* https://github.com/apache/pulsar/discussions/18849
* https://github.com/apache/avro/pull/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.
GitHub link: https://github.com/apache/pulsar/discussions/19946#discussioncomment-5452137
----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org
[GitHub] [pulsar] livk-cloud added a comment to the discussion: The pulsar client generics have an exception
Posted by GitBox <gi...@apache.org>.
GitHub user livk-cloud added a comment to the discussion: The pulsar client generics have an exception
```java
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;
}
}
```
When I modify the code as shown above, the error still occurs
GitHub link: https://github.com/apache/pulsar/discussions/19946#discussioncomment-5452033
----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org
[GitHub] [pulsar] livk-cloud edited a comment on the discussion: The pulsar client generics have an exception
Posted by GitBox <gi...@apache.org>.
GitHub user livk-cloud edited a comment on the discussion: The pulsar client generics have an exception
```java
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, the error still occurs
GitHub link: https://github.com/apache/pulsar/discussions/19946#discussioncomment-5452033
----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org
[GitHub] [pulsar] tisonkun added a comment to the discussion: The pulsar client generics have an exception
Posted by GitBox <gi...@apache.org>.
GitHub user tisonkun added a comment to the discussion: The pulsar client generics have an exception
Another approach is to annotate the field manually:
```java
@Data
@NoArgsConstructor
@AllArgsConstructor
public static class PulsarMessage<T> {
private LocalDateTime time;
@AvroSchema("{\"type\":\"string\"}")
private T data;
}
```
GitHub link: https://github.com/apache/pulsar/discussions/19946#discussioncomment-5452186
----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org
[GitHub] [pulsar] livk-cloud edited a discussion: The pulsar client generics have an exception
Posted by GitBox <gi...@apache.org>.
GitHub user livk-cloud edited a discussion: The pulsar client generics have an exception
The code is shown below
```java
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;
}
}
```
```shell
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
GitHub link: https://github.com/apache/pulsar/discussions/19946
----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org
[GitHub] [pulsar] livk-cloud edited a comment on the discussion: The pulsar client generics have an exception
Posted by GitBox <gi...@apache.org>.
GitHub user livk-cloud edited a comment on the discussion: The pulsar client generics have an exception
```java
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
GitHub link: https://github.com/apache/pulsar/discussions/19946#discussioncomment-5452033
----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org
[GitHub] [pulsar] jak78 added a comment to the discussion: The pulsar client generics have an exception
Posted by GitBox <gi...@apache.org>.
GitHub user jak78 added a comment to the discussion: The pulsar client generics have an exception
Hi @livk-cloud!
Because of the Java compiler's [generics type erasure](https://www.baeldung.com/java-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:
```java
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.
GitHub link: https://github.com/apache/pulsar/discussions/19946#discussioncomment-5451908
----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org