apache / apache/pekko-connectors
UntypedMqttFlowSpec: mqtt client flow should establish a bidirectional connection and subscribe to a topic
- Dominant language
- Scala
- Stars
- 82
- Forks
- 50
- Avg merge
- 1d 2h
- Merged PRs (30d)
- 78
Description
Possibly related to #468
```
[info] UntypedMqttFlowSpec:
[info] mqtt client flow
[info] - should establish a bidirectional connection and subscribe to a topic *** FAILED *** (5 seconds, 42 milliseconds)
[info] java.lang.AssertionError: assertion failed: expected no StreamSupervisor children, but got [Actor[pekko://UntypedMqttFlowSpec/system/Materializers/StreamSupervisor-3/flow-7-0-ignoreSink#-1480604319], Actor[pekko://UntypedMqttFlowSpec/system/Materializers/StreamSupervisor-3/flow-8-0-unnamed#1503904569]]
[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.apache.pekko.testkit.TestKitBase.awaitAssert(TestKit.scala:342)
[info] at org.apache.pekko.testkit.TestKitBase.awaitAssert$(TestKit.scala:331)
[info] at org.apache.pekko.testkit.TestKit.awaitAssert(TestKit.scala:982)
[info] at org.apache.pekko.stream.testkit.scaladsl.StreamTestKit$.$anonfun$assertNoChildren$1(StreamTestKit.scala:61)
[info] at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
[info] at org.apache.pekko.testkit.TestKitBase.within(TestKit.scala:429)
[info] at org.apache.pekko.testkit.TestKitBase.within$(TestKit.scala:416)
[info] at org.apache.pekko.testkit.TestKit.within(TestKit.scala:982)
[info] at org.apache.pekko.testkit.TestKitBase.within(TestKit.scala:444)
[info] at org.apache.pekko.testkit.TestKitBase.within$(TestKit.scala:444)
[info] at org.apache.pekko.testkit.TestKit.within(TestKit.scala:982)
[info] at org.apache.pekko.stream.testkit.scaladsl.StreamTestKit$.assertNoChildren(StreamTestKit.scala:61)
[info] at org.apache.pekko.stream.testkit.scaladsl.StreamTestKit$.assertAllStagesStopped(StreamTestKit.scala:43)
[info] at docs.scaladsl.MqttFlowSpecBase.$anonfun$new$2(MqttFlowSpec.scala:65)
[info] at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
[info] at org.scalatest.OutcomeOf.outcomeOf(OutcomeOf.scala:85)
[info] at org.scalatest.OutcomeOf.outcomeOf$(OutcomeOf.scala:83)
[info] at org.scalatest.OutcomeOf$.outcomeOf(OutcomeOf.scala:104)
[info] at org.scalatest.Transformer.apply(Transformer.scala:22)
[info] at org.scalatest.Transformer.apply(Transformer.scala:20)
[info] at org.scalatest.wordspec.AnyWordSpecLike$$anon$3.apply(AnyWordSpecLike.scala:1240)
[info] at org.apache.pekko.stream.connectors.testkit.scaladsl.LogCapturing.withFixture(LogCapturing.scala:70)
[info] at org.apache.pekko.stream.connectors.testkit.scaladsl.LogCapturing.withFixture$(LogCapturing.scala:66)
[info] at docs.scaladsl.MqttFlowSpecBase.withFixture(MqttFlowSpec.scala:48)
[info] at org.scalatest.wordspec.AnyWordSpecLike.invokeWithFixture$1(AnyWordSpecLike.scala:1238)
[info] at org.scalatest.wordspec.AnyWordSpecLike.$anonfun$runTest$1(AnyWordSpecLike.scala:1250)
[info] at org.scalatest.SuperEngine.runTestImpl(Engine.scala:306)
[info] at org.scalatest.wordspec.AnyWordSpecLike.runTest(AnyWordSpecLike.scala:1250)
[info] at org.scalatest.wordspec.AnyWordSpecLike.runTest$(AnyWordSpecLike.scala:1232)
[info] at docs.scaladsl.MqttFlowSpecBase.runTest(MqttFlowSpec.scala:48)
[info] at org.scalatest.wordspec.AnyWordSpecLike.$anonfun$runTests$1(AnyWordSpecLike.scala:1309)
[info] at org.scalatest.SuperEngine.$anonfun$runTestsInBranch$1(Engine.scala:413)
[info] at scala.collection.immutable.List.foreach(List.scala:431)
[info] at org.scalatest.SuperEngine.traverseSubNodes$1(Engine.scala:401)
[info] at org.scalatest.SuperEngine.runTestsInBranch(Engine.scala:390)
[info] at org.scalatest.SuperEngine.$anonfun$runTestsInBranch$1(Engine.scala:427)
[info] at scala.collection.immutable.List.foreach(List.scala:431)
[info] at org.scalatest.SuperEngine.traverseSubNodes$1(Engine.scala:401)
[info] at org.scalatest.SuperEngine.runTestsInBranch(Engine.scala:396)
[info] at org.scalatest.SuperEngine.runTestsImpl(Engine.scala:475)
[info] at org.scalatest.wordspec.AnyWordSpecLike.runTests(AnyWordSpecLike.scala:1309)
[info] at org.scalatest.wordspec.AnyWordSpecLike.runTests$(AnyWordSpecLike.scala:1308)
[info] at docs.scaladsl.MqttFlowSpecBase.runTests(MqttFlowSpec.scala:48)
[info] at org.scalatest.Suite.run(Suite.scala:1114)
[info] at org.scalatest.Suite.run$(Suite.scala:1096)
[info] at docs.scaladsl.MqttFlowSpecBase.org$scalatest$wordspec$AnyWordSpecLike$$super$run(MqttFlowSpec.scala:48)
[info] at org.scalatest.wordspec.AnyWordSpecLike.$anonfun$run$1(AnyWordSpecLike.scala:1354)
[info] at org.scalatest.SuperEngine.runImpl(Engine.scala:535)
[info] at org.scalatest.wordspec.AnyWordSpecLike.run(AnyWordSpecLike.scala:1354)
[info] at org.scalatest.wordspec.AnyWordSpecLike.run$(AnyWordSpecLike.scala:1352)
[info] at docs.scaladsl.MqttFlowSpecBase.org$scalatest$BeforeAndAfterAll$$super$run(MqttFlowSpec.scala:48)
[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.MqttFlowSpecBase.run(MqttFlowSpec.scala:48)
[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)
```
Contributor guide
Research direction
Start with the failing UntypedMqttFlowSpec test and the shared setup in MqttFlowSpec.scala around line 65, then run that MQTT flow test to reproduce the leftover StreamSupervisor children. Done means the bidirectional connection and topic subscription test passes without leaving those children running.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- scala
- Domain
- networking
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 38/100