SEPURI-SAI-KRISHNA commented on PR #12408: URL: https://github.com/apache/seatunnel/pull/12408#issuecomment-5761447743
Thanks @DanielLeens, this was a good catch and all five are addressed in the push above. **Issue 1.** You are right, and the part I got wrong is in the PR description as much as the code: I wrote that the publish is "gated on RUNNING" as if that meant the source was ready. It does not. `PhysicalVertex#deployInternal` sets RUNNING as soon as the deploy call returns, while `cycle.open()` and therefore `MqttSourceReader#open` run later on the worker, so the subscribe happens after the gate opens and MQTT drops anything sent in between. Your reading of our own fork log makes the margin concrete: the reader subscribed 0.5s after RUNNING and we published 1.8s later, so it passed on slack from the 2s poll interval rather than on anything guaranteed. I took Option A. A uniquely named probe row is published on every poll until one comes back on the echo topic, and only then are the eight asserted rows published. Probe ids start at 1000 so they cannot collide with the asserted range even if one is redelivered. One extra benefit over the version you reviewed: this also covers the echo subscriber's own subscription, which the old code assumed had completed before `connectClient` returned. **Issue 2.** Future kept. Every poll in all three waits now calls a helper that rethrows the submission failure if the job has already ended, so the case this IT exists to catch fails immediately with the real cause instead of after three minutes with `expected RUNNING but was null`. After the cancel the test waits for CANCELED and then joins the future, so the thread leak check in `doExecuteJob` runs before `@AfterAll` closes the container rather than racing it. **Issue 3.** The payloads are parsed now and the assertion is on the `(id, name)` pairs, so a dropped or mistyped `id` fails. Still membership rather than an exact count, as you suggested, since QoS 1 redelivery is not a defect. **Issue 4.** The `try` now starts immediately after the job is launched, so the readiness wait and the publish are inside it. **Issue 5.** Switched to `MqttCallbackExtended` and the subscribe moved into `connectComplete`, so a reconnect re-establishes it. On verification: this is test-only, and I ran `spotless:apply` and a test-compile of `connector-mqtt-e2e` locally. I could not run the container test, since Docker is not available in my environment, so CI is the real signal here. Worth saying plainly given the whole point of this round is that the previous version passed CI while still being racy. On your extensibility note, the checkpoint and restore path is genuinely not covered and I agree it is a separate follow-up rather than something to fold in here. -- 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]
