DanielLeens commented on PR #10935:
URL: https://github.com/apache/seatunnel/pull/10935#issuecomment-4525114038
Thanks for the contribution. I reviewed the latest head
`f5231ced78760bc8e7f33a235e673c83df01e0c2` locally against `upstream/dev`, and
I traced the actual source runtime path from `MqttSource` into
`MqttSourceReader`. I did not run Maven locally in this pass; this is a
source-level review only.
# What problem this PR solves
- User pain point
`connector-mqtt` already has a sink, but there is no MQTT source connector
for pulling messages from a broker into SeaTunnel.
- Fix approach
This PR adds a new source-side implementation with config, factory,
single-reader source, callback-based reader, docs, and basic unit tests.
- One sentence
The overall feature shape is reasonable, but the latest head still has a
risky reconnect behavior: on a long-lasting broker outage, the source can
silently stop ingesting without actually failing the task.
# Runtime chain I checked
```text
SeaTunnel streaming job
-> MqttSource.getBoundedness() => UNBOUNDED
-> MqttSource.createReader() -> new MqttSourceReader(...)
-> open()
-> Paho MqttClient.connect()
-> subscribe(topic, qos)
Message arrival
-> MqttCallbackExtended.messageArrived() [MqttSourceReader.java:165-180]
-> copy payload into LinkedBlockingQueue
-> pollNext() [MqttSourceReader.java:93-116]
-> poll queue
-> deserialize payload
-> collect SeaTunnelRow
Connection loss
-> connectionLost() [MqttSourceReader.java:137-140]
-> only LOG.warn(...)
-> no receiveException, no task failure
-> if reconnect never succeeds
-> no more callbacks
-> pollNext() keeps returning null forever
-> task looks alive but ingestion is stalled
```
# Core review
Key findings:
1. The main callback -> queue -> `pollNext()` design is clear and the normal
path is reachable.
2. Docs and option wiring are mostly aligned.
3. The failure state is incomplete: `pollNext()` only fails when
`receiveException` is set, but `connectionLost()` never sets it.
4. That means a long-lasting outage can degrade into a silent no-data state
instead of an explicit source failure.
# Findings
Issue 1: a long-lasting reconnect failure can silently stall the source
instead of surfacing a task failure
- Location:
`seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/source/MqttSourceReader.java:138`
- Problem description:
`pollNext()` only throws when `receiveException` is set. But
`connectionLost()` currently just logs a warning and relies on Paho
auto-reconnect. If the broker stays unavailable or reconnect never succeeds for
a long time, the reader receives no more callbacks, `receiveException` stays
`null`, and `pollNext()` simply keeps polling an empty queue forever.
- Potential risk:
The task appears healthy from the engine side while ingestion has actually
stopped. For a streaming source, that kind of silent stall is more dangerous
than a direct fail-fast because it delays detection and recovery.
- Best improvement suggestion:
Option A: record the disconnected state and fail `pollNext()` after a
configurable reconnect timeout if recovery never succeeds.
Option B: if you want to keep relying on auto-reconnect, then you still
need explicit health/error propagation for repeated reconnect failure instead
of only logging a warning.
- Severity: High
- Already raised by others: No
Issue 2: the new tests do not cover connection-loss / reconnect-failure
behavior
- Location:
`seatunnel-connectors-v2/connector-mqtt/src/test/java/org/apache/seatunnel/connectors/seatunnel/mqtt/source/MqttSourceTest.java:91`
- Problem description:
The current tests cover JSON happy path and queue-full failure, but they
do not cover `connectionLost()`, reconnect callbacks, or how reconnect failures
are surfaced back through `pollNext()`.
- Potential risk:
The most failure-prone part of a streaming MQTT source is the recovery
path, and that path currently has no regression protection.
- Best improvement suggestion:
Please add reader-level tests for connection loss and
reconnect/resubscribe failure propagation.
- Severity: Medium
- Already raised by others: No
# Compatibility
- API/config/defaults: compatible addition of new source capability
- Protocol: MQTT source support is new; no existing source behavior is broken
- Historical behavior: `connector-mqtt` moves from sink-only to source+sink
# Tests / coverage
- The newly added tests are structurally stable: they are in-memory only and
I did not see obvious flaky patterns such as `Thread.sleep`, external ports, or
unordered assertions.
- The main gap is coverage, not test stability.
# CI status
- GitHub currently reports:
- `mergeable=MERGEABLE`
- `mergeStateStatus=BLOCKED`
- `Notify test workflow=FAILURE`
- I checked that failure. The contributor fork already has a successful
`Build` workflow run for this same head (`JAEKWANG97/seatunnel` run
`26322480447`), but the Apache-side helper workflow failed with `401 Bad
credentials` while querying the fork check-runs. So the visible CI failure
looks workflow/integration-side rather than caused by the source code in this
PR.
### Conclusion: can merge after fixes
1. Blocking items
- Issue 1: the source must not silently stall forever when reconnect never
succeeds.
2. Suggested but non-blocking follow-up
- Issue 2: add regression coverage for connection-loss / reconnect-failure
behavior.
Overall, the feature direction makes sense and most of the source skeleton
is in place. But for the current head, I would not merge yet because the
reconnect/error state is still incomplete for a production streaming source.
--
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]
