SEPURI-SAI-KRISHNA opened a new issue, #12700: URL: https://github.com/apache/seatunnel/issues/12700
### Search before asking - [X] I had searched in the [issues](https://github.com/apache/seatunnel/issues?q=is%3Aissue+label%3A%22Bug%22) and found no similar issues. ### What happened Three `RocketMqIT` sink tests run a fake-source job that writes ten rows to a hard-coded topic, then read the topic back and assert the count is exactly ten. They are `@TestTemplate` methods, so each runs once per engine container against the **same shared RocketMQ broker and consumer group**. Observed on `rocketmq-connector-it (11, ubuntu-latest)`, seven template invocations: ```text testTextFormatSinkRocketMq{TestContainer}[1] expected: <10> but was: <6> FAIL testTextFormatSinkRocketMq{TestContainer}[2] expected: <10> but was: <14> FAIL testTextFormatSinkRocketMq{TestContainer}[3..7] pass ``` `6 + 14 = 20`, exactly two invocations' worth of rows. Nothing was lost; the rows were attributed to the wrong invocation. ### Mechanism Two causes compound, and it matters that it is two, because neither alone produces this. **1. The read does not wait for the rows to be visible.** `getRocketMqConsumerData` (`RocketMqIT.java:460`) assigns the topic's queues, seeks, then polls until a poll comes back empty. There is no wait for the ten rows the job reported writing to become readable, so a read that starts too early returns fewer. That is the `<6>`. **2. The topic and consumer group are shared, so the shortfall is inherited.** The read resolves its start from the group's committed offset (`:492`): ```java long currentOffset = currentOffsets.containsKey(mq) ? currentOffsets.get(mq) // the shared group's committed offset : queueOffsets.get(mq).getMinOffset(); // cold start consumer.seek(mq, currentOffset); ``` Invocation 1 committed after reading only six, so invocation 2 seeked to that point and read the four rows left behind plus its own ten. That is the `<14>`. Worth stating because it rules out the simpler explanation: with a shared topic but no visibility shortfall, invocation 1 cold-starts at `getMinOffset()`, reads all ten and commits at ten; invocation 2 seeks to ten and reads its own ten. Every invocation passes. Sharing alone is not sufficient to fail, and a per-invocation topic alone would not have fixed the `<6>`. Retrying the read is not an available fix: it commits offsets to the broker as it goes (`updateConsumeOffsetToBroker` then `commitSync`), so a second call resumes past what the first consumed and returns less, not more. ### Affected sites | Test | Topic | Assertion | | --- | --- | --- | | `testSinkRocketMq` | `test_topic` (`:195`) | `:202` | | `testTextFormatSinkRocketMq` | `test_text_topic` (`:211`) | `:213` | | `testSinkRocketMqMessageTag` | `test_topic_message_tag` (`:406`) | `:414` | Being precise about the evidence: **only `test_text_topic` has an observed failure.** The other two are the same construct, a fixed topic plus an exact count assertion in a `@TestTemplate` against a shared broker and consumer group, and are reported together because the cause is identical, not because they have been seen failing. ### Prior art #12396 (merged) fixed exactly this class for `RocketMqIT`'s two **tag** tests, by giving each invocation a unique topic and consumer group passed as job variables. The `uniqueTestSuffix()` helper it added is still on `dev` (`:981`). These three sink tests were not covered by it. ### SeaTunnel Version dev (3.0.0-SNAPSHOT) ### SeaTunnel Config Not applicable: this is an E2E test isolation defect, not a job configuration problem. ### Running Command ```shell ./mvnw -pl seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e verify -DskipUT -DskipIT=false ``` ### Error Exception ```log expected: <10> but was: <6> expected: <10> but was: <14> ``` ### Zeta or Flink or Spark Version Reproduced on the Zeta legs of `rocketmq-connector-it`; the defect is independent of engine because the shared state is the broker topic. ### Java or Scala Version JDK 11 ### Screenshots Not applicable. ### Are you willing to submit PR? - [X] Yes I am willing to submit a PR! ### Code of Conduct - [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct) -- 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]
