pjfanning opened a new pull request, #1977: URL: https://github.com/apache/pekko-connectors/pull/1977
### Motivation `MqttFlowTest.establishServerBidirectionalConnectionAndSubscribeToATopic` intermittently times out ([example](https://github.com/apache/pekko-connectors/actions/runs/36731404597/job/109941682609?pr=1975)). The server logs the client's `Connect` and then never replies with `ConnAck`. The connection handler feeds events into a `BroadcastHub` that has two consumers: 1. `source.runForeach(...)`, which sends `ConnAck`/`SubAck`/`PubAck` 2. the `source` returned to `flatMapMerge`, drained by `Sink.ignore` A `BroadcastHub` only holds elements back while it has **no** consumers. If (2) attaches first, the `Connect` can go out before (1) is attached, so nothing ever answers it. The streams the timed-out test leaves running then also fail the next test's `assertAllStagesStopped` (the `establishClientBidirectionalConnectionAndSubscribeToATopic` failure in the same run). ### Modification In the server flow of `MqttFlowTest` and `MqttFlowSpec`, use `BroadcastHub` with `startAfterNrOfConsumers = 2` (buffer size unchanged at the default 256) so nothing is emitted until both consumers are attached. Because this code sits in the `#create-streaming-bind-flow` snippet, the documentation example is fixed too. ### Result The handler always sees the `Connect`, so the server test no longer depends on the order in which consumers attach. ### Tests - `sbt mqtt-streaming/Test/javafmt mqtt-streaming/Test/scalafmt mqtt-streaming/Test/compile` - `sbt 'mqtt-streaming/testOnly docs.javadsl.MqttFlowTest docs.scaladsl.*'` (3 runs): the server flow tests pass in `MqttFlowTest`, `TypedMqttFlowSpec` and `UntypedMqttFlowSpec`. The broker-dependent tests fail locally only because no MQTT broker was running (connection refused on `localhost:1883`); CI starts one. ### References Refs #468 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
