Lost connection caused by ClassCastException

Please fill out the form below before submitting, thank you!

- [x] Bug exists Release Version 1.2.0 ( Master Branch)

If this is a bug regarding the Android Service, please raise the bug here instead: https://github.com/eclipse/paho.mqtt.android/issues/new

public void demo() throws MqttException {
final MqttAsyncClient client = new MqttAsyncClient("tcp://iot.eclipse.org:1883", "testClientId");

final DisconnectedBufferOptions bufferOptions = new DisconnectedBufferOptions();
bufferOptions.setBufferEnabled(true);
bufferOptions.setPersistBuffer(true);

client.setBufferOpts(bufferOptions);

client.setCallback(new MqttCallbackExtended() {
@Override
public void connectComplete(boolean reconnect, String serverURI) {
try {
client.subscribe("testTopic1", 0);
} catch (MqttException e) {
e.printStackTrace();
}
}

@Override
public void connectionLost(Throwable cause) {
cause.printStackTrace();
}

@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
}

@Override
public void deliveryComplete(IMqttDeliveryToken token) {
}
});

client.connect();

client.publish("testTopic2", new MqttMessage("message".getBytes(StandardCharsets.UTF_8)));
}

When running the example code above, the connection is terminated because of a ClassCastException thrown in ClientState.java:583:

MqttException (0) - java.lang.ClassCastException: org.eclipse.paho.client.mqttv3.internal.wire.MqttSubscribe cannot be cast to org.eclipse.paho.client.mqttv3.internal.wire.MqttPublish
at org.eclipse.paho.client.mqttv3.internal.CommsCallback.run(CommsCallback.java:220)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.ClassCastException: org.eclipse.paho.client.mqttv3.internal.wire.MqttSubscribe cannot be cast to org.eclipse.paho.client.mqttv3.internal.wire.MqttPublish
at org.eclipse.paho.client.mqttv3.internal.ClientState.persistBufferedMessage(ClientState.java:583)
at org.eclipse.paho.client.mqttv3.internal.ClientComms.sendNoWait(ClientComms.java:183)
at org.eclipse.paho.client.mqttv3.MqttAsyncClient.subscribe(MqttAsyncClient.java:1091)
at org.eclipse.paho.client.mqttv3.MqttAsyncClient.subscribe(MqttAsyncClient.java:1035)
at com.example.MqttDemo$1.connectComplete(MqttDemo.java:35)
at org.eclipse.paho.client.mqttv3.internal.ConnectActionListener.onSuccess(ConnectActionListener.java:104)
at org.eclipse.paho.client.mqttv3.internal.CommsCallback.fireActionEvent(CommsCallback.java:321)
at org.eclipse.paho.client.mqttv3.internal.CommsCallback.handleActionComplete(CommsCallback.java:260)
at org.eclipse.paho.client.mqttv3.internal.CommsCallback.run(CommsCallback.java:190)
... 7 more
Feb 13, 2019 10:30:20 AM DisconnectedMessageBuffer run
WARNUNG: null: Un-Persisting Buffered message key={0}

I think this might be related to #606.