[jira] [Created] (BEAM-7427) JmsCheckpointMark AVRO Serialisation issue with unbounded Source

["Mourad (JIRA)" <ji...@apache.org>]

Mourad created BEAM-7427:
----------------------------

             Summary: JmsCheckpointMark AVRO Serialisation issue with unbounded Source
                 Key: BEAM-7427
                 URL: https://issues.apache.org/jira/browse/BEAM-7427
             Project: Beam
          Issue Type: Bug
          Components: io-java-jms
    Affects Versions: 2.12.0
         Environment: Message Broker : solace
JMS Client (Over AMQP) : "org.apache.qpid:qpid-jms-client:0.42.0




            Reporter: Mourad


I get the following exception when reading from unbounded JMS Source:
 
{code:java}
Caused by: org.apache.avro.SchemaParseException: Illegal character in: this$0
at org.apache.avro.Schema.validateName(Schema.java:1151)
at org.apache.avro.Schema.access$200(Schema.java:81)
at org.apache.avro.Schema$Field.<init>(Schema.java:403)
at org.apache.avro.Schema$Field.<init>(Schema.java:396)
at org.apache.avro.reflect.ReflectData.createSchema(ReflectData.java:622)
at org.apache.avro.reflect.ReflectData.createFieldSchema(ReflectData.java:740)
at org.apache.avro.reflect.ReflectData.createSchema(ReflectData.java:604)
at org.apache.avro.specific.SpecificData$2.load(SpecificData.java:218)
at org.apache.avro.specific.SpecificData$2.load(SpecificData.java:215)
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)
{code}
 

The exception is thrown by Avro when introspecting {{JmsCheckpointMark}} to generate schema.

JmsIO config :

 

{{PCollection<DFAMessage> messages = pipeline.apply("read messages from the events broker", JmsIO.<DFAMessage>readMessage() .withConnectionFactory(jmsConnectionFactory) .withTopic(options.getTopic()) .withMessageMapper(new DFAMessageMapper()) .withCoder(AvroCoder.of(DFAMessage.class)));}}

 

 

 

 



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)