lizhimins opened a new pull request, #145: URL: https://github.com/apache/rocketmq-flink/pull/145
[ISSUE #141] fix(remoting): stop delivering the wrong data silently 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. -- 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]
