lizhimins opened a new pull request, #149:
URL: https://github.com/apache/rocketmq-flink/pull/149
### Context
An acceptance run of this connector against a commercial RocketMQ 5.5.1
cluster on Flink 1.20.3
(Application mode, K8s) found a family of defects with the same shape as
#141: the job stays
RUNNING, checkpoints succeed, nothing is logged, and the data is wrong or
absent. This PR carries
the fixes from that round that are still open, grouped into four topics.
Each one was verified on
the cluster with clean-room topics before being committed; the per-defect
root causes and code
evidence are in the acceptance report kept outside this repo.
### What is in this PR
**1. Transactional commits are lost when END_TRANSACTION lands on another
node (the headline fix).**
A proxy keeps the transaction context of a half message in its own process
memory
(`TransactionDataManager`, keyed by producer group + transaction id,
expiring after
`transactionDataExpireMillis`, 30s by default). The committer used to
resolve the target address
from its own publish route table, which is populated independently of the
writer's and may name a
different node; with two committer subtasks each rolling its own dice, runs
silently lost half or
all of a checkpoint while every health signal stayed green. Seven controlled
experiments (same
client version, one fresh TRANSACTION topic per run) pin the cause:
same-address commits succeed
4/4, cross-pod commits are answered `code=0 SUCCESS` but land nothing 2/2,
and a same-address commit
delayed 45s is discarded the same way (the 30s TTL). The fix records the
address the writer actually
sent to on the `SendCommittable` right after the send, and the committer
uses it, falling back to
route resolution only for restored committables without an address.
`SendCommittableSerializer`
moves to version 2 and still reads version 1 state.
**2. Poison records and the remaining dead/misleading options.** A record
whose deserialization
throws used to fail the task, which restored from the checkpoint, re-read
the same record and
rewrote everything it had already emitted each round. It is now dropped with
a WARN carrying
message id, topic, queue and offset, while the split offset advances past
it; failures raised while
emitting an already-deserialized record still fail the task, so a real write
failure cannot hide
behind the poison path. `send.pending.max` was declared with exactly this
description but never
read; the writer now bounds in-flight async sends with a semaphore of that
size (and releases the
permit when the send call throws synchronously). The SQL option
`retry.times` mapped onto a key
nobody read; it now acts as an alias for `send.retry.times`.
`InnerConsumerImpl.commitOffset`
updated a local store with `increaseOnly=true` and never persisted, which
had no effect at all; it
now publishes the checkpointed offset, rollbacks included, so the
broker-side display offset tracks
the job.
**3. Transactional commits were submitted on a bounded executor that rejects
under backlog.**
With 10k records queued, `Commit transaction message` was logged 8055 times
instead of 10000 and
694 `RejectedExecutionException`s appeared; the sink stalled at 9650/10000
while the source had
already read everything. Commit and rollback now run on the calling thread.
**4. An idle source spun the fetch loop a hundred times a second.**
`rocketmq.client.poll.timeout`
defaulted to 10ms. The client poll waits on a blocking queue and returns as
soon as records arrive,
so raising the default to 1000ms costs nothing on the data path; it only
cuts the idle spin to one
fetch a second. `wakeUp()` merely sets a flag and the consumer deliberately
has no interruptible
poll, so this timeout is the sole bound on how long a fetch can outlive a
close request — one
second keeps that prompt. A test asserts both the default and a configured
value reach `poll()`.
The docs also record the deployment constraint that follows from topic 1:
the checkpoint interval
must stay well below the server-side transaction state TTL, because Flink
only commits at the end of
a checkpoint.
### Server-side note (not fixable here)
When the transaction context is missing or expired,
`AbstractRemotingActivity` maps
`TRANSACTION_DATA_NOT_FOUND` to `ResponseCode.SUCCESS` and
`TransactionActivity` discards the
returned future, so the client sees success while nothing lands and nothing
is logged. Topic 1
removes the wrong-node cause on the client side; the TTL cause remains a
deployment constraint
(keep the checkpoint interval well below `transactionDataExpireMillis`)
until the server side is
addressed separately.
### Verification
- Clean-room cluster runs on the commercial 5.5.1 deployment: transactional
sink lands 101/101 at
parallelism 1 and 2 (missing 0 / unexpected 0 / duplicate 0), and the
route-refresh fallback
warning drops from 4 to 0.
- Module test suite: 138 tests green, including 4 new ones for the
serializer round trip and v1
compatibility, the writer recording the address, and the committer using
it.
- `mvn spotless:check` under JDK 11 and a full `mvn clean package` of all
five modules pass.
Happy to split this into per-topic PRs if maintainers prefer; the commits
are already separated by
topic.
--
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]
