[lizhimins]

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