lizhimins opened a new issue, #161:
URL: https://github.com/apache/rocketmq-flink/issues/161
## Background
#152 made periodic partition discovery tolerate route-fetch failures: a
round where every topic fails is skipped, and an empty discovery result no
longer revokes the whole assignment. Two gaps remain in
`RocketMQSourceEnumerator`.
## Defect 1: partial fetch failure across multiple topics still revokes the
failed topic's queues
`requestServiceDiscovery` collects `failedTopics`, but only uses it to skip
the round when **every** topic failed. On a partial failure it returns the
queues of the successful topics only. `getSourceChangeResult` is a plain set
difference against `allocatedSet`, so the failed topic's queues land in
`decreaseSet` and are revoked (stop-splits + full re-load-balance). When the
fetch recovers on a later round, the same queues are re-initialized as new.
Example: a job subscribes `t1;t2`, both allocated. One round `t1` succeeds
and `t2` times out:
```
latestSet = {t1-q0, t1-q1}
decreaseSet = allocatedSet - latestSet = {t2-q0, t2-q1} <- t2 was never
gone
```
A failed fetch says nothing about the route; the code comment claims failed
topics are excluded from the round, but only the all-failed case is
implemented. Single-topic jobs are unaffected (the only topic failing = all
failed = skipped), which is why the #152 cluster verification did not surface
this.
## Defect 2: a revoked queue that reappears restarts from the configured
starting selector
When a queue is decreased, `initializeSourceSplits` discards its
`checkedOffsets` entry. If the queue later reappears (broker rolling restart,
scale-down followed by scale-up, or the recovery phase of Defect 1), it goes
through the increase path and its starting offset is resolved with the
configured `startingOffsetsSelector` again. With `earliest` that replays the
whole queue from offset 0.
The migration path already rebuilds surviving queues from `checkedOffsets`
(the last checkpoint-reported offset), so runtime moves resume from
checkpointed progress; re-added queues should follow the same rule. The
configured selector should only apply to genuinely new queues.
## Proposed fix
1. Carry `failedTopics` from the discovery callable to the coordinator
handler and remove queues of failed topics from `decreaseSet` before acting on
the change result.
2. On decrease, retain the last checkpointed offset of the revoked queue; on
increase, prefer that retained offset over the selector, and fall back to the
selector only for queues never seen before.
--
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]
