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 04f20de [ISSUE #137] feat: add a simple consumer mode for
normal-topic consumption (#138)
04f20de is described below
commit 04f20debcd1549911fb4b507cbfadd74157f52c0
Author: lizhimins <[email protected]>
AuthorDate: Tue Sep 22 16:02:33 2026 +0800
[ISSUE #137] feat: add a simple consumer mode for normal-topic consumption
(#138)
* [ISSUE #137] feat: add a simple consumer mode for normal-topic consumption
Add a SIMPLE consumption mode to the gRPC connector so a normal topic can be
consumed through the SDK SimpleConsumer, next to the existing LITE mode that
binds a main lite topic through the LiteSimpleConsumer.
- rocketmq.source.mode (SQL: source.mode) selects the mode and defaults to
SIMPLE; LITE must now be requested explicitly.
- SIMPLE subscribes one topic with an optional tag / SQL92 filter
expression and
acknowledges messages itself: the receipt handles of emitted records are
grouped per checkpoint by CheckpointAckTracker and acked from
notifyCheckpointComplete through the very consumer that received them, so
no
second gRPC connection is needed. Ack failures are tolerated and logged,
so a
single stale handle can never fail the job; un-acked messages are
redelivered
once their invisible duration expires, which is the at-least-once
guarantee.
- PopConsumer abstracts the two SDK Pop clients, which expose the same
receive / ack / changeInvisibleDuration operations but share no common
interface, keeping the split reader mode-agnostic.
- RocketMQLiteAckClient is renamed to RocketMQAckClient and now serves only
the
LITE downstream-ack path, which genuinely needs credentials and pooling.
- The SQL factory exposes source.mode, source.filter-expression and
source.filter-type; docs/grpc-connector.md documents both modes, the
checkpoint-ack semantics and how to size the invisible duration.
Verified end to end against an open-source cluster and an Alibaba Cloud
commercial instance: 50 messages received, acked after checkpoint
completion and
zero redelivery once the invisible duration expired.
BREAKING CHANGE: the source now consumes a normal topic with a
checkpoint-aligned SimpleConsumer by default. Jobs relying on the main lite
topic with downstream acknowledgement must select LITE explicitly via
setMode(ConsumerMode.LITE) or 'source.mode' = 'lite'.
* [ISSUE #137] chore: apply spotless formatting
The changes of this PR were written on a JDK 21 toolchain, where the parent
pom's
java17 profile disables spotless, so google-java-format never validated
them and
a few statements and javadoc paragraphs kept a manual line break that the
formatter collapses. Run spotless:apply to restore the canonical
formatting; the
diff is line breaks only, no behaviour changes.
* [ISSUE #137] fix: drop the empty protected_branches key from .asf.yaml
ASF Infra rejects the file with "when expecting a mapping, found a blank
string": `protected_branches:` has been present without a value since the
file
was created, and YAML parses it as null. Because the parse fails, the whole
`github` section is discarded, so the description, homepage, labels and the
issues / wiki feature flags never took effect.
Remove the empty key so the section parses again. Filling it in instead
would
change the actual branch-protection rules of the repository, which is a
separate
decision.
* [ISSUE #137] chore: set the connector version to 3.0.0
Bump the reactor version from 1.0.0-SNAPSHOT to 3.0.0 via versions:set, so
the
parent and all four modules stay in sync. The external
flink-connector-parent
version is untouched.
---
.asf.yaml | 3 +-
docs/grpc-connector.md | 173 +++++++++++++++++++--
flink-connector-rocketmq-grpc/pom.xml | 2 +-
...MQLiteAckClient.java => RocketMQAckClient.java} | 32 ++--
.../grpc/ack/RocketMQAckProcessFunction.java | 16 +-
.../rocketmq/grpc/source/ConsumerMode.java | 41 +++++
.../rocketmq/grpc/source/RocketMQGrpcSource.java | 25 ++-
.../grpc/source/RocketMQGrpcSourceBuilder.java | 51 +++++-
.../grpc/source/RocketMQGrpcSourceOptions.java | 59 +++++--
.../grpc/source/reader/CheckpointAckTracker.java | 68 ++++++++
.../source/reader/LiteSimpleConsumerProvider.java | 58 -------
.../rocketmq/grpc/source/reader/PopConsumer.java | 48 ++++++
.../grpc/source/reader/PopConsumerProvider.java | 164 +++++++++++++++++++
.../source/reader/RocketMQGrpcSourceReader.java | 81 +++++++++-
.../reader/RocketMQGrpcSourceRecordEmitter.java | 16 ++
.../reader/RocketMQGrpcSourceSplitReader.java | 82 ++++++++--
.../grpc/table/RocketMQGrpcConnectorOptions.java | 41 ++++-
.../table/RocketMQGrpcDynamicTableFactory.java | 30 ++++
.../grpc/table/RocketMQGrpcDynamicTableSource.java | 15 +-
.../rocketmq/grpc/RocketMQGrpcITCase.java | 137 +++++++++++++++-
.../rocketmq/grpc/RocketMQGrpcLiteE2EVerify.java | 3 +
...kClientTest.java => RocketMQAckClientTest.java} | 40 ++---
.../grpc/source/RocketMQGrpcSourceBuilderTest.java | 63 ++++++++
.../source/reader/CheckpointAckTrackerTest.java | 94 +++++++++++
.../source/reader/PopConsumerProviderTest.java | 100 ++++++++++++
.../reader/RocketMQGrpcSourceSplitReaderTest.java | 127 +++++++++++++++
.../table/RocketMQGrpcDynamicTableFactoryTest.java | 32 ++++
flink-connector-rocketmq/pom.xml | 2 +-
flink-sql-connector-rocketmq-grpc/pom.xml | 2 +-
flink-sql-connector-rocketmq/pom.xml | 2 +-
pom.xml | 2 +-
31 files changed, 1430 insertions(+), 179 deletions(-)
diff --git a/.asf.yaml b/.asf.yaml
index a2d2d75..f2c834d 100644
--- a/.asf.yaml
+++ b/.asf.yaml
@@ -10,8 +10,7 @@ github:
issues: true
# Enable wiki
wiki: true
- protected_branches:
-
+
notifications:
commits: [email protected]
issues: [email protected]
diff --git a/docs/grpc-connector.md b/docs/grpc-connector.md
index 017b176..14ba709 100644
--- a/docs/grpc-connector.md
+++ b/docs/grpc-connector.md
@@ -1,8 +1,11 @@
-# gRPC Connector (LiteSimpleConsumer)
+# gRPC Connector
-The gRPC connector (`flink-connector-rocketmq-grpc`) targets RocketMQ 5.x
**lite topics** using
-`rocketmq-client-java` 5.2.1+. The source binds one main lite topic with a
wildcard subscription,
-never acks by itself, and hands the ack / throttle decision to downstream
operators.
+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,
+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
usage of the connector.
@@ -43,8 +46,21 @@ This leads to the two defining properties of the connector:
## 2. Consumption model
+**Two modes.** A single `rocketmq.source.mode` option selects how the source
subscribes and who
+acknowledges:
+
+- **SIMPLE** (default) — the source subscribes to one **normal** topic via a
`SimpleConsumer` with
+ an optional tag / SQL92 filter expression. Here the source itself
acknowledges messages: it
+ 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.
+
RocketMQ 5.x offers a push client (`PushConsumer`) and pull/Pop clients
-(`SimpleConsumer` / `LiteSimpleConsumer`). The connector uses
**`LiteSimpleConsumer`**
+(`SimpleConsumer` / `LiteSimpleConsumer`). Both connector modes build on the
Pop clients
(available since `rocketmq-client-java` 5.2.1), whose consumption is split
into three explicit
phases that map cleanly onto a Flink pipeline:
@@ -82,9 +98,12 @@ start. Per subtask:
```
subtask
└─ SourceReader (mailbox thread)
+ │ SIMPLE mode only: on notifyCheckpointComplete, ack the handles of that
+ │ checkpoint through the SplitReader's consumer (see below)
└─ SingleThreadFetcherManager → 1 fetcher thread
└─ SplitReader
- ├─ 1 shared LiteSimpleConsumer (thread-safe)
+ ├─ 1 shared Pop consumer (thread-safe)
+ │ SimpleConsumer in SIMPLE mode, LiteSimpleConsumer in LITE
mode
├─ N ReceiveWorker threads (N =
rocketmq.source.fetch-concurrency)
│ loop: consumer.receive(maxMessageNum, invisibleDuration)
│ → put into a bounded in-memory queue (capacity =
maxMessageNum × N)
@@ -96,6 +115,11 @@ subtask
flight — raising a single subtask's throughput without extra clientIds,
heartbeats, connections
or route caches. Total throughput scales with
`parallelism × fetch-concurrency × maxMessageNum / receive-latency`.
+- **SIMPLE-mode acks go through that same consumer.** The reader hands its
completed receipt
+ handles to `RocketMQGrpcSourceSplitReader#ack`, which issues the ack RPC on
the consumer that
+ received the messages. No second client, connection, clientId or credential
resolution is
+ involved, and no LITE-only client is used for normal topics. Pop consumers
are thread-safe, so
+ the mailbox thread may ack while the receive workers block in `receive()` on
the same instance.
- **Backpressure = stop receiving.** When the bounded queue is full, workers
block on `put()`;
the pull model has no explicit pause API and does not need one.
- **Receive failures back off exponentially** (100 ms doubling up to 30 s)
instead of
@@ -124,11 +148,14 @@ it on the source side and rebuilding a minimal message
view on the ack side requ
types. Confining that (experimental) coupling to one `@Internal` adapter keeps
the rest of the
connector SDK-clean.
-### 3.3 Downstream acknowledgement
+### 3.3 Downstream acknowledgement (LITE mode)
-Downstream operators obtain a shared, credential-free ack client:
+This section applies to **LITE mode only**: it is the path where the ack
decision is taken outside
+the source, so the ack RPC cannot reuse the receiving consumer. (SIMPLE mode
acks source-side
+through the consumer that received the messages — see §3.1.) Downstream
operators obtain a shared,
+credential-free ack client:
-- **Pooling.** `RocketMQLiteAckClient` is shared per TaskManager JVM via
reference-counted
+- **Pooling.** `RocketMQAckClient` is shared per TaskManager JVM via
reference-counted
`acquire`/`release` keyed by the client configuration. Internally it lazily
keeps one
same-group `LiteSimpleConsumer` per routing triple `(endpoint, namespace,
consumerGroup)`
carried by the incoming handles. The consumer built for a handle uses the
*handle's* endpoints
@@ -191,6 +218,10 @@ Alternatives that were evaluated and rejected:
| Carrying credentials in the record stream | Security red line. Credentials
stay in operator configuration, resolved locally per TaskManager. |
| Serializing the SDK protobuf receipt blob | Unnecessary — the server-side
handle string self-encodes routing; a handful of strings suffice. |
+(The rejection of source-side checkpoint ack applies to the LITE requirement,
where the ack decision
+is only known downstream. SIMPLE mode has no such decision and intentionally
adopts that very
+approach — see §5.)
+
Other decisions:
- **Commit point is user-driven.** The connector does not align acks with
checkpoints; the user
@@ -202,18 +233,66 @@ Other decisions:
## 5. Delivery semantics
-- **At-least-once.** The source never acks; un-acked messages are redelivered
after
- `invisible-duration` expires. Downstream must be idempotent.
+Both modes provide **at-least-once**: un-acked messages are redelivered by the
broker once their
+`invisible-duration` expires, so downstream processing must be idempotent.
What differs is *who*
+acks and *when*.
+
+### LITE mode
+
+- 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
shuffles and slow operators such as model inference); use the renewal policy
when queueing
time is unpredictable.
- Exactly-once is out of scope for the Pop model: there is no offset the
connector could
checkpoint-align, and acks are user-driven by design.
+### SIMPLE mode
+
+- Once a message is emitted, its receipt handle is registered with a
checkpoint-aligned tracker;
+ the handle is snapshotted together with the enclosing checkpoint and the
source batch-acks all
+ handles of a checkpoint from `notifyCheckpointComplete` after that
checkpoint completes.
+- The acks are issued **through the consumer that received the messages** (the
split reader's
+ `SimpleConsumer`), not through the pooled `RocketMQAckClient` of §3.3. The
reader therefore opens
+ no second gRPC connection, resolves no credentials of its own, and never
routes normal-topic acks
+ through a LITE client.
+- On a failure before completion the pending handles are simply dropped; the
un-acked messages are
+ redelivered by the broker after their `invisible-duration` expires. That
redelivery *is* the
+ at-least-once guarantee, which is why the tracker holds no
failure-survivable state.
+- **Checkpointing is required.** Without it `notifyCheckpointComplete` never
fires, nothing is ever
+ acked, and every message is redelivered forever.
+- **Size `invisible-duration` with headroom — do not compute it as a tight
bound.** The invisible
+ window is consumed by more than one checkpoint cycle:
+
+ | Consumer of the window | Why it counts |
+ | --- | --- |
+ | Queueing before emission | The window starts when the broker serves
`receive()`, not when the record is emitted. Time spent in the reader's
internal queue (`max-message-num` × `fetch-concurrency` slots) and in
deserialization is already gone. |
+ | Waiting for the next barrier | A record emitted just after a barrier waits
almost a full checkpoint interval for the next one. |
+ | Completing that checkpoint | Barrier alignment plus state materialisation,
bounded by the checkpoint timeout. |
+ | **A failed or timed-out checkpoint** | The tracker only clears handles for
checkpoints that actually complete, so handles of a failed checkpoint wait for
the next successful one — roughly **another full interval**, more if failures
repeat. |
+ | Draining the ack batch | `notifyCheckpointComplete` acks the batch one
handle per RPC, so the last handles of a large batch wait for the earlier ones.
|
+
+ A practical starting point is therefore
+
+ ```
+ invisible-duration ≥ 2 × checkpoint interval + checkpoint timeout + headroom
+ ```
+
+ Configuring it as merely `interval + timeout` still satisfies at-least-once,
but a single failed
+ checkpoint is then enough to make the batch visible again and reprocessed.
+- Emitted-but-not-yet-acked handles are **not renewed**. The SDK's
`changeInvisibleDuration` does
+ not return the new receipt handle, so the source cannot refresh a handle it
has already handed to
+ the tracker; sizing `invisible-duration` as above is the substitute. A
renewal policy (§7) only
+ covers the *first* row of the table above — messages still queued inside the
reader — because the
+ reader freezes a handle the moment the record is emitted.
+
---
## 6. Prerequisites
+The default SIMPLE mode only needs an ordinary RocketMQ 5.x topic plus a
consumer group. The
+following applies to the opt-in LITE mode:
+
- RocketMQ 5.x cluster with lite topic support enabled on the broker
(`enableMultiDispatch`, `enableLmq`, etc. — preset by the official Helm
chart).
- A LITE-type main topic and a consumer group with the wildcard attribute:
@@ -229,12 +308,16 @@ wildcard subscription receives nothing.
## 7. Source
+The source runs in **SIMPLE** mode unless another mode is selected, i.e. it
subscribes to a normal
+topic and acks the messages itself once the enclosing checkpoint completes
(checkpointing must be
+enabled):
+
```java
RocketMQGrpcSource<String> source = RocketMQGrpcSource.<String>builder()
.setEndpoints("127.0.0.1:8081")
- .setConsumerGroup("GID-lite")
- .setMainTopic("LiteMainTopic")
- .setFetchConcurrency(4)
+ .setConsumerGroup("GID-example")
+ .setTopic("normal-topic")
+ .setFilterExpression("tagA||tagB")
.setValueOnlyDeserializer(new SimpleStringSchema())
.build();
@@ -244,10 +327,31 @@ DataStream<AckableMessage<String>> stream =
The output type is `AckableMessage<OUT>`: the payload plus a serializable
`RocketMQReceiptHandle` (endpoint, namespace, consumerGroup, lite topic,
receipt handle). The
-handle can cross keyBy/shuffle boundaries; credentials never travel with it.
+handle can cross keyBy/shuffle boundaries; credentials never travel with it.
In SIMPLE mode the
+handle is acknowledged automatically by the source once the enclosing
checkpoint completes, so
+downstream neither needs nor should invoke the ack / throttle operators.
+
+### LITE mode
+
+LITE mode is opt-in: select it explicitly and bind a main lite topic instead
of a normal topic. The
+source then never acks by itself and the downstream operators take the ack /
throttle decision.
+
+```java
+RocketMQGrpcSource<String> source = RocketMQGrpcSource.<String>builder()
+ .setEndpoints("127.0.0.1:8081")
+ .setConsumerGroup("GID-lite")
+ .setMode(ConsumerMode.LITE)
+ .setMainTopic("LiteMainTopic")
+ .setFetchConcurrency(4)
+ .setValueOnlyDeserializer(new SimpleStringSchema())
+ .build();
+```
### Downstream ack / throttling
+The ack / throttle wirings below apply to LITE mode only; in SIMPLE mode the
source acks the
+messages itself.
+
Unacked messages are redelivered by the broker after `invisible-duration`
expires
(at-least-once — downstream must be idempotent). Two wirings are provided:
@@ -273,6 +377,12 @@
builder.setRenewalPolicyClass("com.example.MyRenewalPolicy")
.setRenewalAheadTime(Duration.ofSeconds(5));
```
+Renewal only covers records that are **still queued inside the reader**: the
reader freezes a
+handle the moment the record is emitted, because renewing would invalidate the
handle that has
+already travelled downstream. Consequently it does not extend the window for a
record that is
+waiting for its downstream ack (LITE) or for its checkpoint to complete
(SIMPLE) — in SIMPLE mode a
+renewal policy is therefore no substitute for sizing `invisible-duration` as
described in §5.
+
## 8. Sink
```java
@@ -301,11 +411,15 @@ RocketMQGrpcSink<String> sink =
RocketMQGrpcSink.<String>builder()
| Key | Type | Default | Description |
| --- | --- | --- | --- |
-| `rocketmq.source.main-topic` | String | (none) | Main lite topic to bind,
required |
+| `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.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` |
| `rocketmq.source.consumer-group` | String | (none) | Consumer group,
required |
| `rocketmq.source.fetch-concurrency` | Integer | 1 | Concurrent long-polling
receive workers |
| `rocketmq.source.await-duration` | Duration | 20s | Long-polling await time |
-| `rocketmq.source.invisible-duration` | Duration | 60s | Invisible time per
receive (min 10s) |
+| `rocketmq.source.invisible-duration` | Duration | 60s | Invisible time per
receive (min 10s); size it with headroom — see §5 |
| `rocketmq.source.max-message-num` | Integer | 32 | Max messages per receive |
| `rocketmq.source.renewal-policy-class` | String | (none) |
`InvisibleDurationRenewalPolicy` implementation |
| `rocketmq.source.renewal-ahead-time` | Duration | 5s | Renew this long
before invisible expiry |
@@ -325,9 +439,34 @@ The SQL connector identifier is `rocketmq-grpc` (fat-jar
module
`rocketmq.source.fetch-concurrency`. The SQL/Table path consumes message
values only; the
downstream ack / throttling APIs are DataStream-only.
+The SQL source runs in `simple` mode by default (`source.mode` defaults to
`simple`), i.e. the
+configured `topic` is consumed as a normal topic and the source acks on
checkpoint completion, so
+checkpointing must be enabled:
+
+```sql
+CREATE TABLE orders (
+ f0 STRING
+) WITH (
+ 'connector' = 'rocketmq-grpc',
+ 'endpoints' = '127.0.0.1:8081',
+ 'topic' = 'normal-topic',
+ 'source.consumer-group' = 'GID-example',
+ 'source.filter-expression' = 'tagA || tagB',
+ 'format' = 'json'
+);
+```
+
+To consume a main lite topic instead, opt into `lite` mode explicitly with
+`'source.mode' = 'lite'`; the `topic` option is then bound as the main lite
topic and the SQL path
+performs no acknowledgement (the ack / throttling APIs are DataStream-only).
+
## 11. Known limitations
- `LiteSimpleConsumer` is experimental in the SDK (5.2.1+) and offers
synchronous APIs only.
+- SIMPLE mode requires checkpointing; without checkpoints messages are never
acknowledged and are
+ redelivered indefinitely.
+- SIMPLE mode has no way to renew a handle it has already emitted, so an
`invisible-duration` sized
+ as a tight bound turns any failed checkpoint into reprocessed messages (see
§5).
- One main lite topic per source; no startup-offset control in
wildcard-subscription mode
(`OffsetOption` is limited to exact `subscribeLite`).
- Throttling defers messages per message (extra RPC per defer, reordering,
`deliveryAttempt`
diff --git a/flink-connector-rocketmq-grpc/pom.xml
b/flink-connector-rocketmq-grpc/pom.xml
index 512ddce..d035375 100644
--- a/flink-connector-rocketmq-grpc/pom.xml
+++ b/flink-connector-rocketmq-grpc/pom.xml
@@ -26,7 +26,7 @@ under the License.
<parent>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-rocketmq-parent</artifactId>
- <version>1.0.0-SNAPSHOT</version>
+ <version>3.0.0</version>
</parent>
<artifactId>flink-connector-rocketmq-grpc</artifactId>
diff --git
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/ack/RocketMQLiteAckClient.java
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/ack/RocketMQAckClient.java
similarity index 92%
rename from
flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/ack/RocketMQLiteAckClient.java
rename to
flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/ack/RocketMQAckClient.java
index f6754c4..0ffe20c 100644
---
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/ack/RocketMQLiteAckClient.java
+++
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/ack/RocketMQAckClient.java
@@ -47,10 +47,12 @@ import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
/**
- * A credential-free acknowledgement client that a downstream operator uses to
acknowledge or
- * re-schedule RocketMQ Pop messages described by a {@link
RocketMQReceiptHandle}. The credentials
- * used to talk to the RocketMQ proxy are configured on the operator that owns
the client and are
- * <b>never</b> carried in the data stream.
+ * A credential-free acknowledgement client that acknowledges RocketMQ Pop
messages of both lite and
+ * normal topics. A downstream operator uses it to acknowledge or re-schedule
the messages described
+ * by a {@link RocketMQReceiptHandle}, which is what the LITE mode of the
source relies on: the ack
+ * decision is only known after downstream processing, possibly on the other
side of a shuffle. The
+ * credentials used to talk to the RocketMQ proxy are configured on the
operator that owns the
+ * client and are <b>never</b> carried in the data stream.
*
* <p>The client keeps a lazily populated pool of same-group {@link
LiteSimpleConsumer}s, one per
* {@link ConsumerKey routing triple} carried by the incoming handles, and
issues the {@code ack} /
@@ -59,16 +61,18 @@ import java.util.concurrent.ConcurrentHashMap;
* come from the handle (so the ack RPC is routed to the right proxy and
resource) and whose
* credentials/TLS/timeout come from the operator {@link Configuration}. This
is required because
* the SDK stamps the ack request with the <em>consumer's</em> namespace, so a
pooled consumer must
- * share the handle's namespace.
+ * share the handle's namespace. The pooled consumers are built through {@code
LiteSimpleConsumer},
+ * but the ack RPC itself is protocol-level identical for lite and normal
topic handles, so handles
+ * of normal topics are acknowledged without any semantic difference.
*
* <p>Instances are shared per TaskManager JVM: {@link
#acquire(Configuration)} reference-counts a
* client per distinct client configuration and {@link
#release(Configuration)} closes it once no
* operator instance references it anymore, so callers must not close a client
directly.
*/
@Internal
-public final class RocketMQLiteAckClient {
+public final class RocketMQAckClient {
- private static final Logger LOG =
LoggerFactory.getLogger(RocketMQLiteAckClient.class);
+ private static final Logger LOG =
LoggerFactory.getLogger(RocketMQAckClient.class);
/**
* The await duration is only consulted by {@code receive}, which this
ack-only client never
@@ -90,7 +94,7 @@ public final class RocketMQLiteAckClient {
private volatile boolean closed = false;
- private RocketMQLiteAckClient(Configuration configuration) {
+ private RocketMQAckClient(Configuration configuration) {
this.configuration =
Objects.requireNonNull(configuration, "configuration should
not be null");
this.credentialsResolver =
CredentialsResolvers.createFromConfiguration(configuration);
@@ -103,12 +107,12 @@ public final class RocketMQLiteAckClient {
* @param configuration the operator configuration carrying the RocketMQ
client options.
* @return the shared ack client; callers must not close it directly.
*/
- public static synchronized RocketMQLiteAckClient acquire(Configuration
configuration) {
+ public static synchronized RocketMQAckClient acquire(Configuration
configuration) {
Objects.requireNonNull(configuration, "configuration should not be
null");
final String key = keyOf(configuration);
RefCounted ref = CLIENTS.get(key);
if (ref == null) {
- ref = new RefCounted(new RocketMQLiteAckClient(configuration));
+ ref = new RefCounted(new RocketMQAckClient(configuration));
CLIENTS.put(key, ref);
}
ref.count++;
@@ -128,8 +132,8 @@ public final class RocketMQLiteAckClient {
public static void release(Configuration configuration) {
Objects.requireNonNull(configuration, "configuration should not be
null");
final String key = keyOf(configuration);
- RocketMQLiteAckClient toClose = null;
- synchronized (RocketMQLiteAckClient.class) {
+ RocketMQAckClient toClose = null;
+ synchronized (RocketMQAckClient.class) {
final RefCounted ref = CLIENTS.get(key);
if (ref == null) {
return;
@@ -337,10 +341,10 @@ public final class RocketMQLiteAckClient {
}
private static final class RefCounted {
- private final RocketMQLiteAckClient client;
+ private final RocketMQAckClient client;
private int count;
- private RefCounted(RocketMQLiteAckClient client) {
+ private RefCounted(RocketMQAckClient client) {
this.client = client;
}
}
diff --git
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/ack/RocketMQAckProcessFunction.java
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/ack/RocketMQAckProcessFunction.java
index 29330af..fbe215b 100644
---
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/ack/RocketMQAckProcessFunction.java
+++
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/ack/RocketMQAckProcessFunction.java
@@ -29,13 +29,13 @@ import java.util.Objects;
/**
* An abstract {@link ProcessFunction} that gives subclasses a ready-to-use
{@link
- * RocketMQLiteAckClient} so they can acknowledge or re-schedule RocketMQ Pop
messages directly from
+ * RocketMQAckClient} so they can acknowledge or re-schedule RocketMQ Pop
messages directly from
* their own business logic via {@link #ack} and {@link
#changeInvisibleDuration}.
*
- * <p>The client is shared per TaskManager through {@link
RocketMQLiteAckClient#acquire}: it is
- * acquired in {@link #open(OpenContext)} and released in {@link #close()}.
The RocketMQ client
- * options (endpoints, namespace, credentials, TLS, timeout) are provided
through the {@link
- * Configuration} passed to the constructor and never travel in the data
stream.
+ * <p>The client is shared per TaskManager through {@link
RocketMQAckClient#acquire}: it is acquired
+ * in {@link #open(OpenContext)} and released in {@link #close()}. The
RocketMQ client options
+ * (endpoints, namespace, credentials, TLS, timeout) are provided through the
{@link Configuration}
+ * passed to the constructor and never travel in the data stream.
*
* @param <IN> the input record type, typically {@code AckableMessage<T>}.
* @param <OUT> the output record type.
@@ -47,7 +47,7 @@ public abstract class RocketMQAckProcessFunction<IN, OUT>
extends ProcessFunctio
private final Configuration configuration;
- private transient RocketMQLiteAckClient ackClient;
+ private transient RocketMQAckClient ackClient;
private transient boolean acquired;
private transient Counter numAcksSucceeded;
private transient Counter numAcksFailed;
@@ -63,7 +63,7 @@ public abstract class RocketMQAckProcessFunction<IN, OUT>
extends ProcessFunctio
@Override
public void open(OpenContext openContext) throws Exception {
super.open(openContext);
- this.ackClient = RocketMQLiteAckClient.acquire(configuration);
+ this.ackClient = RocketMQAckClient.acquire(configuration);
this.acquired = true;
this.numAcksSucceeded =
getRuntimeContext().getMetricGroup().counter("numAcksSucceeded");
this.numAcksFailed =
getRuntimeContext().getMetricGroup().counter("numAcksFailed");
@@ -80,7 +80,7 @@ public abstract class RocketMQAckProcessFunction<IN, OUT>
extends ProcessFunctio
// Only release when open() actually acquired the client; Flink calls
close() even after a
// failed open(), and a stray release would decrement another
operator's reference.
if (acquired) {
- RocketMQLiteAckClient.release(configuration);
+ RocketMQAckClient.release(configuration);
acquired = false;
}
this.ackClient = null;
diff --git
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/ConsumerMode.java
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/ConsumerMode.java
new file mode 100644
index 0000000..63dcc5a
--- /dev/null
+++
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/ConsumerMode.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.rocketmq.grpc.source;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+/** The consumption mode of the RocketMQ gRPC source. */
+@PublicEvolving
+public enum ConsumerMode {
+
+ /**
+ * Bind a main lite topic via a {@code LiteSimpleConsumer}; the broker
delivers messages of all
+ * sub topics over one receive stream. Messages are acknowledged by a
downstream operator using
+ * the receipt handle carried by each emitted record.
+ */
+ LITE,
+
+ /**
+ * Subscribe to a normal topic via a {@code SimpleConsumer} with an
optional filter expression.
+ * Messages are acknowledged by the source itself once the enclosing
checkpoint completes
+ * (at-least-once); checkpointing must be enabled and the invisible
duration must exceed the
+ * checkpoint interval plus timeout.
+ */
+ SIMPLE
+}
diff --git
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/RocketMQGrpcSource.java
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/RocketMQGrpcSource.java
index 1f21b86..a3e48d2 100644
---
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/RocketMQGrpcSource.java
+++
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/RocketMQGrpcSource.java
@@ -39,6 +39,7 @@ import
org.apache.flink.connector.rocketmq.grpc.source.deserialization.RocketMQG
import
org.apache.flink.connector.rocketmq.grpc.source.enumerator.RocketMQGrpcSourceEnumState;
import
org.apache.flink.connector.rocketmq.grpc.source.enumerator.RocketMQGrpcSourceEnumStateSerializer;
import
org.apache.flink.connector.rocketmq.grpc.source.enumerator.RocketMQGrpcSourceEnumerator;
+import
org.apache.flink.connector.rocketmq.grpc.source.reader.CheckpointAckTracker;
import org.apache.flink.connector.rocketmq.grpc.source.reader.MessageViewImpl;
import
org.apache.flink.connector.rocketmq.grpc.source.reader.RocketMQGrpcSourceFetcherManager;
import
org.apache.flink.connector.rocketmq.grpc.source.reader.RocketMQGrpcSourceReader;
@@ -116,17 +117,27 @@ public class RocketMQGrpcSource<OUT>
final RocketMQGrpcSourceFetcherManager fetcherManager =
new RocketMQGrpcSourceFetcherManager(elementsQueue,
splitReaderSupplier);
+ final ConsumerMode mode =
configuration.get(RocketMQGrpcSourceOptions.MODE);
+ final CheckpointAckTracker ackTracker =
+ mode == ConsumerMode.SIMPLE ? new CheckpointAckTracker() :
null;
+
final RocketMQGrpcSourceRecordEmitter<OUT> recordEmitter =
new RocketMQGrpcSourceRecordEmitter<>(
deserializationSchema,
configuration.get(RocketMQGrpcOptions.NAMESPACE),
-
configuration.get(RocketMQGrpcSourceOptions.CONSUMER_GROUP));
-
- final RocketMQGrpcSourceReader<OUT> reader =
- new RocketMQGrpcSourceReader<>(
- elementsQueue, fetcherManager, recordEmitter,
configuration, readerContext);
-
- return reader;
+
configuration.get(RocketMQGrpcSourceOptions.CONSUMER_GROUP),
+ ackTracker == null ? null : ackTracker::add);
+
+ // The same split reader instance is handed to the fetcher manager and
to the source reader:
+ // in SIMPLE mode the reader acks through the very consumer that
received the messages.
+ return new RocketMQGrpcSourceReader<>(
+ elementsQueue,
+ fetcherManager,
+ recordEmitter,
+ configuration,
+ readerContext,
+ ackTracker,
+ splitReader);
}
@Override
diff --git
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/RocketMQGrpcSourceBuilder.java
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/RocketMQGrpcSourceBuilder.java
index f38dcf8..ebc0861 100644
---
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/RocketMQGrpcSourceBuilder.java
+++
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/RocketMQGrpcSourceBuilder.java
@@ -37,6 +37,8 @@ public class RocketMQGrpcSourceBuilder<OUT> {
private final Configuration configuration;
private String mainTopic;
+ private ConsumerMode mode = ConsumerMode.SIMPLE;
+ private String topic;
private Boundedness boundedness;
private RocketMQGrpcDeserializationSchema<OUT> deserializationSchema;
@@ -55,7 +57,11 @@ public class RocketMQGrpcSourceBuilder<OUT> {
return setConfig(RocketMQGrpcSourceOptions.CONSUMER_GROUP,
consumerGroup);
}
- /** Set the main lite topic bound by the LiteSimpleConsumer. Every subtask
binds this topic. */
+ /**
+ * Set the main lite topic bound by the LiteSimpleConsumer. Every subtask
binds this topic.
+ * Required in {@link ConsumerMode#LITE}, which must be selected
explicitly via {@link
+ * #setMode(ConsumerMode)}; ignored in {@link ConsumerMode#SIMPLE}.
+ */
public RocketMQGrpcSourceBuilder<OUT> setMainTopic(String mainTopic) {
checkArgument(
mainTopic != null && !mainTopic.trim().isEmpty(),
@@ -64,6 +70,32 @@ public class RocketMQGrpcSourceBuilder<OUT> {
return this;
}
+ /** Set the consumption mode. Defaults to {@link ConsumerMode#SIMPLE}. */
+ public RocketMQGrpcSourceBuilder<OUT> setMode(ConsumerMode mode) {
+ this.mode = checkNotNull(mode);
+ return this;
+ }
+
+ /**
+ * Set the normal topic subscribed by the SimpleConsumer. Required in the
default {@link
+ * ConsumerMode#SIMPLE}; ignored in {@link ConsumerMode#LITE}.
+ */
+ public RocketMQGrpcSourceBuilder<OUT> setTopic(String topic) {
+ checkArgument(topic != null && !topic.trim().isEmpty(), "topic must
not be null or blank");
+ this.topic = topic;
+ return this;
+ }
+
+ /** Set the SIMPLE mode filter expression, e.g. {@code "tagA||tagB"} or a
SQL92 expression. */
+ public RocketMQGrpcSourceBuilder<OUT> setFilterExpression(String
filterExpression) {
+ return setConfig(RocketMQGrpcSourceOptions.FILTER_EXPRESSION,
filterExpression);
+ }
+
+ /** Set the SIMPLE mode filter expression type: {@code TAG} (default) or
{@code SQL92}. */
+ public RocketMQGrpcSourceBuilder<OUT> setFilterType(String filterType) {
+ return setConfig(RocketMQGrpcSourceOptions.FILTER_TYPE, filterType);
+ }
+
/**
* Set the fetch concurrency of each subtask, i.e. the number of
concurrent fetch requests it
* issues. The requests are served by worker threads sharing a single
thread-safe {@code
@@ -131,14 +163,23 @@ public class RocketMQGrpcSourceBuilder<OUT> {
checkNotNull(
configuration.get(RocketMQGrpcSourceOptions.CONSUMER_GROUP),
"consumer group must be configured");
- checkArgument(
- mainTopic != null && !mainTopic.trim().isEmpty(),
- "the main topic must be configured");
+ checkNotNull(mode, "the consumer mode must be configured");
+ if (mode == ConsumerMode.LITE) {
+ checkArgument(
+ mainTopic != null && !mainTopic.trim().isEmpty(),
+ "the main topic must be configured");
+ configuration.set(RocketMQGrpcSourceOptions.MAIN_TOPIC, mainTopic);
+ } else {
+ checkArgument(
+ topic != null && !topic.trim().isEmpty(),
+ "the topic must be configured in SIMPLE mode");
+ configuration.set(RocketMQGrpcSourceOptions.TOPIC, topic);
+ }
checkArgument(
configuration.get(RocketMQGrpcSourceOptions.FETCH_CONCURRENCY)
>= 1,
"fetch concurrency must be at least 1");
checkNotNull(deserializationSchema, "deserializer must be configured");
- configuration.set(RocketMQGrpcSourceOptions.MAIN_TOPIC, mainTopic);
+ configuration.set(RocketMQGrpcSourceOptions.MODE, mode);
return new RocketMQGrpcSource<>(configuration, boundedness,
deserializationSchema);
}
}
diff --git
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/RocketMQGrpcSourceOptions.java
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/RocketMQGrpcSourceOptions.java
index 94b1e5a..f54d473 100644
---
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/RocketMQGrpcSourceOptions.java
+++
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/RocketMQGrpcSourceOptions.java
@@ -47,14 +47,52 @@ public class RocketMQGrpcSourceOptions {
.noDefaultValue()
.withDescription("The consumer group of the
SimpleConsumer.");
+ public static final ConfigOption<ConsumerMode> MODE =
+ ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "mode")
+ .enumType(ConsumerMode.class)
+ .defaultValue(ConsumerMode.SIMPLE)
+ .withDescription(
+ "The consumption mode. SIMPLE (the default)
subscribes to a normal "
+ + "topic via a SimpleConsumer and the
source acknowledges "
+ + "messages when their checkpoint
completes; LITE binds a main "
+ + "lite topic via a LiteSimpleConsumer and
defers "
+ + "acknowledgement to a downstream
operator.");
+
+ public static final ConfigOption<String> TOPIC =
+ ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "topic")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The normal topic subscribed by the
SimpleConsumer. Required in "
+ + "SIMPLE mode; ignored in LITE mode.");
+
+ public static final ConfigOption<String> FILTER_EXPRESSION =
+ ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "filter-expression")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The filter expression applied to the SIMPLE mode
subscription, e.g. "
+ + "a tag expression 'tagA||tagB' or a
SQL92 expression. When "
+ + "absent, all messages are received. Only
effective in "
+ + "SIMPLE mode.");
+
+ public static final ConfigOption<String> FILTER_TYPE =
+ ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "filter-type")
+ .stringType()
+ .defaultValue("TAG")
+ .withDescription(
+ "The type of 'filter-expression': TAG or SQL92.
Only effective in "
+ + "SIMPLE mode.");
+
public static final ConfigOption<String> MAIN_TOPIC =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "main-topic")
.stringType()
.noDefaultValue()
.withDescription(
- "The main lite topic bound by the
LiteSimpleConsumer. Every subtask "
- + "binds this topic; the broker performs
message-level load "
- + "balancing across the consumer group.");
+ "The main lite topic bound by the
LiteSimpleConsumer. Required in "
+ + "LITE mode; ignored in SIMPLE mode.
Every subtask binds this "
+ + "topic; the broker performs
message-level load balancing "
+ + "across the consumer group.");
public static final ConfigOption<Integer> FETCH_CONCURRENCY =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "fetch-concurrency")
@@ -79,12 +117,15 @@ public class RocketMQGrpcSourceOptions {
.durationType()
.defaultValue(Duration.ofSeconds(60))
.withDescription(
- "The invisible duration of a received message.
Because messages are "
- + "acknowledged by a downstream operator,
it must be larger "
- + "than the full downstream processing
time of a message so "
- + "that a message is not prematurely
redelivered while still "
- + "in flight; un-acked messages are
redelivered after this "
- + "duration, which provides the
at-least-once guarantee.");
+ "The invisible duration of a received message;
un-acked messages are "
+ + "redelivered after it expires, which
provides the "
+ + "at-least-once guarantee. In LITE mode
it must be larger than "
+ + "the full downstream processing time of
a message, because a "
+ + "downstream operator acknowledges it. In
SIMPLE mode the "
+ + "source acknowledges on checkpoint
completion, so size it "
+ + "with headroom above two checkpoint
intervals plus the "
+ + "checkpoint timeout: a failed checkpoint
defers the ack to "
+ + "the next successful one.");
public static final ConfigOption<Integer> MAX_MESSAGE_NUM =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "max-message-num")
diff --git
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/reader/CheckpointAckTracker.java
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/reader/CheckpointAckTracker.java
new file mode 100644
index 0000000..d699de0
--- /dev/null
+++
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/reader/CheckpointAckTracker.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.rocketmq.grpc.source.reader;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.rocketmq.grpc.ack.RocketMQReceiptHandle;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.TreeMap;
+
+/**
+ * Tracks the receipt handles of emitted-but-unacknowledged messages of the
SIMPLE mode, grouped by
+ * the checkpoint that observed them. Handles snapshotted with checkpoint N
are acknowledged once
+ * checkpoint N completes; a failure before completion simply leaves the
messages un-acked so the
+ * broker redelivers them after their invisible duration (at-least-once).
+ *
+ * <p>The tracker holds no state that must survive failures: losing pending
handles only causes
+ * redelivery, never loss. All methods are synchronized because the record
emitter (reader thread)
+ * and the checkpoint callbacks (mailbox thread) may touch it concurrently.
+ */
+@Internal
+public class CheckpointAckTracker {
+
+ private final List<RocketMQReceiptHandle> pending = new ArrayList<>();
+ private final TreeMap<Long, List<RocketMQReceiptHandle>> snapshotted = new
TreeMap<>();
+
+ /** Track the handle of an emitted message. */
+ public synchronized void add(RocketMQReceiptHandle handle) {
+ pending.add(handle);
+ }
+
+ /** Associate all handles emitted so far with the given checkpoint. */
+ public synchronized void snapshot(long checkpointId) {
+ if (!pending.isEmpty()) {
+ snapshotted.put(checkpointId, new ArrayList<>(pending));
+ pending.clear();
+ }
+ }
+
+ /**
+ * Return every handle whose checkpoint id is at most the completed one,
removing them from the
+ * tracker.
+ */
+ public synchronized List<RocketMQReceiptHandle> completeUpTo(long
checkpointId) {
+ final List<RocketMQReceiptHandle> completed = new ArrayList<>();
+ while (!snapshotted.isEmpty() && snapshotted.firstKey() <=
checkpointId) {
+ completed.addAll(snapshotted.pollFirstEntry().getValue());
+ }
+ return completed;
+ }
+}
diff --git
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/reader/LiteSimpleConsumerProvider.java
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/reader/LiteSimpleConsumerProvider.java
deleted file mode 100644
index 82b0c8d..0000000
---
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/reader/LiteSimpleConsumerProvider.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.connector.rocketmq.grpc.source.reader;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.configuration.Configuration;
-import
org.apache.flink.connector.rocketmq.grpc.common.ClientConfigurationProvider;
-import
org.apache.flink.connector.rocketmq.grpc.source.RocketMQGrpcSourceOptions;
-
-import org.apache.rocketmq.client.apis.ClientException;
-import org.apache.rocketmq.client.apis.ClientServiceProvider;
-import org.apache.rocketmq.client.apis.consumer.LiteSimpleConsumer;
-
-/**
- * A factory that builds a gRPC {@link LiteSimpleConsumer} (Pop model) from a
Flink configuration.
- *
- * <p>The consumer binds the configured main topic and relies on its wildcard
(generalized)
- * subscription to receive messages from all of its sub topics; the consumer
group must carry the
- * {@code lite.sub.wildcard=true} attribute on the broker.
- */
-@Internal
-public class LiteSimpleConsumerProvider {
-
- private LiteSimpleConsumerProvider() {}
-
- /**
- * Build and start a {@link LiteSimpleConsumer} from the given Flink
configuration.
- *
- * @param configuration the Flink configuration.
- * @return a started {@link LiteSimpleConsumer}.
- */
- public static LiteSimpleConsumer create(Configuration configuration)
throws ClientException {
- final ClientServiceProvider provider =
ClientServiceProvider.loadService();
- return provider.newLiteSimpleConsumerBuilder()
- .setClientConfiguration(
-
ClientConfigurationProvider.getClientConfiguration(configuration))
-
.setConsumerGroup(configuration.get(RocketMQGrpcSourceOptions.CONSUMER_GROUP))
-
.setAwaitDuration(configuration.get(RocketMQGrpcSourceOptions.AWAIT_DURATION))
-
.bindTopic(configuration.get(RocketMQGrpcSourceOptions.MAIN_TOPIC))
- .build();
- }
-}
diff --git
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/reader/PopConsumer.java
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/reader/PopConsumer.java
new file mode 100644
index 0000000..a770aa8
--- /dev/null
+++
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/reader/PopConsumer.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.rocketmq.grpc.source.reader;
+
+import org.apache.flink.annotation.Internal;
+
+import org.apache.rocketmq.client.apis.ClientException;
+import org.apache.rocketmq.client.apis.message.MessageView;
+
+import java.io.Closeable;
+import java.time.Duration;
+import java.util.List;
+
+/**
+ * The minimal consumer surface the split reader needs. Both SDK consumers
({@code
+ * LiteSimpleConsumer} and {@code SimpleConsumer}) expose the same Pop-model
receive / ack /
+ * change-invisible-duration operations but share no common interface, so the
connector adapts them
+ * to this single type.
+ */
+@Internal
+public interface PopConsumer extends Closeable {
+
+ /** Blocking long-polling receive. */
+ List<MessageView> receive(int maxMessageNum, Duration invisibleDuration)
throws ClientException;
+
+ /** Acknowledge a received message, removing it from the Pop invisible
set. */
+ void ack(MessageView messageView) throws ClientException;
+
+ /** Extend the invisible duration of a still-buffered message. */
+ void changeInvisibleDuration(MessageView messageView, Duration
invisibleDuration)
+ throws ClientException;
+}
diff --git
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/reader/PopConsumerProvider.java
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/reader/PopConsumerProvider.java
new file mode 100644
index 0000000..7111912
--- /dev/null
+++
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/reader/PopConsumerProvider.java
@@ -0,0 +1,164 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.rocketmq.grpc.source.reader;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.configuration.Configuration;
+import
org.apache.flink.connector.rocketmq.grpc.common.ClientConfigurationProvider;
+import org.apache.flink.connector.rocketmq.grpc.common.FilterExpressionParser;
+import org.apache.flink.connector.rocketmq.grpc.source.ConsumerMode;
+import
org.apache.flink.connector.rocketmq.grpc.source.RocketMQGrpcSourceOptions;
+
+import org.apache.rocketmq.client.apis.ClientException;
+import org.apache.rocketmq.client.apis.ClientServiceProvider;
+import org.apache.rocketmq.client.apis.consumer.FilterExpression;
+import org.apache.rocketmq.client.apis.consumer.LiteSimpleConsumer;
+import org.apache.rocketmq.client.apis.consumer.SimpleConsumer;
+import org.apache.rocketmq.client.apis.message.MessageView;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * A factory that builds a gRPC Pop consumer from a Flink configuration: a
{@code
+ * LiteSimpleConsumer} bound to the main lite topic in {@link
ConsumerMode#LITE}, or a {@code
+ * SimpleConsumer} subscribed to a normal topic with a filter expression in
{@link
+ * ConsumerMode#SIMPLE}.
+ */
+@Internal
+public class PopConsumerProvider {
+
+ private PopConsumerProvider() {}
+
+ /**
+ * Build and start the {@link PopConsumer} selected by {@link
RocketMQGrpcSourceOptions#MODE}.
+ */
+ public static PopConsumer create(Configuration configuration) throws
ClientException {
+ final ClientServiceProvider provider =
ClientServiceProvider.loadService();
+ if (configuration.get(RocketMQGrpcSourceOptions.MODE) ==
ConsumerMode.SIMPLE) {
+ final FilterExpression filter =
+ FilterExpressionParser.parse(
+
configuration.get(RocketMQGrpcSourceOptions.FILTER_EXPRESSION),
+
configuration.get(RocketMQGrpcSourceOptions.FILTER_TYPE));
+ final SimpleConsumer consumer =
+ provider.newSimpleConsumerBuilder()
+ .setClientConfiguration(
+
ClientConfigurationProvider.getClientConfiguration(
+ configuration))
+ .setConsumerGroup(
+
configuration.get(RocketMQGrpcSourceOptions.CONSUMER_GROUP))
+ .setAwaitDuration(
+
configuration.get(RocketMQGrpcSourceOptions.AWAIT_DURATION))
+ .setSubscriptionExpressions(
+ Collections.singletonMap(
+
configuration.get(RocketMQGrpcSourceOptions.TOPIC),
+ filter))
+ .build();
+ return new SimplePopConsumer(consumer);
+ }
+ final LiteSimpleConsumer consumer =
+ provider.newLiteSimpleConsumerBuilder()
+ .setClientConfiguration(
+
ClientConfigurationProvider.getClientConfiguration(configuration))
+ .setConsumerGroup(
+
configuration.get(RocketMQGrpcSourceOptions.CONSUMER_GROUP))
+ .setAwaitDuration(
+
configuration.get(RocketMQGrpcSourceOptions.AWAIT_DURATION))
+
.bindTopic(configuration.get(RocketMQGrpcSourceOptions.MAIN_TOPIC))
+ .build();
+ return new LitePopConsumer(consumer);
+ }
+
+ /** Wrap an SDK {@code LiteSimpleConsumer} into the connector's {@link
PopConsumer} surface. */
+ @VisibleForTesting
+ static PopConsumer wrap(LiteSimpleConsumer consumer) {
+ return new LitePopConsumer(consumer);
+ }
+
+ /** Wrap an SDK {@code SimpleConsumer} into the connector's {@link
PopConsumer} surface. */
+ @VisibleForTesting
+ static PopConsumer wrap(SimpleConsumer consumer) {
+ return new SimplePopConsumer(consumer);
+ }
+
+ private static final class LitePopConsumer implements PopConsumer {
+
+ private final LiteSimpleConsumer delegate;
+
+ private LitePopConsumer(LiteSimpleConsumer delegate) {
+ this.delegate = delegate;
+ }
+
+ @Override
+ public List<MessageView> receive(int maxMessageNum, Duration
invisibleDuration)
+ throws ClientException {
+ return delegate.receive(maxMessageNum, invisibleDuration);
+ }
+
+ @Override
+ public void ack(MessageView messageView) throws ClientException {
+ delegate.ack(messageView);
+ }
+
+ @Override
+ public void changeInvisibleDuration(MessageView messageView, Duration
invisibleDuration)
+ throws ClientException {
+ delegate.changeInvisibleDuration(messageView, invisibleDuration);
+ }
+
+ @Override
+ public void close() throws IOException {
+ delegate.close();
+ }
+ }
+
+ private static final class SimplePopConsumer implements PopConsumer {
+
+ private final SimpleConsumer delegate;
+
+ private SimplePopConsumer(SimpleConsumer delegate) {
+ this.delegate = delegate;
+ }
+
+ @Override
+ public List<MessageView> receive(int maxMessageNum, Duration
invisibleDuration)
+ throws ClientException {
+ return delegate.receive(maxMessageNum, invisibleDuration);
+ }
+
+ @Override
+ public void ack(MessageView messageView) throws ClientException {
+ delegate.ack(messageView);
+ }
+
+ @Override
+ public void changeInvisibleDuration(MessageView messageView, Duration
invisibleDuration)
+ throws ClientException {
+ delegate.changeInvisibleDuration(messageView, invisibleDuration);
+ }
+
+ @Override
+ public void close() throws IOException {
+ delegate.close();
+ }
+ }
+}
diff --git
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/reader/RocketMQGrpcSourceReader.java
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/reader/RocketMQGrpcSourceReader.java
index 63fd328..b9c560a 100644
---
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/reader/RocketMQGrpcSourceReader.java
+++
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/reader/RocketMQGrpcSourceReader.java
@@ -26,17 +26,35 @@ import
org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
import
org.apache.flink.connector.base.source.reader.SingleThreadMultiplexSourceReaderBase;
import
org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
import org.apache.flink.connector.rocketmq.grpc.ack.AckableMessage;
+import org.apache.flink.connector.rocketmq.grpc.ack.RocketMQReceiptHandle;
import
org.apache.flink.connector.rocketmq.grpc.source.split.RocketMQGrpcSourceSplit;
import
org.apache.flink.connector.rocketmq.grpc.source.split.RocketMQGrpcSourceSplitState;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.List;
import java.util.Map;
/**
* The source reader for the RocketMQ gRPC connector. It emits {@link
AckableMessage} records and
- * does not acknowledge messages itself; acknowledgement is performed by a
downstream operator using
- * the self-contained receipt handle carried by each emitted record. Un-acked
messages are
- * redelivered by the broker after their invisible duration expires, providing
the at-least-once
- * guarantee.
+ * its acknowledgement behaviour depends on the consumer mode:
+ *
+ * <ul>
+ * <li>LITE mode: the reader does not acknowledge messages itself;
acknowledgement is performed by
+ * a downstream operator using the self-contained receipt handle carried
by each emitted
+ * record.
+ * <li>SIMPLE mode: the reader acknowledges the emitted messages itself once
the checkpoint that
+ * observed them completes, issuing the acks through the very {@link
+ * RocketMQGrpcSourceSplitReader} consumer that received them — no
separate ack client and no
+ * second connection are involved. This requires checkpointing to be
enabled and the invisible
+ * duration to be larger than the checkpoint interval plus the
checkpoint timeout.
+ * </ul>
+ *
+ * <p>In both modes un-acked messages are redelivered by the broker after
their invisible duration
+ * expires, providing the at-least-once guarantee.
*/
@Internal
public class RocketMQGrpcSourceReader<T>
@@ -46,13 +64,68 @@ public class RocketMQGrpcSourceReader<T>
RocketMQGrpcSourceSplit,
RocketMQGrpcSourceSplitState> {
+ private static final Logger LOG =
LoggerFactory.getLogger(RocketMQGrpcSourceReader.class);
+
+ @Nullable private final CheckpointAckTracker ackTracker;
+
+ /**
+ * The split reader that received the messages; only consulted in SIMPLE
mode, i.e. when {@link
+ * #ackTracker} is non-null. It is not closed here: the fetcher manager
owns its lifecycle and
+ * shuts it down as part of the base reader close.
+ */
+ @Nullable private final RocketMQGrpcSourceSplitReader splitReader;
+
public RocketMQGrpcSourceReader(
FutureCompletingBlockingQueue<RecordsWithSplitIds<MessageViewImpl>>
elementsQueue,
RocketMQGrpcSourceFetcherManager fetcherManager,
RocketMQGrpcSourceRecordEmitter<T> recordEmitter,
Configuration config,
SourceReaderContext context) {
+ this(elementsQueue, fetcherManager, recordEmitter, config, context,
null, null);
+ }
+
+ public RocketMQGrpcSourceReader(
+
FutureCompletingBlockingQueue<RecordsWithSplitIds<MessageViewImpl>>
elementsQueue,
+ RocketMQGrpcSourceFetcherManager fetcherManager,
+ RocketMQGrpcSourceRecordEmitter<T> recordEmitter,
+ Configuration config,
+ SourceReaderContext context,
+ @Nullable CheckpointAckTracker ackTracker,
+ @Nullable RocketMQGrpcSourceSplitReader splitReader) {
super(elementsQueue, fetcherManager, recordEmitter, config, context);
+ this.ackTracker = ackTracker;
+ this.splitReader = splitReader;
+ }
+
+ @Override
+ public List<RocketMQGrpcSourceSplit> snapshotState(long checkpointId) {
+ final List<RocketMQGrpcSourceSplit> splits =
super.snapshotState(checkpointId);
+ if (ackTracker != null) {
+ ackTracker.snapshot(checkpointId);
+ }
+ return splits;
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) {
+ if (ackTracker == null || splitReader == null) {
+ return;
+ }
+ for (RocketMQReceiptHandle handle :
ackTracker.completeUpTo(checkpointId)) {
+ try {
+ splitReader.ack(handle);
+ } catch (Exception e) {
+ // The handle is already removed from the tracker, so
tolerating the failure
+ // redelivers only this one message once its invisible
duration expires, while
+ // failing the job would redeliver everything since the last
checkpoint.
+ LOG.warn(
+ "Failed to acknowledge message {} on checkpoint {}
completion; "
+ + "the broker will redeliver the message.",
+ handle.getMessageId(),
+ checkpointId,
+ e);
+ }
+ }
}
@Override
diff --git
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/reader/RocketMQGrpcSourceRecordEmitter.java
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/reader/RocketMQGrpcSourceRecordEmitter.java
index 2dae65e..186d254 100644
---
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/reader/RocketMQGrpcSourceRecordEmitter.java
+++
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/reader/RocketMQGrpcSourceRecordEmitter.java
@@ -28,7 +28,10 @@ import
org.apache.flink.connector.rocketmq.grpc.source.deserialization.RocketMQG
import
org.apache.flink.connector.rocketmq.grpc.source.split.RocketMQGrpcSourceSplitState;
import org.apache.flink.util.Collector;
+import javax.annotation.Nullable;
+
import java.io.IOException;
+import java.util.function.Consumer;
/**
* The {@link RecordEmitter} implementation for the RocketMQ gRPC source. It
deserializes the
@@ -43,15 +46,25 @@ public class RocketMQGrpcSourceRecordEmitter<T>
private final RocketMQGrpcDeserializationSchema<T> deserializationSchema;
private final String namespace;
private final String consumerGroup;
+ @Nullable private final Consumer<RocketMQReceiptHandle>
emittedHandleListener;
private final AckableCollector<T> collector = new AckableCollector<>();
public RocketMQGrpcSourceRecordEmitter(
RocketMQGrpcDeserializationSchema<T> deserializationSchema,
String namespace,
String consumerGroup) {
+ this(deserializationSchema, namespace, consumerGroup, null);
+ }
+
+ public RocketMQGrpcSourceRecordEmitter(
+ RocketMQGrpcDeserializationSchema<T> deserializationSchema,
+ String namespace,
+ String consumerGroup,
+ @Nullable Consumer<RocketMQReceiptHandle> emittedHandleListener) {
this.deserializationSchema = deserializationSchema;
this.namespace = namespace;
this.consumerGroup = consumerGroup;
+ this.emittedHandleListener = emittedHandleListener;
}
@Override
@@ -66,6 +79,9 @@ public class RocketMQGrpcSourceRecordEmitter<T>
element.getMessageView(), namespace,
consumerGroup);
collector.reset(output, element.getEventTime(), handle);
deserializationSchema.deserialize(element, collector);
+ if (emittedHandleListener != null) {
+ emittedHandleListener.accept(handle);
+ }
splitState.incrementProcessedRecords();
} catch (Exception e) {
throw new IOException("Failed to deserialize message due to", e);
diff --git
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/reader/RocketMQGrpcSourceSplitReader.java
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/reader/RocketMQGrpcSourceSplitReader.java
index 570269a..846639b 100644
---
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/reader/RocketMQGrpcSourceSplitReader.java
+++
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/reader/RocketMQGrpcSourceSplitReader.java
@@ -19,11 +19,15 @@
package org.apache.flink.connector.rocketmq.grpc.source.reader;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.base.source.reader.RecordsBySplits;
import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
import org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
import org.apache.flink.connector.base.source.reader.splitreader.SplitsChange;
+import org.apache.flink.connector.rocketmq.grpc.ack.RocketMQReceiptHandle;
+import org.apache.flink.connector.rocketmq.grpc.ack.RocketMQReceiptHandleCodec;
+import org.apache.flink.connector.rocketmq.grpc.source.ConsumerMode;
import
org.apache.flink.connector.rocketmq.grpc.source.InvisibleDurationRenewalPolicies;
import
org.apache.flink.connector.rocketmq.grpc.source.InvisibleDurationRenewalPolicy;
import
org.apache.flink.connector.rocketmq.grpc.source.RocketMQGrpcSourceOptions;
@@ -31,7 +35,6 @@ import
org.apache.flink.connector.rocketmq.grpc.source.split.RocketMQGrpcSourceS
import org.apache.flink.util.FlinkRuntimeException;
import org.apache.rocketmq.client.apis.ClientException;
-import org.apache.rocketmq.client.apis.consumer.LiteSimpleConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -53,15 +56,19 @@ import static
org.apache.flink.util.Preconditions.checkArgument;
* The {@link SplitReader} implementation for the RocketMQ gRPC Pop model.
*
* <p>Because the broker performs message-level load balancing across the
consumer group, this
- * reader does not partition topics. Instead every subtask binds the same main
lite topic. To raise
- * a single subtask's throughput it runs {@code fetch-concurrency} worker
threads that share a
- * single {@link LiteSimpleConsumer}, each issuing a blocking {@code
receive()} long poll. Received
- * messages are handed to the (single) fetcher thread through an internal
queue.
+ * reader does not partition topics. Instead every subtask consumes the same
topic: the main lite
+ * topic in {@link ConsumerMode#LITE}, or the subscribed normal topic in {@link
+ * ConsumerMode#SIMPLE}. To raise a single subtask's throughput it runs {@code
fetch-concurrency}
+ * worker threads that share a single {@link PopConsumer}, each issuing a
blocking {@code receive()}
+ * long poll. Received messages are handed to the (single) fetcher thread
through an internal queue.
*
- * <p>This reader never acknowledges messages: acknowledgement is deferred to
a downstream operator
- * that receives the {@code AckableMessage} records carrying a self-contained
receipt handle.
- * Messages that are never acked become visible again after their invisible
duration and are
- * redelivered by the broker, which provides the at-least-once guarantee.
+ * <p>Acknowledgement depends on the mode. In {@link ConsumerMode#LITE} this
reader never
+ * acknowledges messages: acknowledgement is deferred to a downstream operator
that receives the
+ * {@code AckableMessage} records carrying a self-contained receipt handle. In
{@link
+ * ConsumerMode#SIMPLE} the source reader acknowledges through {@link
#ack(RocketMQReceiptHandle)}
+ * on this very consumer once the checkpoint that observed the messages
completes. Messages that are
+ * never acked become visible again after their invisible duration and are
redelivered by the
+ * broker, which provides the at-least-once guarantee.
*
* <p>When an {@link InvisibleDurationRenewalPolicy} is configured, messages
that are still buffered
* in the internal queue shortly before they would become visible again are
offered to the policy,
@@ -90,7 +97,13 @@ public class RocketMQGrpcSourceSplitReader
private volatile boolean closed = false;
private ReceiveWorker[] workers;
- private LiteSimpleConsumer consumer;
+
+ /**
+ * Written by the fetcher thread in {@link #startWorkers()} and read by
the receive workers, the
+ * renewal thread and the mailbox thread ({@link
#ack(RocketMQReceiptHandle)}), hence volatile.
+ */
+ private volatile PopConsumer consumer;
+
@Nullable private ScheduledThreadPoolExecutor renewalExecutor;
public RocketMQGrpcSourceSplitReader(Configuration configuration) {
@@ -158,10 +171,10 @@ public class RocketMQGrpcSourceSplitReader
return;
}
try {
- consumer = LiteSimpleConsumerProvider.create(configuration);
+ consumer = PopConsumerProvider.create(configuration);
} catch (ClientException e) {
started.set(false);
- throw new FlinkRuntimeException("Failed to create RocketMQ gRPC
LiteSimpleConsumer", e);
+ throw new FlinkRuntimeException("Failed to create the RocketMQ
gRPC consumer", e);
}
if (renewalPolicy != null) {
renewalExecutor =
@@ -180,10 +193,49 @@ public class RocketMQGrpcSourceSplitReader
workers[i] = worker;
worker.start();
}
+ final boolean simple =
+ configuration.get(RocketMQGrpcSourceOptions.MODE) ==
ConsumerMode.SIMPLE;
LOG.info(
- "Started {} receive worker(s) sharing one lite consumer bound
to topic {}",
+ "Started {} receive worker(s) sharing one {} consumer bound to
topic {}",
fetchConcurrency,
- configuration.get(RocketMQGrpcSourceOptions.MAIN_TOPIC));
+ simple ? "simple" : "lite",
+ configuration.get(
+ simple
+ ? RocketMQGrpcSourceOptions.TOPIC
+ : RocketMQGrpcSourceOptions.MAIN_TOPIC));
+ }
+
+ /**
+ * Acknowledge a message that this reader received, removing it from the
Pop invisible set. Used
+ * by the source reader in SIMPLE mode once the checkpoint that observed
the message completes.
+ *
+ * <p>The credential-free {@link RocketMQReceiptHandle} is passed instead
of the SDK {@code
+ * MessageView} on purpose: it keeps the checkpoint ack tracker from
pinning message bodies in
+ * memory, and the minimal view the ack RPC needs is rebuilt here.
+ *
+ * <p>SDK Pop consumers are thread-safe, so acking from the mailbox thread
while the receive
+ * workers block in {@code receive()} on the same instance is safe.
+ *
+ * @param handle the receipt handle of a message emitted by this reader.
+ * @throws ClientException if the ack RPC fails.
+ * @throws IllegalStateException if the consumer has not been started yet
or is already closed.
+ */
+ public void ack(RocketMQReceiptHandle handle) throws ClientException {
+ final PopConsumer currentConsumer = consumer;
+ if (currentConsumer == null || closed) {
+ throw new IllegalStateException(
+ "Cannot acknowledge message "
+ + handle.getMessageId()
+ + ": the RocketMQ gRPC consumer is "
+ + (currentConsumer == null ? "not started yet" :
"already closed"));
+ }
+ currentConsumer.ack(RocketMQReceiptHandleCodec.toAckable(handle));
+ }
+
+ /** Inject a consumer so that ack / renewal behaviour can be tested
without a live cluster. */
+ @VisibleForTesting
+ void setConsumer(PopConsumer consumer) {
+ this.consumer = consumer;
}
@Override
@@ -262,7 +314,7 @@ public class RocketMQGrpcSourceSplitReader
}
/**
- * A single receive loop over the shared {@link LiteSimpleConsumer}.
Because the consumer is
+ * A single receive loop over the shared {@link PopConsumer}. Because the
consumer is
* thread-safe, several workers can call {@code receive()} on it
concurrently; a blocking {@code
* receive()} only occupies its calling thread, which is why several
threads are used to keep
* that many long polls in flight.
diff --git
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/table/RocketMQGrpcConnectorOptions.java
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/table/RocketMQGrpcConnectorOptions.java
index 03b3d36..a59ad69 100644
---
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/table/RocketMQGrpcConnectorOptions.java
+++
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/table/RocketMQGrpcConnectorOptions.java
@@ -122,6 +122,37 @@ public class RocketMQGrpcConnectorOptions {
.noDefaultValue()
.withDescription("The consumer group of the
SimpleConsumer.");
+ public static final ConfigOption<String> MODE =
+ ConfigOptions.key("source.mode")
+ .stringType()
+ .defaultValue("simple")
+ .withDescription(
+ "The consumption mode. 'simple' (the default)
subscribes to the "
+ + "configured topic as a normal topic via
a SimpleConsumer and "
+ + "the source acknowledges messages when
their checkpoint "
+ + "completes (checkpointing must be
enabled and "
+ + "'source.invisible-duration' must exceed
the checkpoint "
+ + "interval plus timeout); 'lite' binds
the configured topic "
+ + "as a main lite topic via a
LiteSimpleConsumer and defers "
+ + "acknowledgement to a downstream
operator.");
+
+ public static final ConfigOption<String> FILTER_EXPRESSION =
+ ConfigOptions.key("source.filter-expression")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The filter expression of the 'simple' mode
subscription, e.g. a tag "
+ + "expression 'tagA||tagB' or a SQL92
expression. When absent, "
+ + "all messages are received.");
+
+ public static final ConfigOption<String> FILTER_TYPE =
+ ConfigOptions.key("source.filter-type")
+ .stringType()
+ .defaultValue("TAG")
+ .withDescription(
+ "The type of 'source.filter-expression': TAG or
SQL92. Only effective "
+ + "in 'simple' mode.");
+
public static final ConfigOption<Duration> AWAIT_DURATION =
ConfigOptions.key("source.await-duration")
.durationType()
@@ -134,9 +165,13 @@ public class RocketMQGrpcConnectorOptions {
.durationType()
.defaultValue(Duration.ofSeconds(60))
.withDescription(
- "The invisible duration of a received message. It
must be larger than "
- + "the checkpoint interval plus the
checkpoint timeout so that "
- + "un-acked messages are redelivered after
a failure.");
+ "The invisible duration of a received message;
un-acked messages are "
+ + "redelivered after it expires. In
'simple' mode the source "
+ + "acknowledges on checkpoint completion,
so size it with "
+ + "headroom above two checkpoint intervals
plus the checkpoint "
+ + "timeout, because a failed checkpoint
defers the ack to the "
+ + "next successful one. In 'lite' mode it
must cover the full "
+ + "downstream processing time instead.");
public static final ConfigOption<Integer> MAX_MESSAGE_NUM =
ConfigOptions.key("source.max-message-num")
diff --git
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/table/RocketMQGrpcDynamicTableFactory.java
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/table/RocketMQGrpcDynamicTableFactory.java
index d6968f9..8c0d0b9 100644
---
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/table/RocketMQGrpcDynamicTableFactory.java
+++
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/table/RocketMQGrpcDynamicTableFactory.java
@@ -27,7 +27,9 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.connector.rocketmq.grpc.RocketMQGrpcOptions;
import org.apache.flink.connector.rocketmq.grpc.sink.RocketMQGrpcSinkOptions;
+import org.apache.flink.connector.rocketmq.grpc.source.ConsumerMode;
import
org.apache.flink.connector.rocketmq.grpc.source.RocketMQGrpcSourceOptions;
+import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.connector.format.DecodingFormat;
import org.apache.flink.table.connector.format.EncodingFormat;
import org.apache.flink.table.connector.sink.DynamicTableSink;
@@ -80,6 +82,9 @@ public class RocketMQGrpcDynamicTableFactory
options.add(RocketMQGrpcConnectorOptions.REQUEST_TIMEOUT);
options.add(RocketMQGrpcConnectorOptions.LITE_TOPIC);
options.add(RocketMQGrpcConnectorOptions.CONSUMER_GROUP);
+ options.add(RocketMQGrpcConnectorOptions.MODE);
+ options.add(RocketMQGrpcConnectorOptions.FILTER_EXPRESSION);
+ options.add(RocketMQGrpcConnectorOptions.FILTER_TYPE);
options.add(RocketMQGrpcConnectorOptions.AWAIT_DURATION);
options.add(RocketMQGrpcConnectorOptions.INVISIBLE_DURATION);
options.add(RocketMQGrpcConnectorOptions.MAX_MESSAGE_NUM);
@@ -194,6 +199,31 @@ public class RocketMQGrpcDynamicTableFactory
configuration,
RocketMQGrpcConnectorOptions.CONSUMER_GROUP,
RocketMQGrpcSourceOptions.CONSUMER_GROUP);
+ final String mode = options.get(RocketMQGrpcConnectorOptions.MODE);
+ if (mode != null) {
+ final ConsumerMode consumerMode;
+ try {
+ consumerMode = ConsumerMode.valueOf(mode.trim().toUpperCase());
+ } catch (IllegalArgumentException e) {
+ throw new ValidationException(
+ "Unsupported value '"
+ + mode
+ + "' for option '"
+ + RocketMQGrpcConnectorOptions.MODE.key()
+ + "'. Supported values are 'lite' and
'simple'.");
+ }
+ configuration.set(RocketMQGrpcSourceOptions.MODE, consumerMode);
+ }
+ copyMapped(
+ options,
+ configuration,
+ RocketMQGrpcConnectorOptions.FILTER_EXPRESSION,
+ RocketMQGrpcSourceOptions.FILTER_EXPRESSION);
+ copyMapped(
+ options,
+ configuration,
+ RocketMQGrpcConnectorOptions.FILTER_TYPE,
+ RocketMQGrpcSourceOptions.FILTER_TYPE);
copyMapped(
options,
configuration,
diff --git
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/table/RocketMQGrpcDynamicTableSource.java
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/table/RocketMQGrpcDynamicTableSource.java
index 5cb7f54..c4d8d50 100644
---
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/table/RocketMQGrpcDynamicTableSource.java
+++
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/table/RocketMQGrpcDynamicTableSource.java
@@ -25,7 +25,10 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.connector.source.Boundedness;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.rocketmq.grpc.ack.AckableMessage;
+import org.apache.flink.connector.rocketmq.grpc.source.ConsumerMode;
import org.apache.flink.connector.rocketmq.grpc.source.RocketMQGrpcSource;
+import
org.apache.flink.connector.rocketmq.grpc.source.RocketMQGrpcSourceBuilder;
+import
org.apache.flink.connector.rocketmq.grpc.source.RocketMQGrpcSourceOptions;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
@@ -119,13 +122,17 @@ public class RocketMQGrpcDynamicTableSource implements
ScanTableSource, Supports
RocketMQGrpcRowDataConverter.forSource(
valueDeserialization, metadataConverters,
producedType);
- final RocketMQGrpcSource<RowData> source =
+ final RocketMQGrpcSourceBuilder<RowData> sourceBuilder =
RocketMQGrpcSource.<RowData>builder()
.setConfig(configuration)
- .setMainTopic(topic)
.setBoundedness(boundedness)
- .setDeserializer(converter)
- .build();
+ .setDeserializer(converter);
+ if (configuration.get(RocketMQGrpcSourceOptions.MODE) ==
ConsumerMode.SIMPLE) {
+ sourceBuilder.setTopic(topic);
+ } else {
+ sourceBuilder.setMainTopic(topic);
+ }
+ final RocketMQGrpcSource<RowData> source = sourceBuilder.build();
// The source produces AckableMessage<RowData>; the SQL/Table path
does not perform
// downstream acknowledgement, so the receipt handle is stripped here
and only the RowData
diff --git
a/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/RocketMQGrpcITCase.java
b/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/RocketMQGrpcITCase.java
index 9ce6d06..c041f9f 100644
---
a/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/RocketMQGrpcITCase.java
+++
b/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/RocketMQGrpcITCase.java
@@ -21,8 +21,10 @@ package org.apache.flink.connector.rocketmq.grpc;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.rocketmq.grpc.ack.AckableMessage;
import
org.apache.flink.connector.rocketmq.grpc.ack.RocketMQThrottleProcessFunction;
import org.apache.flink.connector.rocketmq.grpc.sink.RocketMQGrpcSink;
+import org.apache.flink.connector.rocketmq.grpc.source.ConsumerMode;
import org.apache.flink.connector.rocketmq.grpc.source.RocketMQGrpcSource;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -31,13 +33,16 @@ import org.apache.flink.test.util.AbstractTestBase;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.StaticSessionCredentialsProvider;
+import org.apache.rocketmq.client.apis.consumer.FilterExpression;
+import org.apache.rocketmq.client.apis.consumer.SimpleConsumer;
import org.apache.rocketmq.client.apis.message.Message;
+import org.apache.rocketmq.client.apis.message.MessageView;
import org.apache.rocketmq.client.apis.producer.Producer;
-import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.MethodOrderer;
import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestMethodOrder;
+import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -55,14 +60,12 @@ import java.util.concurrent.TimeUnit;
import static org.assertj.core.api.Assertions.assertThat;
/**
- * Integration tests for the RocketMQ gRPC connector. Requires a running
RocketMQ 5.x instance with
- * gRPC proxy enabled. Connection details are read from environment variables,
so this test is
- * {@link Disabled} by default and must be run manually after exporting {@code
- * ROCKETMQ_GRPC_ENDPOINTS}, {@code ROCKETMQ_GRPC_ACCESS_KEY} and {@code
ROCKETMQ_GRPC_SECRET_KEY}.
+ * Integration tests for the RocketMQ gRPC connector. They require a running
RocketMQ 5.x instance
+ * with the gRPC proxy enabled and are therefore skipped unless {@code
ROCKETMQ_GRPC_ENDPOINTS} is
+ * set; export it together with {@code ROCKETMQ_GRPC_ACCESS_KEY} and {@code
+ * ROCKETMQ_GRPC_SECRET_KEY} to run them against a real cluster.
*/
-@Disabled(
- "Requires a running RocketMQ 5.x instance with gRPC proxy. Set the
ROCKETMQ_GRPC_* "
- + "environment variables and run manually.")
+@EnabledIfEnvironmentVariable(named = "ROCKETMQ_GRPC_ENDPOINTS", matches =
".+")
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
class RocketMQGrpcITCase extends AbstractTestBase {
@@ -80,6 +83,8 @@ class RocketMQGrpcITCase extends AbstractTestBase {
private static final String E2E_SOURCE_TOPIC = "flink-source-1";
private static final String E2E_SINK_TOPIC = "flink-sink-2";
private static final String E2E_SINK_LITE_TOPIC = "flink-sink-2-lite";
+ private static final String SIMPLE_SOURCE_TOPIC =
+ System.getProperty("rocketmq.simple.topic", "flink-source-simple");
private static final String CONSUMER_GROUP = "GID-flink";
private static final int NUM_MESSAGES = 50;
@@ -141,6 +146,7 @@ class RocketMQGrpcITCase extends AbstractTestBase {
RocketMQGrpcSource.<String>builder()
.setEndpoints(ENDPOINTS)
.setConsumerGroup(CONSUMER_GROUP)
+ .setMode(ConsumerMode.LITE)
.setMainTopic(SOURCE_TOPIC)
.setConfig(RocketMQGrpcOptions.ACCESS_KEY, ACCESS_KEY)
.setConfig(RocketMQGrpcOptions.SECRET_KEY, SECRET_KEY)
@@ -186,6 +192,7 @@ class RocketMQGrpcITCase extends AbstractTestBase {
RocketMQGrpcSource.<String>builder()
.setEndpoints(ENDPOINTS)
.setConsumerGroup(CONSUMER_GROUP)
+ .setMode(ConsumerMode.LITE)
.setMainTopic(E2E_SOURCE_TOPIC)
.setConfig(RocketMQGrpcOptions.ACCESS_KEY, ACCESS_KEY)
.setConfig(RocketMQGrpcOptions.SECRET_KEY, SECRET_KEY)
@@ -246,6 +253,120 @@ class RocketMQGrpcITCase extends AbstractTestBase {
}
}
+ // -----------------------------------------------------------------------
+ // Test: Simple mode consumes a normal topic with checkpoint-based ack
+ // -----------------------------------------------------------------------
+
+ @Test
+ void testSimpleModeConsumesNormalTopic() throws Exception {
+ final String prefix = "simple-" + runId + "-";
+
+ seedMessagesViaProducer(SIMPLE_SOURCE_TOPIC, NUM_MESSAGES, prefix);
+
+ final StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(1);
+ env.enableCheckpointing(5_000);
+
+ final RocketMQGrpcSource<String> source =
+ RocketMQGrpcSource.<String>builder()
+ .setEndpoints(ENDPOINTS)
+ .setConsumerGroup(CONSUMER_GROUP)
+ .setMode(ConsumerMode.SIMPLE)
+ .setTopic(SIMPLE_SOURCE_TOPIC)
+ .setConfig(RocketMQGrpcOptions.ACCESS_KEY, ACCESS_KEY)
+ .setConfig(RocketMQGrpcOptions.SECRET_KEY, SECRET_KEY)
+ .setValueOnlyDeserializer(new SimpleStringSchema())
+ .build();
+
+ final DataStream<String> stream =
+ env.fromSource(source, WatermarkStrategy.noWatermarks(),
"RocketMQ gRPC Source")
+ .map(AckableMessage::getValue)
+ .returns(String.class);
+ stream.addSink(new CollectingSinkFunction());
+
+ // Start the job in a daemon thread (same pattern as startAndCollect),
but keep it
+ // alive after enough records are collected: SIMPLE mode acks only
after a checkpoint
+ // completes, and closing the job immediately would leave nothing
acked.
+ CollectingSinkFunction.QUEUE.clear();
+ final Thread jobThread =
+ new Thread(
+ () -> {
+ try {
+ env.execute("Simple Mode Collecting Job");
+ } catch (Exception e) {
+ LOG.info("Simple mode collecting job ended:
{}", e.getMessage());
+ }
+ });
+ jobThread.setDaemon(true);
+ jobThread.start();
+
+ // Collect messages matching the prefix until we have enough or
timeout.
+ final List<String> results = new ArrayList<>();
+ final long deadline = System.currentTimeMillis() +
COLLECT_TIMEOUT_SECONDS * 1000;
+ while (results.size() < NUM_MESSAGES && System.currentTimeMillis() <
deadline) {
+ final String item = CollectingSinkFunction.QUEUE.poll(500,
TimeUnit.MILLISECONDS);
+ if (item != null && item.startsWith(prefix)) {
+ results.add(item);
+ }
+ }
+ LOG.info("Collected {} messages (prefix={}) in simple mode",
results.size(), prefix);
+
+ assertThat(results).hasSize(NUM_MESSAGES);
+ for (int i = 0; i < NUM_MESSAGES; i++) {
+ assertThat(results).contains(prefix + i);
+ }
+
+ // Keep the job running so that at least one checkpoint completes
(interval is 5s)
+ // and the reader acks the tracked receipt handles.
+ Thread.sleep(20_000);
+
+ env.close();
+ jobThread.interrupt();
+ jobThread.join(10_000);
+
+ // Wait for the default 60s invisible duration to expire so that any
un-acked message
+ // would become visible for redelivery again.
+ Thread.sleep(75_000);
+
+ // Verify no un-acked messages remain: a fresh consumer under the same
group must not
+ // receive any message of this run's prefix.
+ final ClientServiceProvider provider =
ClientServiceProvider.loadService();
+ final ClientConfiguration clientConfig =
+ ClientConfiguration.newBuilder()
+ .setEndpoints(ENDPOINTS)
+ .setCredentialProvider(
+ new
StaticSessionCredentialsProvider(ACCESS_KEY, SECRET_KEY))
+ .build();
+
+ try (SimpleConsumer consumer =
+ provider.newSimpleConsumerBuilder()
+ .setClientConfiguration(clientConfig)
+ .setConsumerGroup(CONSUMER_GROUP)
+ .setAwaitDuration(Duration.ofSeconds(5))
+ .setSubscriptionExpressions(
+ Collections.singletonMap(
+ SIMPLE_SOURCE_TOPIC,
FilterExpression.SUB_ALL))
+ .build()) {
+ int redelivered = 0;
+ final List<MessageView> messages = consumer.receive(32,
Duration.ofSeconds(20));
+ for (MessageView msg : messages) {
+ final ByteBuffer buf = msg.getBody();
+ final byte[] bytes = new byte[buf.remaining()];
+ buf.get(bytes);
+ final String body = new String(bytes, StandardCharsets.UTF_8);
+ if (body.startsWith(prefix)) {
+ redelivered++;
+ }
+ }
+ assertThat(redelivered)
+ .withFailMessage(
+ "Expected no redelivered messages after
checkpoint-aligned ack,"
+ + " but got %d",
+ redelivered)
+ .isEqualTo(0);
+ }
+ }
+
// -----------------------------------------------------------------------
// Helpers
// -----------------------------------------------------------------------
diff --git
a/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/RocketMQGrpcLiteE2EVerify.java
b/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/RocketMQGrpcLiteE2EVerify.java
index bea6673..070247b 100644
---
a/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/RocketMQGrpcLiteE2EVerify.java
+++
b/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/RocketMQGrpcLiteE2EVerify.java
@@ -23,6 +23,7 @@ import
org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import
org.apache.flink.connector.rocketmq.grpc.ack.RocketMQThrottleProcessFunction;
import org.apache.flink.connector.rocketmq.grpc.sink.RocketMQGrpcSink;
+import org.apache.flink.connector.rocketmq.grpc.source.ConsumerMode;
import
org.apache.flink.connector.rocketmq.grpc.source.InvisibleDurationRenewalPolicy;
import org.apache.flink.connector.rocketmq.grpc.source.RocketMQGrpcSource;
import
org.apache.flink.connector.rocketmq.grpc.source.RocketMQGrpcSourceOptions;
@@ -111,6 +112,7 @@ public class RocketMQGrpcLiteE2EVerify {
RocketMQGrpcSource.<String>builder()
.setEndpoints(ENDPOINTS)
.setConsumerGroup(CONSUMER_GROUP)
+ .setMode(ConsumerMode.LITE)
.setMainTopic(MAIN_TOPIC)
.setValueOnlyDeserializer(new SimpleStringSchema())
.build();
@@ -195,6 +197,7 @@ public class RocketMQGrpcLiteE2EVerify {
RocketMQGrpcSource.<String>builder()
.setEndpoints(ENDPOINTS)
.setConsumerGroup(CONSUMER_GROUP)
+ .setMode(ConsumerMode.LITE)
.setMainTopic(MAIN_TOPIC)
.setConfig(
RocketMQGrpcSourceOptions.INVISIBLE_DURATION,
diff --git
a/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/ack/RocketMQLiteAckClientTest.java
b/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/ack/RocketMQAckClientTest.java
similarity index 59%
rename from
flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/ack/RocketMQLiteAckClientTest.java
rename to
flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/ack/RocketMQAckClientTest.java
index 5c9dc5e..3acd987 100644
---
a/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/ack/RocketMQLiteAckClientTest.java
+++
b/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/ack/RocketMQAckClientTest.java
@@ -25,8 +25,8 @@ import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
-/** Tests for the reference counting behaviour of {@link
RocketMQLiteAckClient#acquire}. */
-class RocketMQLiteAckClientTest {
+/** Tests for the reference counting behaviour of {@link
RocketMQAckClient#acquire}. */
+class RocketMQAckClientTest {
private static Configuration config(String endpoints) {
final Configuration configuration = new Configuration();
@@ -37,20 +37,20 @@ class RocketMQLiteAckClientTest {
@Test
void sameConfigurationSharesASingleReferenceCountedClient() {
final Configuration configuration = config("127.0.0.1:8080");
-
assertThat(RocketMQLiteAckClient.getReferenceCount(configuration)).isZero();
+
assertThat(RocketMQAckClient.getReferenceCount(configuration)).isZero();
- final RocketMQLiteAckClient first =
RocketMQLiteAckClient.acquire(configuration);
- final RocketMQLiteAckClient second =
RocketMQLiteAckClient.acquire(configuration);
+ final RocketMQAckClient first =
RocketMQAckClient.acquire(configuration);
+ final RocketMQAckClient second =
RocketMQAckClient.acquire(configuration);
// The same client is shared and the reference count is incremented
per acquire.
assertThat(second).isSameAs(first);
-
assertThat(RocketMQLiteAckClient.getReferenceCount(configuration)).isEqualTo(2);
+
assertThat(RocketMQAckClient.getReferenceCount(configuration)).isEqualTo(2);
- RocketMQLiteAckClient.release(configuration);
-
assertThat(RocketMQLiteAckClient.getReferenceCount(configuration)).isEqualTo(1);
+ RocketMQAckClient.release(configuration);
+
assertThat(RocketMQAckClient.getReferenceCount(configuration)).isEqualTo(1);
- RocketMQLiteAckClient.release(configuration);
-
assertThat(RocketMQLiteAckClient.getReferenceCount(configuration)).isZero();
+ RocketMQAckClient.release(configuration);
+
assertThat(RocketMQAckClient.getReferenceCount(configuration)).isZero();
}
@Test
@@ -58,25 +58,25 @@ class RocketMQLiteAckClientTest {
final Configuration a = config("host-a:8080");
final Configuration b = config("host-b:8080");
- final RocketMQLiteAckClient clientA = RocketMQLiteAckClient.acquire(a);
- final RocketMQLiteAckClient clientB = RocketMQLiteAckClient.acquire(b);
+ final RocketMQAckClient clientA = RocketMQAckClient.acquire(a);
+ final RocketMQAckClient clientB = RocketMQAckClient.acquire(b);
try {
assertThat(clientA).isNotSameAs(clientB);
-
assertThat(RocketMQLiteAckClient.getReferenceCount(a)).isEqualTo(1);
-
assertThat(RocketMQLiteAckClient.getReferenceCount(b)).isEqualTo(1);
+ assertThat(RocketMQAckClient.getReferenceCount(a)).isEqualTo(1);
+ assertThat(RocketMQAckClient.getReferenceCount(b)).isEqualTo(1);
} finally {
- RocketMQLiteAckClient.release(a);
- RocketMQLiteAckClient.release(b);
+ RocketMQAckClient.release(a);
+ RocketMQAckClient.release(b);
}
- assertThat(RocketMQLiteAckClient.getReferenceCount(a)).isZero();
- assertThat(RocketMQLiteAckClient.getReferenceCount(b)).isZero();
+ assertThat(RocketMQAckClient.getReferenceCount(a)).isZero();
+ assertThat(RocketMQAckClient.getReferenceCount(b)).isZero();
}
@Test
void releaseWithoutAcquireIsANoOp() {
final Configuration configuration = config("no-acquire:8080");
- RocketMQLiteAckClient.release(configuration);
-
assertThat(RocketMQLiteAckClient.getReferenceCount(configuration)).isZero();
+ RocketMQAckClient.release(configuration);
+
assertThat(RocketMQAckClient.getReferenceCount(configuration)).isZero();
}
}
diff --git
a/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/source/RocketMQGrpcSourceBuilderTest.java
b/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/source/RocketMQGrpcSourceBuilderTest.java
index 21a8c6a..1f78bda 100644
---
a/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/source/RocketMQGrpcSourceBuilderTest.java
+++
b/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/source/RocketMQGrpcSourceBuilderTest.java
@@ -34,6 +34,7 @@ class RocketMQGrpcSourceBuilderTest {
RocketMQGrpcSource.<String>builder()
.setEndpoints("127.0.0.1:8080")
.setConsumerGroup("group")
+ .setMode(ConsumerMode.LITE)
.setMainTopic("topic")
.setValueOnlyDeserializer(new SimpleStringSchema())
.build();
@@ -46,6 +47,7 @@ class RocketMQGrpcSourceBuilderTest {
() ->
RocketMQGrpcSource.<String>builder()
.setConsumerGroup("group")
+ .setMode(ConsumerMode.LITE)
.setMainTopic("topic")
.setValueOnlyDeserializer(new
SimpleStringSchema())
.build())
@@ -59,6 +61,7 @@ class RocketMQGrpcSourceBuilderTest {
() ->
RocketMQGrpcSource.<String>builder()
.setEndpoints("127.0.0.1:8080")
+ .setMode(ConsumerMode.LITE)
.setMainTopic("topic")
.setValueOnlyDeserializer(new
SimpleStringSchema())
.build())
@@ -73,6 +76,7 @@ class RocketMQGrpcSourceBuilderTest {
RocketMQGrpcSource.<String>builder()
.setEndpoints("127.0.0.1:8080")
.setConsumerGroup("group")
+ .setMode(ConsumerMode.LITE)
.setValueOnlyDeserializer(new
SimpleStringSchema())
.build())
.isInstanceOf(IllegalArgumentException.class)
@@ -86,9 +90,68 @@ class RocketMQGrpcSourceBuilderTest {
RocketMQGrpcSource.<String>builder()
.setEndpoints("127.0.0.1:8080")
.setConsumerGroup("group")
+ .setMode(ConsumerMode.LITE)
.setMainTopic("topic")
.build())
.isInstanceOf(NullPointerException.class)
.hasMessageContaining("deserializer");
}
+
+ @Test
+ void buildSucceedsWithSimpleModeTest() {
+ final RocketMQGrpcSource<String> source =
+ RocketMQGrpcSource.<String>builder()
+ .setEndpoints("127.0.0.1:8080")
+ .setConsumerGroup("group")
+ .setMode(ConsumerMode.SIMPLE)
+ .setTopic("normal-topic")
+ .setFilterExpression("tagA||tagB")
+ .setValueOnlyDeserializer(new SimpleStringSchema())
+ .build();
+ assertThat(source).isNotNull();
+ }
+
+ @Test
+ void buildFailsWithSimpleModeWithoutTopicTest() {
+ assertThatThrownBy(
+ () ->
+ RocketMQGrpcSource.<String>builder()
+ .setEndpoints("127.0.0.1:8080")
+ .setConsumerGroup("group")
+ .setMode(ConsumerMode.SIMPLE)
+ .setValueOnlyDeserializer(new
SimpleStringSchema())
+ .build())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("topic");
+ }
+
+ @Test
+ void buildFailsWithLiteModeWhenOnlySimpleTopicIsSetTest() {
+ assertThatThrownBy(
+ () ->
+ RocketMQGrpcSource.<String>builder()
+ .setEndpoints("127.0.0.1:8080")
+ .setConsumerGroup("group")
+ .setMode(ConsumerMode.LITE)
+ .setTopic("normal-topic")
+ .setValueOnlyDeserializer(new
SimpleStringSchema())
+ .build())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("main topic");
+ }
+
+ @Test
+ void buildDefaultsToSimpleModeTest() {
+ // Without setMode() the builder must run in SIMPLE mode: a LITE-mode
build would reject
+ // the missing main topic.
+ final RocketMQGrpcSource<String> source =
+ RocketMQGrpcSource.<String>builder()
+ .setEndpoints("127.0.0.1:8080")
+ .setConsumerGroup("group")
+ .setTopic("normal-topic")
+ .setValueOnlyDeserializer(new SimpleStringSchema())
+ .build();
+ assertThat(source).isNotNull();
+
assertThat(RocketMQGrpcSourceOptions.MODE.defaultValue()).isEqualTo(ConsumerMode.SIMPLE);
+ }
}
diff --git
a/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/source/reader/CheckpointAckTrackerTest.java
b/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/source/reader/CheckpointAckTrackerTest.java
new file mode 100644
index 0000000..6604a61
--- /dev/null
+++
b/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/source/reader/CheckpointAckTrackerTest.java
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.rocketmq.grpc.source.reader;
+
+import org.apache.flink.connector.rocketmq.grpc.ack.RocketMQReceiptHandle;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link CheckpointAckTracker}. */
+class CheckpointAckTrackerTest {
+
+ @Test
+ void completeUpToReturnsSnapshottedHandlesTest() {
+ final CheckpointAckTracker tracker = new CheckpointAckTracker();
+ tracker.add(handle(1));
+ tracker.add(handle(2));
+ tracker.snapshot(7L);
+
+ final List<RocketMQReceiptHandle> completed = tracker.completeUpTo(7L);
+ assertThat(completed).containsExactly(handle(1), handle(2));
+ }
+
+ @Test
+ void completeUpToKeepsLaterCheckpointsTest() {
+ final CheckpointAckTracker tracker = new CheckpointAckTracker();
+ tracker.add(handle(1));
+ tracker.snapshot(7L);
+ tracker.add(handle(2));
+ tracker.snapshot(8L);
+
+ assertThat(tracker.completeUpTo(7L)).containsExactly(handle(1));
+ assertThat(tracker.completeUpTo(8L)).containsExactly(handle(2));
+ assertThat(tracker.completeUpTo(9L)).isEmpty();
+ }
+
+ @Test
+ void completeUpToMergesAllCheckpointsUpToIdTest() {
+ final CheckpointAckTracker tracker = new CheckpointAckTracker();
+ tracker.add(handle(1));
+ tracker.snapshot(1L);
+ tracker.add(handle(2));
+ tracker.snapshot(2L);
+
+ assertThat(tracker.completeUpTo(2L)).containsExactly(handle(1),
handle(2));
+ }
+
+ @Test
+ void handlesAddedAfterSnapshotGoToNextCheckpointTest() {
+ final CheckpointAckTracker tracker = new CheckpointAckTracker();
+ tracker.snapshot(1L);
+ tracker.add(handle(1));
+
+ assertThat(tracker.completeUpTo(1L)).isEmpty();
+ tracker.snapshot(2L);
+ assertThat(tracker.completeUpTo(2L)).containsExactly(handle(1));
+ }
+
+ @Test
+ void emptySnapshotIsNotRecordedTest() {
+ final CheckpointAckTracker tracker = new CheckpointAckTracker();
+ tracker.snapshot(1L);
+ tracker.add(handle(1));
+ tracker.snapshot(2L);
+
+ // Checkpoint 1 carried nothing; handle(1) belongs to checkpoint 2
only.
+ assertThat(tracker.completeUpTo(1L)).isEmpty();
+ assertThat(tracker.completeUpTo(2L)).containsExactly(handle(1));
+ }
+
+ private static RocketMQReceiptHandle handle(int id) {
+ return new RocketMQReceiptHandle(
+ "127.0.0.1:8080", "", "group", "topic", null, "msg-" + id,
"receipt-" + id, 1);
+ }
+}
diff --git
a/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/source/reader/PopConsumerProviderTest.java
b/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/source/reader/PopConsumerProviderTest.java
new file mode 100644
index 0000000..cfa76ad
--- /dev/null
+++
b/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/source/reader/PopConsumerProviderTest.java
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.rocketmq.grpc.source.reader;
+
+import org.apache.rocketmq.client.apis.consumer.LiteSimpleConsumer;
+import org.apache.rocketmq.client.apis.consumer.SimpleConsumer;
+import org.apache.rocketmq.client.apis.message.MessageView;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Proxy;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests that the {@link PopConsumer} adapters of {@link PopConsumerProvider}
delegate to the
+ * wrapped SDK consumer. Building a real consumer requires a live proxy, so
the SDK interfaces are
+ * stubbed with dynamic proxies that only record the invoked method and its
arguments.
+ */
+class PopConsumerProviderTest {
+
+ @Test
+ void litePopConsumerDelegatesAckTest() throws Exception {
+ final List<Invocation> invocations = new ArrayList<>();
+ final PopConsumer consumer =
+
PopConsumerProvider.wrap(recordingProxy(LiteSimpleConsumer.class, invocations));
+
+ final MessageView messageView = messageView();
+ consumer.ack(messageView);
+
+ assertThat(invocations).hasSize(1);
+ assertThat(invocations.get(0).method).isEqualTo("ack");
+ assertThat(invocations.get(0).argument).isSameAs(messageView);
+ }
+
+ @Test
+ void simplePopConsumerDelegatesAckTest() throws Exception {
+ final List<Invocation> invocations = new ArrayList<>();
+ final PopConsumer consumer =
+ PopConsumerProvider.wrap(recordingProxy(SimpleConsumer.class,
invocations));
+
+ final MessageView messageView = messageView();
+ consumer.ack(messageView);
+
+ assertThat(invocations).hasSize(1);
+ assertThat(invocations.get(0).method).isEqualTo("ack");
+ assertThat(invocations.get(0).argument).isSameAs(messageView);
+ }
+
+ private static <T> T recordingProxy(Class<T> type, List<Invocation>
invocations) {
+ return type.cast(
+ Proxy.newProxyInstance(
+ type.getClassLoader(),
+ new Class<?>[] {type},
+ (proxy, method, args) -> {
+ invocations.add(
+ new Invocation(
+ method.getName(), args == null ?
null : args[0]));
+ return null;
+ }));
+ }
+
+ private static MessageView messageView() {
+ return (MessageView)
+ Proxy.newProxyInstance(
+ MessageView.class.getClassLoader(),
+ new Class<?>[] {MessageView.class},
+ (proxy, method, args) -> {
+ throw new UnsupportedOperationException();
+ });
+ }
+
+ private static final class Invocation {
+
+ private final String method;
+ private final Object argument;
+
+ private Invocation(String method, Object argument) {
+ this.method = method;
+ this.argument = argument;
+ }
+ }
+}
diff --git
a/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/source/reader/RocketMQGrpcSourceSplitReaderTest.java
b/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/source/reader/RocketMQGrpcSourceSplitReaderTest.java
new file mode 100644
index 0000000..ab635c3
--- /dev/null
+++
b/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/source/reader/RocketMQGrpcSourceSplitReaderTest.java
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.rocketmq.grpc.source.reader;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.rocketmq.grpc.ack.RocketMQReceiptHandle;
+
+import org.apache.rocketmq.client.apis.ClientException;
+import org.apache.rocketmq.client.apis.message.MessageView;
+import org.apache.rocketmq.client.java.message.MessageIdCodec;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for the SIMPLE-mode acknowledgement path of {@link
RocketMQGrpcSourceSplitReader}. */
+class RocketMQGrpcSourceSplitReaderTest {
+
+ @Test
+ void ackBeforeStartThrowsTest() throws Exception {
+ final RocketMQGrpcSourceSplitReader reader =
+ new RocketMQGrpcSourceSplitReader(new Configuration());
+ try {
+ assertThatThrownBy(() -> reader.ack(handle()))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("not started yet");
+ } finally {
+ reader.close();
+ }
+ }
+
+ @Test
+ void ackAfterCloseThrowsTest() throws Exception {
+ final RocketMQGrpcSourceSplitReader reader =
+ new RocketMQGrpcSourceSplitReader(new Configuration());
+ final RecordingPopConsumer consumer = new RecordingPopConsumer();
+ reader.setConsumer(consumer);
+ reader.close();
+
+ assertThatThrownBy(() -> reader.ack(handle()))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("already closed");
+ assertThat(consumer.acked).isEmpty();
+ }
+
+ @Test
+ void ackDelegatesRebuiltMessageViewToConsumerTest() throws Exception {
+ final RocketMQGrpcSourceSplitReader reader =
+ new RocketMQGrpcSourceSplitReader(new Configuration());
+ final RecordingPopConsumer consumer = new RecordingPopConsumer();
+ reader.setConsumer(consumer);
+
+ final RocketMQReceiptHandle handle = handle();
+ try {
+ reader.ack(handle);
+ } finally {
+ reader.close();
+ }
+
+ assertThat(consumer.acked).hasSize(1);
+ final org.apache.rocketmq.client.java.message.MessageViewImpl acked =
+ (org.apache.rocketmq.client.java.message.MessageViewImpl)
consumer.acked.get(0);
+ assertThat(acked.getMessageId())
+
.isEqualTo(MessageIdCodec.getInstance().decode(handle.getMessageId()));
+ assertThat(acked.getTopic()).isEqualTo(handle.getTopic());
+
assertThat(acked.getReceiptHandle()).isEqualTo(handle.getReceiptHandle());
+ }
+
+ private static RocketMQReceiptHandle handle() {
+ return new RocketMQReceiptHandle(
+ "127.0.0.1:8081",
+ "ns-1",
+ "GID-test",
+ "normal-topic",
+ null,
+ MessageIdCodec.getInstance().nextMessageId().toString(),
+ "receipt-handle-abc",
+ 1);
+ }
+
+ private static final class RecordingPopConsumer implements PopConsumer {
+
+ private final List<MessageView> acked = new ArrayList<>();
+
+ @Override
+ public List<MessageView> receive(int maxMessageNum, Duration
invisibleDuration)
+ throws ClientException {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public void ack(MessageView messageView) throws ClientException {
+ acked.add(messageView);
+ }
+
+ @Override
+ public void changeInvisibleDuration(MessageView messageView, Duration
invisibleDuration)
+ throws ClientException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void close() throws IOException {}
+ }
+}
diff --git
a/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/table/RocketMQGrpcDynamicTableFactoryTest.java
b/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/table/RocketMQGrpcDynamicTableFactoryTest.java
index 67aa3d4..8db4bc6 100644
---
a/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/table/RocketMQGrpcDynamicTableFactoryTest.java
+++
b/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/table/RocketMQGrpcDynamicTableFactoryTest.java
@@ -53,6 +53,7 @@ class RocketMQGrpcDynamicTableFactoryTest {
@Test
void createTableSourceTest() {
+ // No 'source.mode' is configured, so the source runs in the default
'simple' mode.
final Map<String, String> options = baseOptions();
options.put("source.consumer-group", "test-group");
@@ -60,6 +61,16 @@ class RocketMQGrpcDynamicTableFactoryTest {
assertThat(source).isInstanceOf(RocketMQGrpcDynamicTableSource.class);
}
+ @Test
+ void createLiteModeTableSourceTest() {
+ final Map<String, String> options = baseOptions();
+ options.put("source.consumer-group", "test-group");
+ options.put("source.mode", "lite");
+
+ final DynamicTableSource source =
FactoryMocks.createTableSource(SCHEMA, options);
+ assertThat(source).isInstanceOf(RocketMQGrpcDynamicTableSource.class);
+ }
+
@Test
void createTableSinkTest() {
final DynamicTableSink sink = FactoryMocks.createTableSink(SCHEMA,
baseOptions());
@@ -80,4 +91,25 @@ class RocketMQGrpcDynamicTableFactoryTest {
assertThatThrownBy(() -> FactoryMocks.createTableSink(SCHEMA, options))
.hasStackTraceContaining("lite-topic");
}
+
+ @Test
+ void createSimpleModeTableSourceTest() {
+ final Map<String, String> options = baseOptions();
+ options.put("source.consumer-group", "test-group");
+ options.put("source.mode", "simple");
+ options.put("source.filter-expression", "tagA||tagB");
+
+ final DynamicTableSource source =
FactoryMocks.createTableSource(SCHEMA, options);
+ assertThat(source).isInstanceOf(RocketMQGrpcDynamicTableSource.class);
+ }
+
+ @Test
+ void createTableSourceFailsWithInvalidModeTest() {
+ final Map<String, String> options = baseOptions();
+ options.put("source.consumer-group", "test-group");
+ options.put("source.mode", "bogus");
+
+ assertThatThrownBy(() -> FactoryMocks.createTableSource(SCHEMA,
options))
+ .hasStackTraceContaining("bogus");
+ }
}
diff --git a/flink-connector-rocketmq/pom.xml b/flink-connector-rocketmq/pom.xml
index 8f71191..b022b86 100644
--- a/flink-connector-rocketmq/pom.xml
+++ b/flink-connector-rocketmq/pom.xml
@@ -26,7 +26,7 @@ under the License.
<parent>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-rocketmq-parent</artifactId>
- <version>1.0.0-SNAPSHOT</version>
+ <version>3.0.0</version>
</parent>
<artifactId>flink-connector-rocketmq</artifactId>
diff --git a/flink-sql-connector-rocketmq-grpc/pom.xml
b/flink-sql-connector-rocketmq-grpc/pom.xml
index 963169a..92444ab 100644
--- a/flink-sql-connector-rocketmq-grpc/pom.xml
+++ b/flink-sql-connector-rocketmq-grpc/pom.xml
@@ -26,7 +26,7 @@ under the License.
<parent>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-rocketmq-parent</artifactId>
- <version>1.0.0-SNAPSHOT</version>
+ <version>3.0.0</version>
</parent>
<artifactId>flink-sql-connector-rocketmq-grpc</artifactId>
diff --git a/flink-sql-connector-rocketmq/pom.xml
b/flink-sql-connector-rocketmq/pom.xml
index 1724963..ad3bc8f 100644
--- a/flink-sql-connector-rocketmq/pom.xml
+++ b/flink-sql-connector-rocketmq/pom.xml
@@ -26,7 +26,7 @@ under the License.
<parent>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-rocketmq-parent</artifactId>
- <version>1.0.0-SNAPSHOT</version>
+ <version>3.0.0</version>
</parent>
<artifactId>flink-sql-connector-rocketmq</artifactId>
diff --git a/pom.xml b/pom.xml
index 7570a25..377451f 100644
--- a/pom.xml
+++ b/pom.xml
@@ -32,7 +32,7 @@ under the License.
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-rocketmq-parent</artifactId>
- <version>1.0.0-SNAPSHOT</version>
+ <version>3.0.0</version>
<name>Flink : Connectors : RocketMQ : Parent</name>
<packaging>pom</packaging>
<url>https://flink.apache.org</url>
