apache / apache/pekko-connectors

UntypedMqttFlowSpec: mqtt client flow should establish a bidirectional connection and subscribe to a topic

Open
#772 0 comments 0 reactions 0 assignees View on GitHub
flaky test
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.