Repository navigation
[ISSUE #142] fix(grpc): subscribe the configured lite topics and fail on a rejected consumer - #146
Merged
Merged
Conversation
…ntly An acceptance run against a commercial 5.5.1 cluster found a family of defects with the same shape: the job stays RUNNING, checkpoints succeed, nothing is logged, and the data is wrong or absent. Every fix below either makes the behaviour correct or makes the failure loud. Source. rocketmq.source.filter.sql was accepted, logged as a warning and then ignored, so every message was delivered regardless of the predicate; assign mode cannot push a SQL92 expression to the broker, so it is now evaluated on the client with rocketmq-filter, the implementation the broker uses, and an expression that does not compile fails at reader construction. RowDeserializationSchema read its format options from keys such as "fielddelimiter" while the table factory registers and documents "rocketmq.client.message.field.delimiter". The DDL values never arrived, the field delimiter stayed at its default, and every record of a multi column table was dropped as "field missing" -- a SQL source that returns zero rows without failing. Both row deserializers now read the canonical keys, and an unsupported message.length.check is rejected instead of ignored. RocketMQSourceReaderMetrics declared nine metric names and was threaded through four classes, but its constructor and both registration methods had empty bodies, so a running job exposed no RocketMQ metric at all. Offsets, commit counters and per queue lag are now reported, and the lag is summed into Flink's pendingRecords. Lag needs the broker's maximum offset, which the reader refreshes asynchronously at most once every ten seconds. RocketMQSourceFunction could not start at all: open() allocates the subtask's queues itself and then seeks each one, but the consumer was in subscribe mode, where allocation belongs to the client's asynchronous rebalance, so the assigned list was still empty and every seek failed with "The message queue is not in assigned list". The allocated queues are now assigned explicitly, which also matches the new source. offset.commit.auto was read only to print its value while the consumer was hardcoded to setAutoCommit(false); pull.rpc.timeout was declared and never read. Both are applied now, and the auto-commit default becomes false to match the behaviour the code always had. An unknown startup.scan.mode fell through to latest, making a misspelled mode look like a healthy job reading from the wrong position; it is rejected. Sink. RocketMQSinkBuilder accepted a MessageQueueSelector, null checked it and stored it, and both send paths then called send(Message), so records that had to stay ordered were spread over every queue while the API said otherwise. The selector now reaches the send call with the record as its sharding argument. The committer logged a failed commit without the throwable, so a failure produced a message with no cause and no stack trace. A record that failed to emit was reported as "Failed to deserialize message due to" wherever the failure actually came from, and named no record; the wrapper now covers only the deserialization call and the message carries the message id, topic, queue and offset. The async path of the legacy RocketMQSink logged a send failure and carried on, so a job whose every record was rejected stayed RUNNING, kept checkpointing and wrote nothing. The first failure is now parked and rethrown from invoke() and finish(). Table. A validation failure repeated every table option, and a WITH clause normally carries the credentials, publishing the access key to the JobManager log and the REST exception history. Both factories now report the missing keys only, before delegating to the helper, which still detects unknown options. Removed. rocketmq.source.startup.offset.strategy and rocketmq.sink.serialize.format were read by nothing, and the first was even registered as a DDL option; an option that does nothing should not be offered. The duplicate CollectorOption key sets are gone with the fix above, as is the InnerConsumerImpl.wakeup() that paused every assigned queue with no matching resume anywhere in the codebase, and the message key separator that was an empty string instead of MessageConst.KEY_SEPARATOR.
…d fail on a rejected consumer LITE mode never subscribed to a single sub lite topic. The builder bound the main topic and nothing else: there was no subscribeLite call anywhere in the module and no option to name the sub topics, so the SDK's lite topic set stayed empty and the broker delivered nothing. The job ran, checkpointed and logged nothing while reading zero rows. Add rocketmq.source.sub-lite-topics and fail in build() when LITE mode is configured without it, so a missing subscription cannot be mistaken for an idle topic. Acknowledgement stays with the user's downstream operator: the source exposes the receipt handle and RocketMQAckClient performs the ack, because a message that is never acknowledged is redelivered forever once its invisible duration expires. The Table path strips the receipt handle and can therefore never ack, so source.mode='lite' is rejected there instead of being accepted into an endless redelivery loop. A consumer group the server registered as LITE cannot be used by a SIMPLE mode source: the proxy answers ILLEGAL_CONSUMER_GROUP on every request. The receive workers caught every ClientException and retried forever, so the job stayed RUNNING, read no record and reported nothing. Failures that no retry can fix now stop the reader and are rethrown on the fetcher thread, which is the only thread whose exception reaches the Flink task. FORBIDDEN stays retryable on purpose, since the proxy also returns it as backpressure while a lite consumer holds too many unacknowledged messages. Consumer creation caught ClientException only, while the SDK reports a rejected client as an unchecked exception wrapped by its Guava service. That escaped uncaught and left the reader marked as started with no consumer, so a later split assignment silently did nothing. The catch is widened to Throwable, and a lite subscription rejected after the consumer is built now closes it instead of leaking its channels and threads. A validation failure also repeated every table option, publishing the access key to the JobManager log and the REST exception history; the factory now reports the missing keys only.
lizhimins
force-pushed
the
issue-142-grpc-lite
branch
from
October 26, 2026 06:43
a64ada6 to
b30b9ad
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
[ISSUE #142] fix(grpc): subscribe the configured lite topics and fail on a rejected consumer
LITE mode never subscribed to a single sub lite topic. The builder bound the
main topic and nothing else: there was no subscribeLite call anywhere in the
module and no option to name the sub topics, so the SDK's lite topic set stayed
empty and the broker delivered nothing. The job ran, checkpointed and logged
nothing while reading zero rows.
Add rocketmq.source.sub-lite-topics and fail in build() when LITE mode is
configured without it, so a missing subscription cannot be mistaken for an idle
topic. Acknowledgement stays with the user's downstream operator: the source
exposes the receipt handle and RocketMQAckClient performs the ack, because a
message that is never acknowledged is redelivered forever once its invisible
duration expires. The Table path strips the receipt handle and can therefore
never ack, so source.mode='lite' is rejected there instead of being accepted
into an endless redelivery loop.
A consumer group the server registered as LITE cannot be used by a SIMPLE mode
source: the proxy answers ILLEGAL_CONSUMER_GROUP on every request. The receive
workers caught every ClientException and retried forever, so the job stayed
RUNNING, read no record and reported nothing. Failures that no retry can fix now
stop the reader and are rethrown on the fetcher thread, which is the only thread
whose exception reaches the Flink task. FORBIDDEN stays retryable on purpose,
since the proxy also returns it as backpressure while a lite consumer holds too
many unacknowledged messages.
Consumer creation caught ClientException only, while the SDK reports a rejected
client as an unchecked exception wrapped by its Guava service. That escaped
uncaught and left the reader marked as started with no consumer, so a later
split assignment silently did nothing. The catch is widened to Throwable, and a
lite subscription rejected after the consumer is built now closes it instead of
leaking its channels and threads.
A validation failure also repeated every table option, publishing the access key
to the JobManager log and the REST exception history; the factory now reports
the missing keys only.