ClassCastException storing message in persistBufferedMessage

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

- [x] Bug exists Release Version 1.2.5 ( Master Branch)
- [x] Bug exists in MQTTv3 Client on Snapshot Version 1.2.6-SNAPSHOT (Develop Branch)
- [?] Bug exists in MQTTv5 Client on Snapshot Version 1.2.6-SNAPSHOT (Develop Branch)

This bug relates to:
- [https://github.com/eclipse/paho.mqtt.java/issues/606](https://github.com/eclipse/paho.mqtt.java/issues/606)
- [https://github.com/eclipse/paho.mqtt.java/issues/622](https://github.com/eclipse/paho.mqtt.java/issues/622)

And this fix:
- [https://github.com/eclipse/paho.mqtt.java/commit/cdde8046e9c6a677323cb57e9e879f47f7452e0b](https://github.com/eclipse/paho.mqtt.java/commit/cdde8046e9c6a677323cb57e9e879f47f7452e0b)

Under some circunstances, a SUBSCRIBE or UNSUBSCRIBE message might be persisted in the persistBufferedMessage is set **when the client has lost the connection with the server**.

```
public void sendNoWait(MqttWireMessage message, MqttToken token) throws MqttException {
final String methodName = "sendNoWait";
if (isConnected() ||
(!isConnected() && message instanceof MqttConnect) ||
(isDisconnecting() && message instanceof MqttDisconnect)) {
if(disconnectedMessageBuffer != null && disconnectedMessageBuffer.getMessageCount() != 0){
//@TRACE 507=Client Connected, Offline Buffer available, but not empty. Adding message to buffer. message={0}
log.fine(CLASS_NAME, methodName, "507", new Object[] {message.getKey()});
if(disconnectedMessageBuffer.isPersistBuffer()){
if (message instanceof MqttPublish) {
this.clientState.persistBufferedMessage(message);
}
}
disconnectedMessageBuffer.putMessage(message, token);
} else {
this.internalSend(message, token);
}
} else if(disconnectedMessageBuffer != null) {
//@TRACE 508=Offline Buffer available. Adding message to buffer. message={0}
log.fine(CLASS_NAME, methodName, "508", new Object[] {message.getKey()});
if(disconnectedMessageBuffer.isPersistBuffer()){
this.clientState.persistBufferedMessage(message);
}
disconnectedMessageBuffer.putMessage(message, token);
} else {
//@TRACE 208=failed: not connected
log.fine(CLASS_NAME, methodName, "208");
throw ExceptionHelper.createMqttException(MqttException.REASON_CODE_CLIENT_NOT_CONNECTED);
}
}
```

In this scenario, the message enters in the second branch of the conditional: `else if (disconnectedMessageBuffer != null) { ...`
And the code tries to persist it in the buffer, what ends up causing a ClassCastException when trying to store the message in the buffer as a MQTTPublish object in the ClientState class:

```
this.persistence.put(key, (MqttPublish)message);
```

My suggestion, which may be insufficient in case of a race condition, is to implement the same safe-check that was added in the issue 606 validating the message is an instance of the MqttPublish class:

```
(...)
} else if(disconnectedMessageBuffer != null) {
//@TRACE 508=Offline Buffer available. Adding message to buffer. message={0}
log.fine(CLASS_NAME, methodName, "508", new Object[] {message.getKey()});
if(disconnectedMessageBuffer.isPersistBuffer()){
if (message instanceof MqttPublish) {
this.clientState.persistBufferedMessage(message);
}
}
disconnectedMessageBuffer.putMessage(message, token);
}
(...)
```

Start at sendNoWait and trace the disconnectedMessageBuffer persistence path into ClientState.persistBufferedMessage, especially the cast shown in the issue. Compare the guard added for issue 606 and verify that SUBSCRIBE and UNSUBSCRIBE messages no longer cause a ClassCastException while buffered publishes retain their persistence behavior.

Written by the indexing model from the issue text.