MqttSessionSpec: MQTT server connector should produce a duplicate publish on the server given two client connections

Leaked children:

```
[info] MqttSessionSpec:
...
[info] MQTT server connector
...
[info] - should produce a duplicate publish on the server given two client connections *** FAILED *** (46 seconds, 689 milliseconds)
[info] java.lang.AssertionError: assertion failed: expected no StreamSupervisor children, but got [Actor[pekko://mqtt-spec/system/Materializers/StreamSupervisor-4/flow-336-0-seqSink#845195087], Actor[pekko://mqtt-spec/system/Materializers/StreamSupervisor-4/flow-310-0-unnamed#558388173], Actor[pekko://mqtt-spec/system/Materializers/StreamSupervisor-4/flow-325-0-unnamed#-636644616], Actor[pekko://mqtt-spec/system/Materializers/StreamSupervisor-4/flow-309-0-unnamed#-638425441], Actor[pekko://mqtt-spec/system/Materializers/StreamSupervisor-4/flow-418-0-seqSink#-120447922], Actor[pekko://mqtt-spec/system/Materializers/StreamSupervisor-4/flow-390-0-unnamed#-60431972], Actor[pekko://mqtt-spec/system/Materializers/StreamSupervisor-4/flow-386-0-seqSink#1379330535], Actor[pekko://mqtt-spec/system/Materializers/StreamSupervisor-4/flow-358-0-unnamed#-1021969704], Actor[pekko://mqtt-spec/system/Materializers/StreamSupervisor-4/flow-392-0-unnamed#1493844900], Actor[pekko://mqtt-spec/system/Materializers/StreamSupervisor-4/flow-406-0-unnamed#-781711807], Actor[pekko://mqtt-spec/system/Materializers/StreamSupervisor-4/flow-391-0-unnamed#514514286], Actor[pekko://mqtt-spec/system/Materializers/StreamSupervisor-4/flow-408-0-unnamed#307681820], Actor[pekko://mqtt-spec/system/Materializers/StreamSupervisor-4/flow-340-0-unnamed#-1553917495], Actor[pekko://mqtt-spec/system/Materializers/StreamSupervisor-4/flow-402-0-seqSink#-1890725493], Actor[pekko://mqtt-spec/system/Materializers/StreamSupervisor-4/flow-368-0-seqSink#-1022558259], Actor[pekko://mqtt-spec/system/Materializers/StreamSupervisor-4/flow-341-0-unnamed#-203018638], Actor[pekko://mqtt-spec/system/Materializers/StreamSupervisor-4/flow-357-0-unnamed#2057313086], Actor[pekko://mqtt-spec/system/Materializers/StreamSupervisor-4/flow-366-0-unnamed#611842993], Actor[pekko://mqtt-spec/system/Materializers/StreamSupervisor-4/flow-342-0-unnamed#300813464], Actor[pekko://mqtt-spec/system/Materializers/StreamSupervisor-4/flow-374-0-unnamed#1363580884], Actor[pekko://mqtt-spec/system/Materializers/StreamSupervisor-4/flow-320-0-seqSink#-10399093], Actor[pekko://mqtt-spec/system/Materializers/StreamSupervisor-4/flow-326-0-unnamed#-1336180487], Actor[pekko://mqtt-spec/system/Materializers/StreamSupervisor-4/flow-407-0-unnamed#-2136156977], Actor[pekko://mqtt-spec/system/Materializers/StreamSupervisor-4/flow-375-0-unnamed#-552257211], Actor[pekko://mqtt-spec/system/Materializers/StreamSupervisor-4/flow-367-0-ignoreSink#294263610], Actor[pekko://mqtt-spec/system/Materializers/StreamSupervisor-4/flow-324-0-unnamed#-84133769], Actor[pekko://mqtt-spec/system/Materializers/StreamSupervisor-4/flow-308-0-unnamed#-628749277], Actor[pekko://mqtt-spec/system/Materializers/StreamSupervisor-4/flow-356-0-unnamed#953562305], Actor[pekko://mqtt-spec/system/Materializers/StreamSupervisor-4/flow-376-0-unnamed#2071441288], Actor[pekko://mqtt-spec/system/Materializers/StreamSupervisor-4/flow-352-0-seqSink#704675480]]
[info] at scala.Predef$.assert(Predef.scala:223)
[info] at org.apache.pekko.stream.testkit.scaladsl.StreamTestKit$.$anonfun$assertNoChildren$2(StreamTestKit.scala:64)
[info] at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
[info] at org.scalatest.wordspec.AnyWordSpecLike.run$(AnyWordSpecLike.scala:1352)
[info] at docs.scaladsl.MqttSessionSpec.org$scalatest$BeforeAndAfterAll$$super$run(MqttSessionSpec.scala:45)
[info] at org.scalatest.BeforeAndAfterAll.liftedTree1$1(BeforeAndAfterAll.scala:213)
[info] at org.scalatest.BeforeAndAfterAll.run(BeforeAndAfterAll.scala:210)
[info] at org.scalatest.BeforeAndAfterAll.run$(BeforeAndAfterAll.scala:208)
[info] at docs.scaladsl.MqttSessionSpec.run(MqttSessionSpec.scala:45)
[info] at org.scalatest.tools.Framework.org$scalatest$tools$Framework$$runSuite(Framework.scala:321)
[info] at org.scalatest.tools.Framework$ScalaTestTask.execute(Framework.scala:517)
[info] at sbt.TestRunner.runTest$1(TestFramework.scala:153)
[info] at sbt.TestRunner.run(TestFramework.scala:168)
[info] at sbt.TestFramework$$anon$3$$anonfun$$lessinit$greater$1.$anonfun$apply$1(TestFramework.scala:336)
[info] at sbt.TestFramework$.sbt$TestFramework$$withContextLoader(TestFramework.scala:296)
[info] at sbt.TestFramework$$anon$3$$anonfun$$lessinit$greater$1.apply(TestFramework.scala:336)
[info] at sbt.TestFramework$$anon$3$$anonfun$$lessinit$greater$1.apply(TestFramework.scala:336)
[info] at sbt.TestFunction.apply(TestFramework.scala:348)
[info] at sbt.Tests$.processRunnable$1(Tests.scala:475)
[info] at sbt.Tests$.$anonfun$makeSerial$1(Tests.scala:481)
[info] at sbt.std.Transform$$anon$3.$anonfun$apply$2(Transform.scala:47)
[info] at sbt.std.Transform$$anon$4.work(Transform.scala:69)
[info] at sbt.Execute.$anonfun$submit$2(Execute.scala:283)
[info] at sbt.internal.util.ErrorHandling$.wideConvert(ErrorHandling.scala:24)
[info] at sbt.Execute.work(Execute.scala:292)
[info] at sbt.Execute.$anonfun$submit$1(Execute.scala:283)
[info] at sbt.ConcurrentRestrictions$$anon$4.$anonfun$submitValid$1(ConcurrentRestrictions.scala:265)
[info] at sbt.CompletionService$$anon$2.call(CompletionService.scala:65)
[info] at java.util.concurrent.FutureTask.run(FutureTask.java:266)
[info] at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
[info] at java.util.concurrent.FutureTask.run(FutureTask.java:266)
[info] at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
[info] at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
[info] at java.lang.Thread.run(Thread.java:750)
```

Start by running the failing MqttSessionSpec test and read MqttSessionSpec.scala around its BeforeAndAfterAll cleanup at line 45. Use the StreamTestKit.assertNoChildren failure to trace the leaked stream actors; done means the duplicate-publish scenario passes without remaining StreamSupervisor children.

Written by the indexing model from the issue text.