[lizhimins]

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 RejectedExecutionExceptions 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.