This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/rocketmq-flink.git
The following commit(s) were added to refs/heads/main by this push:
new 0d78ac2 [ISSUE #143] docs: document the row format, the source
metrics and the lite topic list (#147)
0d78ac2 is described below
commit 0d78ac2ec745e385cf02dd910ed36d5c75915f91
Author: lizhimins <[email protected]>
AuthorDate: Mon Oct 26 14:50:31 2026 +0800
[ISSUE #143] docs: document the row format, the source metrics and the lite
topic list (#147)
* [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.
* [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.
* [ISSUE #143] docs: document the row format, the source metrics and the
lite topic list
The SQL row format was undocumented, which is how a source that reads zero
rows
stays undiagnosed: the body is split into lines and then into fields, the
field
delimiter defaults to \u0001, and a row whose field count does not match the
column count is dropped. List the format options, the length check values
and
the two cases where a column comes from a message property instead of the
body,
and say plainly that there is no format option.
Add the source metrics, including that a positive records-lag with nothing
being
emitted is how an unregistered consumer group shows up.
Correct the scan modes to the ones the code accepts, the offset.commit.auto
default to false, and the note about client side filtering: SQL92 is
evaluated
with the broker's own filter implementation, so only the network traffic
differs
from server side filtering.
The gRPC docs described LITE mode as a wildcard subscription on the main
topic.
The subscription is the list of sub lite topics, and a main topic bound
without
it receives nothing, so document rocketmq.source.sub-lite-topics as
required and
note that the list is also the isolation boundary.
Record that the legacy source assigns the queues Flink allocated to the
subtask,
and that an async send failure now fails the task.
---
docs/grpc-connector.md | 17 ++++++++++++-----
docs/legacy-connector.md | 9 +++++++++
docs/remoting-connector.md | 32 +++++++++++++++++++++++++++-----
docs/sql-connector.md | 28 +++++++++++++++++++++++++++-
4 files changed, 75 insertions(+), 11 deletions(-)
diff --git a/docs/grpc-connector.md b/docs/grpc-connector.md
index 14ba709..d4fe7bf 100644
--- a/docs/grpc-connector.md
+++ b/docs/grpc-connector.md
@@ -4,7 +4,7 @@ The gRPC connector (`flink-connector-rocketmq-grpc`) targets
RocketMQ 5.x using
`rocketmq-client-java` 5.2.1+ and offers two consumption modes. In **SIMPLE**
mode — the default —
the source subscribes to a normal topic (with an optional tag / SQL92 filter)
and acknowledges the
messages itself once the enclosing checkpoint completes. In **LITE** mode,
which must be selected
-explicitly, the source binds one main lite topic with a wildcard subscription,
never acks by itself,
+explicitly, the source binds one main lite topic and subscribes the sub lite
topics you list, never acks by itself,
and hands the ack / throttle decision to downstream operators. Section 2
details both modes.
This document covers both the design (architecture, semantics, decisions and
trade-offs) and the
@@ -54,10 +54,14 @@ acknowledges:
batches the receipt handles and acks them through that very consumer once
the enclosing
checkpoint completes (at-least-once).
- **LITE** (opt-in, requires `rocketmq.source.mode` = `LITE`) — the source
binds one main lite topic
- via a `LiteSimpleConsumer` with a wildcard (generalized) subscription, so
the broker funnels the
- messages of all sub topics through one receive stream. The source never
acks: every emitted record
- carries a self-contained receipt handle, and the downstream operator takes
the ack / throttle
- decision — which is what enables fair per-sub-topic throttling.
+ via a `LiteSimpleConsumer` and subscribes every sub lite topic named in
+ `rocketmq.source.sub-lite-topics`, so the broker funnels the messages of
those sub topics through
+ one receive stream. Binding the main topic on its own subscribes nothing:
the SDK's lite topic set
+ stays empty and the broker delivers no message, which is why `build()`
rejects LITE mode without
+ that option. The source never acks: every emitted record carries a
self-contained receipt handle,
+ and the downstream operator takes the ack / throttle decision — which is
what enables fair
+ per-sub-topic throttling. The list is also an isolation boundary: a sub
topic you do not list is
+ not delivered to you, its backlog included.
RocketMQ 5.x offers a push client (`PushConsumer`) and pull/Pop clients
(`SimpleConsumer` / `LiteSimpleConsumer`). Both connector modes build on the
Pop clients
@@ -239,6 +243,8 @@ acks and *when*.
### LITE mode
+- `rocketmq.source.sub-lite-topics` is required and is the whole subscription:
the broker delivers
+ only the sub topics listed there.
- The source never acks; the ack decision is deferred to downstream operators,
which gives
at-least-once with user-controlled granularity.
- `invisible-duration` must cover the full downstream processing time of a
message (including
@@ -413,6 +419,7 @@ RocketMQGrpcSink<String> sink =
RocketMQGrpcSink.<String>builder()
| --- | --- | --- | --- |
| `rocketmq.source.mode` | ConsumerMode | `SIMPLE` | Consumption mode:
`SIMPLE` (default; normal topic, checkpoint-aligned source ack) or `LITE` (main
lite topic, downstream ack) |
| `rocketmq.source.main-topic` | String | (none) | Main lite topic to bind,
required in LITE mode |
+| `rocketmq.source.sub-lite-topics` | List<String> | (none) | Sub lite topics
to subscribe, required in LITE mode. A main topic bound without them subscribes
nothing and receives no message |
| `rocketmq.source.topic` | String | (none) | Normal topic to subscribe,
required in SIMPLE mode |
| `rocketmq.source.filter-expression` | String | (none) | SIMPLE-mode tag /
SQL92 filter expression (absent = receive all messages) |
| `rocketmq.source.filter-type` | String | `TAG` | Type of
`filter-expression`: `TAG` or `SQL92` |
diff --git a/docs/legacy-connector.md b/docs/legacy-connector.md
index 1a2bb66..7543555 100644
--- a/docs/legacy-connector.md
+++ b/docs/legacy-connector.md
@@ -40,12 +40,21 @@ public interface KeyValueDeserializationSchema<T> extends
ResultTypeQueryable<T>
These policies only take effect when the job starts without state. When
recovering from a
checkpoint, offsets are restored from the checkpointed state.
+Queues are allocated by Flink, not by the client: `open()` fetches the queues
of the topic, keeps
+the share of this subtask and assigns exactly those to the consumer. A tag
sub-expression is still
+applied by the broker; a SQL92 expression cannot be pushed down in assign mode
and is evaluated
+against the message properties on the client, with the same semantics as
server side filtering.
+
## RocketMQSink (SinkFunction)
Construct it with a `Properties`; provides at-least-once when checkpointing is
enabled and
`withBatchFlushOnCheckpoint(true)` is set. Otherwise reliability depends on
the producer retry
policy (sync by default; `withAsync(true)` switches to async).
+With `withAsync(true)` and batch flush off, a send failure cannot fail the
call that triggered it,
+so the first failure is kept and rethrown from the next `invoke` or from
`finish`. The task then
+fails instead of continuing to checkpoint records that were never written.
+
```java
stream.addSink(new
RocketMQSink(producerProps).withBatchFlushOnCheckpoint(true));
```
diff --git a/docs/remoting-connector.md b/docs/remoting-connector.md
index 74981f2..b31415a 100644
--- a/docs/remoting-connector.md
+++ b/docs/remoting-connector.md
@@ -74,9 +74,9 @@ Shared by source and sink.
| --- | --- | --- | --- |
| `rocketmq.source.topic` | String | (none) | Topic, required |
| `rocketmq.source.group` | String | (none) | Consumer group, required |
-| `rocketmq.source.filter.tag` | String | `*` | Tag filter (client-side in
assign mode) |
-| `rocketmq.source.filter.sql` | String | (none) | SQL92 filter (client-side
in assign mode) |
-| `rocketmq.source.startup.scan.mode` | String | `latest` | `earliest` /
`latest` / `group` / `timestamp` / `specific` |
+| `rocketmq.source.filter.tag` | String | `*` | Tag filter, `tagA\|\|tagB` for
several. Evaluated on the client |
+| `rocketmq.source.filter.sql` | String | (none) | SQL92 filter over message
properties. Evaluated on the client with the same `rocketmq-filter`
implementation the broker uses, so predicate semantics match server side
filtering. An expression that does not compile fails when the reader is built |
+| `rocketmq.source.startup.scan.mode` | String | `latest` | `earliest` /
`latest` / `timestamp` / `committed`. Any other value is rejected |
| `rocketmq.source.startup.offset.timestamp` | Long | (none) | Startup
timestamp; auto-infers `scan.mode=timestamp` |
| `rocketmq.source.allocate.strategy` | String | `consistent-hash` |
Queue-to-subtask allocation strategy |
| `rocketmq.source.pull.threads.num` | Integer | 20 | Pull thread pool size |
@@ -85,12 +85,34 @@ Shared by source and sink.
| `rocketmq.source.pull.threshold.all` | Long | 10000 | Total flow control
threshold |
| `rocketmq.source.pull.rpc.timeout` | Long | 20000 | Pull RPC timeout (ms) |
| `rocketmq.source.pull.rpc.exception.delay` | Long | 3000 | Pull retry delay
on exception (ms) |
-| `rocketmq.source.offset.commit.auto` | Boolean | true | Auto offset commit |
-| `rocketmq.source.offset.commit.interval` | Long | 5000 | Auto commit
interval (ms) |
+| `rocketmq.source.offset.commit.auto` | Boolean | false | Let the client
commit the offset of everything it polled on its own timer. Off by default:
Flink commits the offsets of checkpointed state instead, and broker side
offsets never decide where a restored job starts reading, so this only changes
what consumer progress monitoring reports |
+| `rocketmq.source.offset.commit.interval` | Long | 5000 | Interval of the
client side auto commit (ms) |
| `rocketmq.source.offset.commit.checkpoint` | Boolean | true | Commit offsets
on checkpoint |
| `rocketmq.source.trace.enable` | Boolean | true | Message trace |
| `rocketmq.source.trace.topic` | String | (none) | Customized trace topic |
+## Source metrics
+
+Reported per message queue under `RocketmqSourceReader`, keyed by `topic` and
+`queue`:
+
+| Metric | Type | Description |
+| --- | --- | --- |
+| `currentOffset` | Gauge | Offset the reader has consumed up to |
+| `committedOffset` | Gauge | Offset last committed to the broker |
+| `records-lag` | Gauge | Records the broker still holds beyond
`currentOffset` |
+| `commitsSucceeded` / `commitsFailed` | Counter | Offset commit results,
under the reader group |
+| `bytes-consumed-total` | Counter | Under
`RocketMQConsumer.consumer-fetch-manager-metrics` |
+
+The lag of every assigned queue is also summed into Flink's standard
+`pendingRecords`. Lag needs the broker's maximum offset, so the reader
refreshes
+it asynchronously at most once every ten seconds; reading a metric never
+triggers a remote call.
+
+A queue with a positive `records-lag` and no records being emitted means the
+broker is not delivering to this consumer group, which is how an unregistered
+group or a missing subscription relationship shows up.
+
## Sink options (`rocketmq.sink.*`)
| Key | Type | Default | Description |
diff --git a/docs/sql-connector.md b/docs/sql-connector.md
index 4a65649..f0f07c8 100644
--- a/docs/sql-connector.md
+++ b/docs/sql-connector.md
@@ -60,11 +60,37 @@ CREATE TABLE rocketmq_source (
);
```
+## Row format
+
+The source maps a message body onto a row by splitting the body into lines and
+then each line into fields. Both delimiters have to be set for anything but the
+defaults, and a body that does not use them produces no row at all.
+
+| Key | Type | Default | Description |
+| --- | --- | --- | --- |
+| `rocketmq.client.message.encoding` | String | `UTF-8` | Character set of the
body |
+| `rocketmq.client.message.line.delimiter` | String | `\n` | Splits the body
into rows |
+| `rocketmq.client.message.field.delimiter` | String | `\u0001` | Splits a row
into columns |
+| `rocketmq.client.message.length.check` | String | `NONE` | What to do when
the field count does not match the column count: `NONE`, `SKIP`, `SKIP_SILENT`,
`PAD`, `EXCEPTION`. Any other value is rejected |
+| `rocketmq.source.column.error.debug` | Boolean | true | Log every field that
fails to convert, instead of at most once a minute |
+
+A column whose name matches a message property is filled from that property
+rather than from the body. A single `VARBINARY` column receives the whole body
+unchanged.
+
+There is no `format` option: JSON or CSV bodies need the delimiters above, or a
+DataStream job with its own `DeserializationSchema`.
+
## Notes
- `rocketmq.source.api.new.enable` (default true) selects the FLIP-27 source;
setting it to
false falls back to the legacy `SourceFunction` path, which has no FLIP-27
fault-tolerance
integration and is not recommended.
-- In assign mode, tag / SQL92 filters are evaluated on the client side.
+- Tag and SQL92 filters are evaluated on the client: assign mode, which is
what keeps queue
+ allocation under the enumerator's control, can only carry a tag
sub-expression to the broker.
+ SQL92 uses the same `rocketmq-filter` implementation the broker uses, so
predicate semantics
+ are the same; only the network traffic differs.
+- A missing required option is reported by key name alone. Flink's own
validation message
+ repeats every table option, so keep credentials out of the `WITH` clause
where you can.
- Message keys can be derived from columns via `rocketmq.sink.key.columns`; a
dynamic tag column
via `rocketmq.sink.tag.dynamic.enable` + `rocketmq.sink.tag.dynamic.column`.
