Luke Cwik created BEAM-5496:
-------------------------------
Summary: MqttIO fails to deserialize checkpoint
Key: BEAM-5496
URL: https://issues.apache.org/jira/browse/BEAM-5496
Project: Beam
Issue Type: Bug
Components: io-java-mqtt
Reporter: Luke Cwik
Source of bug report:
[https://lists.apache.org/thread.html/3de5a946bcb539dea9f18a31f712d6af5b66f9fbb6b01eed452c5afb@%3Cdev.beam.apache.org%3E]
There is a bug of the built-in MqttIO, please check the
<https://github.com/apache/beam/blob/master/sdks/java/io/mqtt/src/main/java/org/apache/beam/sdk/io/mqtt/MqttIO.java#L336>,
this readObject() method forget to invoke the "stream.defaultReadObject()"
method.
{code:java}
// set an empty list to messages when deserialize
private void readObject(java.io.ObjectInputStream stream)
throws IOException, ClassNotFoundException {
messages = new ArrayList<>();
}{code}
So there is an exception while the runner tried to deserialize the checkpoint
object.
{code:java}
java.lang.RuntimeException: org.apache.beam.sdk.coders.CoderException: 95
unexpected extra bytes after decoding
org.apache.beam.sdk.io.mqtt.MqttIO$MqttCheckpointMark@6764e219
at
org.apache.beam.runners.direct.DirectRunner$DirectPipelineResult.waitUntilFinish(DirectRunner.java:340)
...{code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
