SEPURI-SAI-KRISHNA opened a new pull request, #12408: URL: https://github.com/apache/seatunnel/pull/12408
### Purpose of this pull request Follow-up to @DanielLeens's review on #12406, which recorded the missing source IT as Issue 2, and to the offer I made in that PR's description to open it separately. `connector-mqtt-e2e` contained only `MqttSinkIT`. Nothing exercised the source end to end, which is why a source that could not be submitted at all survived from the connector's introduction until a user reported it in #12405: `MqttSourceConfig` was not `Serializable`, so Zeta failed to serialize the logical DAG. #12406 fixed that and added a unit test pinning the field. This covers the submission path that the missing field broke. ### How the test works The MQTT source rejects `BATCH` (`MqttSource.getBoundedness()` throws), so the job is `STREAMING` and does not finish on its own. The test therefore follows the shape `FlussSourceIT` already uses for its streaming test: start the job asynchronously with a numeric job id, wait for `RUNNING`, drive data through it, assert, then `cancelJob` in a `finally`. It is Zeta only, `@DisabledOnContainer` for Flink and Spark, for the same reason Fluss gives: the status and cancel calls are Zeta-only container APIs. The job reads from one topic and echoes to a second through the MQTT sink, so the test can observe what the source produced from outside the job. That mirrors how `FlussSourceIT` reads its streaming results back out of a sink table, and it reuses the broker and subscriber setup `MqttSinkIT` already established. Two details are specific to MQTT rather than copied from Fluss: * **Messages are published only after the job reports `RUNNING`.** Fluss writes rows before the job starts because its log is durable and an earliest read picks them up. MQTT keeps no backlog for a subscriber that was not connected, so anything published before the source subscribes is dropped rather than replayed. * **The assertion waits for every published row, not for an exact count.** Both hops run at QoS 1, which is at-least-once, so a redelivery would fail an equality assertion for a reason that is not a defect in the source. ### Does this PR introduce _any_ user-facing change? No. One new E2E test class and one new job config. No production code. ### How was this patch tested? `./mvnw -q -DskipTests verify -pl seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mqtt-e2e` on JDK 11 passes, covering the enforcer checks, `spotless:check` and compilation, and `test-compile` passes. `spotless:check` was also run separately with up-to-date caching disabled. I have not run the IT locally; it needs Docker and the SeaTunnel engine container, so this PR's own `mqtt-connector-it` job is the first real execution. I would rather say that than imply local coverage I do not have. ### On adding a new test class This creates `MqttSourceIT` rather than extending an existing class, against the usual guidance. There is no source IT to extend, and folding a streaming source test into `MqttSinkIT`, whose lifecycle and assertions are built around a batch sink job, would make both harder to read. If you would prefer it inside `MqttSinkIT` regardless, say so and I will move it. ### A caveat I would rather state than have found An asynchronous streaming job, a broker container, a subscribe-then-publish ordering and a cancel at the end is a shape that can become flaky, which is exactly the class of problem #12322 and #12323 dealt with in `RocketMqIT`. I have tried to keep it tight: the publish is gated on the job reporting `RUNNING` rather than on a sleep, there are no fixed ports, the waits are Awaitility with generous ceilings, and the cancel is in a `finally` so a failed assertion still tears the job down. If it proves unstable in practice I would rather shrink or remove it than let it become noise. -- 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]
