0.14: NotSerializableException: io.moquette.broker.SessionRegistry$PublishedMessage

[hylkevds]

I'm testing 0.14 on our server, and woke up to a console filled with the below exception.
SessionRegistry.PublishedMessage and its superclass do not implement Serializable. Should they be put into the store?

Expected behavior

No exceptions...

Actual behavior

07:41:12.367 [nioEventLoopGroup-3-1] ERROR i.m.broker.NewNettyMQTTHandler - Error processing protocol message: SUBSCRIBE
java.lang.IllegalStateException: This store is closed [1.4.199/4]
	at org.h2.mvstore.DataUtils.newIllegalStateException(DataUtils.java:883)
	at org.h2.mvstore.MVStore.checkOpen(MVStore.java:2634)
	at org.h2.mvstore.MVStore.getMapName(MVStore.java:2736)
	at org.h2.mvstore.MVMap.beforeWrite(MVMap.java:889)
	at org.h2.mvstore.MVMap.operate(MVMap.java:1649)
	at org.h2.mvstore.MVMap.put(MVMap.java:151)
	at io.moquette.persistence.H2SubscriptionsRepository.addNewSubscription(H2SubscriptionsRepository.java:41)
	at io.moquette.broker.subscriptions.CTrieSubscriptionDirectory.add(CTrieSubscriptionDirectory.java:101)
	at io.moquette.broker.PostOffice.subscribeClientToTopics(PostOffice.java:81)
	at io.moquette.broker.MQTTConnection.processSubscribe(MQTTConnection.java:329)
	at io.moquette.broker.MQTTConnection.handleMessage(MQTTConnection.java:72)
	at io.moquette.broker.NewNettyMQTTHandler.channelRead(NewNettyMQTTHandler.java:58)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
...
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.IllegalStateException: java.lang.IllegalArgumentException: Could not serialize io.moquette.broker.SessionRegistry$PublishedMessage@2c5195b1 [1.4.199/0] [1.4.199/3]
	at org.h2.mvstore.DataUtils.newIllegalStateException(DataUtils.java:883)
	at org.h2.mvstore.MVStore.store(MVStore.java:1194)
	at org.h2.mvstore.MVStore.commit(MVStore.java:1166)
	at io.moquette.persistence.H2Builder.lambda$initStore$0(H2Builder.java:45)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	... 1 common frames omitted
Caused by: java.lang.IllegalArgumentException: Could not serialize io.moquette.broker.SessionRegistry$PublishedMessage@2c5195b1 [1.4.199/0]
	at org.h2.mvstore.DataUtils.newIllegalArgumentException(DataUtils.java:857)
	at org.h2.mvstore.type.ObjectDataType.serialize(ObjectDataType.java:359)
	at org.h2.mvstore.type.ObjectDataType$SerializedObjectType.write(ObjectDataType.java:1527)
	at org.h2.mvstore.type.ObjectDataType.write(ObjectDataType.java:127)
	at org.h2.mvstore.type.ObjectDataType.write(ObjectDataType.java:121)
	at org.h2.mvstore.Page$Leaf.writeValues(Page.java:1556)
	at org.h2.mvstore.Page.write(Page.java:732)
	at org.h2.mvstore.Page$Leaf.writeUnsavedRecursive(Page.java:1565)
	at org.h2.mvstore.Page$NonLeaf.writeChildrenRecursive(Page.java:1298)
	at org.h2.mvstore.Page$NonLeaf.writeUnsavedRecursive(Page.java:1284)
	at org.h2.mvstore.Page$NonLeaf.writeChildrenRecursive(Page.java:1298)
	at org.h2.mvstore.Page$NonLeaf.writeUnsavedRecursive(Page.java:1284)
	at org.h2.mvstore.MVStore.storeNow(MVStore.java:1290)
	at org.h2.mvstore.MVStore.store(MVStore.java:1190)
	... 9 common frames omitted
Caused by: java.io.NotSerializableException: io.moquette.broker.SessionRegistry$PublishedMessage
	at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1184)
	at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
	at org.h2.mvstore.type.ObjectDataType.serialize(ObjectDataType.java:356)
	... 21 common frames omitted

Steps to reproduce

See below

Minimal yet complete reproducer code (or URL to code) or complete log file

See below

Moquette MQTT version

• 0.14
• 0.15-SNAPSHOT

JVM version (e.g. java -version)

• OpenJDK 1.8
• OpenJDK 11

OS version (e.g. uname -a)

Linux

[hylkevds]

I just got the exception again, but this time I could still get at the entire log file. The exception happens out of the blue, with no prior errors.
In the log I've found 6 variations of the exception that differ in the first part, the the "causes" sections are all identical.

[hylkevds]

Finally managed to reliably reproduce it reliably. All one needs to do is have a SessionRegistry$PublishedMessage be stored in H2. That however, is easier said than done :)

Fast way to reproduce:

1. git clone https://github.com/hylkevds/moquetteTests.git
2. In your favourite IDE
   1. Set a breakpoint at Session.java:337
   2. debug tests.moquettetests.MoquetteTest

What it does:

• Create a broker
  • With H2 store
  • With BrokerConstants.AUTOSAVE_INTERVAL_PROPERTY_NAME set to 1 (second)
• Create a paho client
  • That has cleansession = false
  • That listens on some topics with QOS=1
• Create a thread that posts messages
  • With QOS=2
  • With a frequency high enough to force messages on the queue

Why this reproduces it: The breakpoint makes sure that any message put on the queue is not removed, thus making H2 store it as soon as the autosave triggers (every second)

[hylkevds]

I've created a PR that seems to fix the problem. I'm pretty sure I got the serialization of the ByteBuf correct.
PR: #568