SEPURI-SAI-KRISHNA opened a new pull request, #12701:
URL: https://github.com/apache/seatunnel/pull/12701
### Purpose of this pull request
Three `RocketMqIT` sink tests write to a hard-coded topic and read it back
with an exact count assertion. They are `@TestTemplate` methods, so each runs
once per engine container against the same shared broker, and the read resolves
its start from a shared consumer group's committed offset. What one invocation
reads therefore depends on where the previous one stopped.
Closes #12700
Observed on `rocketmq-connector-it (11, ubuntu-latest)`:
```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` is two invocations' worth of rows: nothing was lost, twenty
rows sat in one topic and the two reads split them unevenly.
### What changes
Two changes per test, because the failure has two causes and neither fix
alone is sufficient.
**A per-invocation topic**, from the existing `uniqueTestSuffix()` helper,
passed to the job as a variable, with the config taking the variable and
keeping its old name as the default:
```java
final String topicName = "test_text_topic_" + uniqueTestSuffix();
waitForTopicRoute(topicName);
container.executeJob(
"/rocketmq-text-sink_fake_to_rocketmq.conf",
Arrays.asList("sinkTopic=" + topicName));
```
```
topic = "${sinkTopic:test_text_topic}"
```
That removes the inheritance: a topic no previous invocation wrote to has no
committed offset for the shared group, so each invocation cold-starts at
`getMinOffset()`.
**A wait for the rows to be visible** before the single consuming read,
using the `awaitTopicMaxOffset` helper already in this file:
```java
awaitTopicMaxOffset(topicName, EXPECTED_SINK_ROWS, Duration.ofMinutes(1));
Map<String, RocketMqConsumerMessage> data =
getRocketMqConsumerData(topicName);
Assertions.assertEquals(EXPECTED_SINK_ROWS, data.size());
```
Because the topic is new, every queue starts at offset 0 and the summed max
offset is exactly the number of rows stored, so this is a direct check that all
ten are readable. Retrying the read instead is not an option: it commits
offsets as it goes (`updateConsumeOffsetToBroker` then `commitSync`), so a
second call resumes past what the first consumed.
The two are complementary rather than belt-and-braces. The unique topic is
what makes the wait meaningful, since on a shared topic a count of ten could be
satisfied by another invocation's rows. The wait is what fixes the `<6>`, which
the unique topic alone would not have.
**One ordering change that comes with it.** `testSinkRocketMqMessageTag`
used to run the job and warm the route afterwards. That was survivable while
the topic was fixed, because every invocation after the first found it already
created. A per-invocation topic does not exist yet, so leaving the order alone
would make the sink depend on broker auto-creation racing the name server,
which is the `No topic route info in name server` failure this file has hit
before. All three tests now warm the route first.
The repeated literal `10` in these three tests is now a named constant,
`EXPECTED_SINK_ROWS`, since the wait and the assertion have to agree.
This follows #12396, which fixed the same isolation problem for the two tag
tests and is merged; `uniqueTestSuffix()` comes from it.
### Scope
Three sites, but **only `test_text_topic` has an observed failure.** The
other two, `test_topic` and `test_topic_message_tag`, are the identical
construct, a fixed topic plus an exact count assertion read back through a
shared consumer group, and are fixed in the same change because the cause is
the same, not because they have been seen failing. I would rather state that
than let three fixes read as three observed bugs.
### What I checked before changing the topic names
A previous attempt of mine in this file (#12393) went wrong because the
tests passed **because** of the defect, so the dependencies were checked first
this time:
- Inside `RocketMqIT`, no remaining reference to the three old topic names:
no other test reads them expecting a sink test to have populated them.
- Inside `connector-rocketmq-e2e/src/test/resources`, no config uses them as
a source.
- Repo-wide, every other occurrence is either the Kafka connector, which has
its own container and topics, or an illustrative docs example. None is affected.
The config defaults mean each file still behaves exactly as before when run
without the variable.
### Does this PR introduce any user-facing change?
No. Test resources and one test class only. No option, default, signature or
documented behaviour changes, so there is nothing to add to
`incompatible-changes.md` and no `docs/en` or `docs/zh` update.
### How was this patch tested?
Honestly, not as far as I would like, and I would rather say so than imply
otherwise.
- `test-compile` and `spotless:check` are clean on `connector-rocketmq-e2e`.
- The variable path is verified rather than assumed: `executeJob(conf,
variables)` reaches `AbstractTestContainer`, which turns each entry into a `-i
key=value` argument, which is the substitution `${sinkTopic:...}` consumes. It
is the same path #12396 uses in this module today.
- **I could not run the IT.** Docker's daemon is not reachable on this
machine, so CI is the first execution of these three tests.
- Even a green run would not prove the fix, because the failure is
intermittent. What it can show is that the three tests still pass with
per-invocation topics. The argument that the flake cannot recur is by
construction: a topic no previous invocation has written to has no committed
offset for the shared group, so there is nothing to inherit.
### Check list
* [x] If any new Jar binary package adding in your PR, please add License
Notice according [New License
Guide](https://github.com/apache/seatunnel/blob/dev/docs/en/developer/new-license.md)
(no new dependency)
* [x] If necessary, please update the documentation to describe the new
feature. (no user-facing change)
* [x] If necessary, please update `incompatible-changes.md` to describe the
incompatibility caused by this PR. (none)
* [x] If you are contributing the connector code, please check that the
following files are updated. (test-only change)
--
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]
