lizhimins opened a new issue, #141: URL: https://github.com/apache/rocketmq-flink/issues/141
An acceptance run against a commercial RocketMQ 5.5.1 cluster on Flink 1.20.3 found a family of defects in the remoting module with the same shape: the job stays RUNNING, checkpoints succeed, nothing is logged, and the data is wrong or absent. **Source** - `rocketmq.source.filter.sql` is accepted, logged as a warning and then ignored, so every message is delivered regardless of the predicate. - `RowDeserializationSchema` reads its format options from keys such as `fielddelimiter`, while the table factory registers and documents `rocketmq.client.message.field.delimiter`. The DDL values never arrive, the field delimiter stays at its `\u0001` default, and every record of a multi column table is dropped as "field missing". A SQL source returns zero rows without failing. - `RocketMQSourceReaderMetrics` declares nine metric names and is threaded through four classes, but its constructor and both registration methods have empty bodies. A running job exposes no RocketMQ metric at all: no `currentOffset`, no `committedOffset`, no commit counters, no `records-lag`, and Flink's `pendingRecords` stays unset. - `RocketMQSourceFunction` cannot start. `open()` allocates the subtask's queues itself and then seeks each one, but the consumer is in subscribe mode where allocation belongs to the client's asynchronous rebalance, so the assigned list is still empty and every seek fails with `The message queue is not in assigned list`. Reproduces at parallelism 1 and 2. - `rocketmq.source.offset.commit.auto` is read only to print its value while the consumer is hardcoded to `setAutoCommit(false)`; `rocketmq.source.pull.rpc.timeout` is declared and never read. - An unknown `rocketmq.source.startup.scan.mode` falls through to `latest`, so a misspelled mode looks like a healthy job reading from the wrong position. **Sink** - `RocketMQSinkBuilder` accepts a `MessageQueueSelector` and null checks it, but both send paths call `send(Message)`. The selector is never read, so records that must stay ordered are spread over every queue while the API says otherwise. - `RocketMQCommitter` logs a failed commit without the throwable, so a failure produces a message with no cause and no stack trace. - A record that fails to emit is reported as `Failed to deserialize message due to` wherever the failure came from, and names no record. - The async path of the legacy `RocketMQSink` logs a send failure and carries on, so a job whose every record was rejected stays RUNNING, keeps checkpointing and writes nothing. **Table** - A validation failure repeats every table option, and a WITH clause normally carries the credentials, publishing the access key to the JobManager log and the REST exception history. **Dead options** - `rocketmq.source.startup.offset.strategy` and `rocketmq.sink.serialize.format` are read by nothing, and the first is even registered as a DDL option. -- 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]
