lizhimins opened a new pull request, #146: URL: https://github.com/apache/rocketmq-flink/pull/146
[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. -- 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]
