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 71511a9 [ISSUE #142] fix(grpc): subscribe the configured lite topics
and fail on a rejected consumer (#146)
71511a9 is described below
commit 71511a903261e9202e8bfb703d663c24f4dc5eb0
Author: lizhimins <[email protected]>
AuthorDate: Mon Oct 26 14:47:57 2026 +0800
[ISSUE #142] fix(grpc): subscribe the configured lite topics and fail on a
rejected consumer (#146)
* [ISSUE #141] fix(remoting): stop delivering the wrong data silently
An acceptance run against a commercial 5.5.1 cluster found a family of
defects
with the same shape: the job stays RUNNING, checkpoints succeed, nothing is
logged, and the data is wrong or absent. Every fix below either makes the
behaviour correct or makes the failure loud.
Source. rocketmq.source.filter.sql was accepted, logged as a warning and
then
ignored, so every message was delivered regardless of the predicate; assign
mode
cannot push a SQL92 expression to the broker, so it is now evaluated on the
client with rocketmq-filter, the implementation the broker uses, and an
expression that does not compile fails at reader construction.
RowDeserializationSchema read its format options from keys such as
"fielddelimiter" while the table factory registers and documents
"rocketmq.client.message.field.delimiter". The DDL values never arrived, the
field delimiter stayed at its default, and every record of a multi column
table
was dropped as "field missing" -- a SQL source that returns zero rows
without
failing. Both row deserializers now read the canonical keys, and an
unsupported
message.length.check is rejected instead of ignored.
RocketMQSourceReaderMetrics declared nine metric names and was threaded
through
four classes, but its constructor and both registration methods had empty
bodies, so a running job exposed no RocketMQ metric at all. Offsets, commit
counters and per queue lag are now reported, and the lag is summed into
Flink's
pendingRecords. Lag needs the broker's maximum offset, which the reader
refreshes asynchronously at most once every ten seconds.
RocketMQSourceFunction could not start at all: open() allocates the
subtask's
queues itself and then seeks each one, but the consumer was in subscribe
mode,
where allocation belongs to the client's asynchronous rebalance, so the
assigned
list was still empty and every seek failed with "The message queue is not in
assigned list". The allocated queues are now assigned explicitly, which also
matches the new source.
offset.commit.auto was read only to print its value while the consumer was
hardcoded to setAutoCommit(false); pull.rpc.timeout was declared and never
read.
Both are applied now, and the auto-commit default becomes false to match the
behaviour the code always had. An unknown startup.scan.mode fell through to
latest, making a misspelled mode look like a healthy job reading from the
wrong
position; it is rejected.
Sink. RocketMQSinkBuilder accepted a MessageQueueSelector, null checked it
and
stored it, and both send paths then called send(Message), so records that
had to
stay ordered were spread over every queue while the API said otherwise. The
selector now reaches the send call with the record as its sharding argument.
The committer logged a failed commit without the throwable, so a failure
produced a message with no cause and no stack trace. A record that failed to
emit was reported as "Failed to deserialize message due to" wherever the
failure
actually came from, and named no record; the wrapper now covers only the
deserialization call and the message carries the message id, topic, queue
and
offset.
The async path of the legacy RocketMQSink logged a send failure and carried
on,
so a job whose every record was rejected stayed RUNNING, kept checkpointing
and
wrote nothing. The first failure is now parked and rethrown from invoke()
and
finish().
Table. A validation failure repeated every table option, and a WITH clause
normally carries the credentials, publishing the access key to the
JobManager
log and the REST exception history. Both factories now report the missing
keys
only, before delegating to the helper, which still detects unknown options.
Removed. rocketmq.source.startup.offset.strategy and
rocketmq.sink.serialize.format were read by nothing, and the first was even
registered as a DDL option; an option that does nothing should not be
offered.
The duplicate CollectorOption key sets are gone with the fix above, as is
the
InnerConsumerImpl.wakeup() that paused every assigned queue with no matching
resume anywhere in the codebase, and the message key separator that was an
empty
string instead of MessageConst.KEY_SEPARATOR.
* [ISSUE #142] fix(grpc): subscribe the configured lite topics and fail on
a rejected consumer
LITE mode never subscribed to a single sub lite topic. The builder bound the
main topic and nothing else: there was no subscribeLite call anywhere in the
module and no option to name the sub topics, so the SDK's lite topic set
stayed
empty and the broker delivered nothing. The job ran, checkpointed and logged
nothing while reading zero rows.
Add rocketmq.source.sub-lite-topics and fail in build() when LITE mode is
configured without it, so a missing subscription cannot be mistaken for an
idle
topic. Acknowledgement stays with the user's downstream operator: the source
exposes the receipt handle and RocketMQAckClient performs the ack, because a
message that is never acknowledged is redelivered forever once its invisible
duration expires. The Table path strips the receipt handle and can therefore
never ack, so source.mode='lite' is rejected there instead of being accepted
into an endless redelivery loop.
A consumer group the server registered as LITE cannot be used by a SIMPLE
mode
source: the proxy answers ILLEGAL_CONSUMER_GROUP on every request. The
receive
workers caught every ClientException and retried forever, so the job stayed
RUNNING, read no record and reported nothing. Failures that no retry can
fix now
stop the reader and are rethrown on the fetcher thread, which is the only
thread
whose exception reaches the Flink task. FORBIDDEN stays retryable on
purpose,
since the proxy also returns it as backpressure while a lite consumer holds
too
many unacknowledged messages.
Consumer creation caught ClientException only, while the SDK reports a
rejected
client as an unchecked exception wrapped by its Guava service. That escaped
uncaught and left the reader marked as started with no consumer, so a later
split assignment silently did nothing. The catch is widened to Throwable,
and a
lite subscription rejected after the consumer is built now closes it
instead of
leaking its channels and threads.
A validation failure also repeated every table option, publishing the
access key
to the JobManager log and the REST exception history; the factory now
reports
the missing keys only.
---
.../rocketmq/grpc/source/ConsumerMode.java | 8 +-
.../grpc/source/RocketMQGrpcSourceBuilder.java | 36 +++-
.../grpc/source/RocketMQGrpcSourceOptions.java | 19 +-
.../grpc/source/reader/PopConsumerProvider.java | 62 +++++-
.../reader/RocketMQGrpcSourceSplitReader.java | 153 +++++++++++---
.../grpc/table/RocketMQGrpcConnectorOptions.java | 8 +-
.../table/RocketMQGrpcDynamicTableFactory.java | 44 ++++-
.../grpc/table/RocketMQGrpcDynamicTableSource.java | 17 +-
.../rocketmq/grpc/RocketMQGrpcLiteE2EVerify.java | 16 +-
.../grpc/source/RocketMQGrpcSourceBuilderTest.java | 30 +++
.../source/reader/PopConsumerProviderTest.java | 62 ++++++
.../reader/RocketMQGrpcSourceSplitReaderTest.java | 219 +++++++++++++++++++++
.../table/RocketMQGrpcDynamicTableFactoryTest.java | 60 +++++-
13 files changed, 680 insertions(+), 54 deletions(-)
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
index 63dcc5a..7a537d5 100644
---
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
@@ -25,9 +25,11 @@ import org.apache.flink.annotation.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.
+ * Bind a main lite topic via a {@code LiteSimpleConsumer} and subscribe
the configured sub lite
+ * topics, whose messages the broker delivers over one receive stream.
Binding the main topic
+ * without subscribing any sub lite topic leaves the subscription empty
and receives nothing.
+ * Messages are acknowledged by a downstream operator using the receipt
handle carried by each
+ * emitted record.
*/
LITE,
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 ebc0861..ea0c51d 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
@@ -27,6 +27,8 @@ import
org.apache.flink.connector.rocketmq.grpc.RocketMQGrpcOptions;
import
org.apache.flink.connector.rocketmq.grpc.source.deserialization.RocketMQGrpcDeserializationSchema;
import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -37,6 +39,7 @@ public class RocketMQGrpcSourceBuilder<OUT> {
private final Configuration configuration;
private String mainTopic;
+ private List<String> subLiteTopics;
private ConsumerMode mode = ConsumerMode.SIMPLE;
private String topic;
private Boundedness boundedness;
@@ -60,7 +63,8 @@ public class RocketMQGrpcSourceBuilder<OUT> {
/**
* 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}.
+ * #setMode(ConsumerMode)}; ignored in {@link ConsumerMode#SIMPLE}. The
sub lite topics to
+ * consume are set separately via {@link #setSubLiteTopics(List)}.
*/
public RocketMQGrpcSourceBuilder<OUT> setMainTopic(String mainTopic) {
checkArgument(
@@ -70,6 +74,25 @@ public class RocketMQGrpcSourceBuilder<OUT> {
return this;
}
+ /**
+ * Set the sub lite topics subscribed under the main lite topic. Required
and non-empty in
+ * {@link ConsumerMode#LITE}; ignored in {@link ConsumerMode#SIMPLE}. Only
messages carrying one
+ * of these lite topics are delivered, so a source that binds the main
topic without subscribing
+ * any sub lite topic receives nothing.
+ */
+ public RocketMQGrpcSourceBuilder<OUT> setSubLiteTopics(List<String>
subLiteTopics) {
+ checkNotNull(subLiteTopics, "sub lite topics must not be null");
+ final List<String> normalized = new ArrayList<>(subLiteTopics.size());
+ for (String subLiteTopic : subLiteTopics) {
+ checkArgument(
+ subLiteTopic != null && !subLiteTopic.trim().isEmpty(),
+ "sub lite topic must not be null or blank");
+ normalized.add(subLiteTopic);
+ }
+ this.subLiteTopics = normalized;
+ return this;
+ }
+
/** Set the consumption mode. Defaults to {@link ConsumerMode#SIMPLE}. */
public RocketMQGrpcSourceBuilder<OUT> setMode(ConsumerMode mode) {
this.mode = checkNotNull(mode);
@@ -169,6 +192,17 @@ public class RocketMQGrpcSourceBuilder<OUT> {
mainTopic != null && !mainTopic.trim().isEmpty(),
"the main topic must be configured");
configuration.set(RocketMQGrpcSourceOptions.MAIN_TOPIC, mainTopic);
+ if (subLiteTopics != null) {
+ configuration.set(RocketMQGrpcSourceOptions.SUB_LITE_TOPICS,
subLiteTopics);
+ }
+ final List<String> configuredSubLiteTopics =
+
configuration.get(RocketMQGrpcSourceOptions.SUB_LITE_TOPICS);
+ checkArgument(
+ configuredSubLiteTopics != null &&
!configuredSubLiteTopics.isEmpty(),
+ "at least one sub lite topic must be configured in LITE
mode via '%s';"
+ + " binding the main topic alone subscribes to
nothing, so the source"
+ + " would receive no message",
+ RocketMQGrpcSourceOptions.SUB_LITE_TOPICS.key());
} else {
checkArgument(
topic != null && !topic.trim().isEmpty(),
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 f54d473..c2e9487 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
@@ -23,6 +23,7 @@ import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import java.time.Duration;
+import java.util.List;
/**
* Configuration options for the RocketMQ gRPC {@code SimpleConsumer} based
source. These options
@@ -92,7 +93,23 @@ public class RocketMQGrpcSourceOptions {
"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.");
+ + "across the consumer group. Binding
alone does not select any "
+ + "message: the sub lite topics to consume
must be listed in "
+ + "'rocketmq.source.sub-lite-topics'.");
+
+ public static final ConfigOption<List<String>> SUB_LITE_TOPICS =
+ ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "sub-lite-topics")
+ .stringType()
+ .asList()
+ .noDefaultValue()
+ .withDescription(
+ "The sub lite topics subscribed under the main
lite topic, e.g. "
+ + "'subA;subB'. Required and non-empty in
LITE mode; ignored in "
+ + "SIMPLE mode. Each entry is registered
through "
+ + "LiteSimpleConsumer#subscribeLite, and
only messages carrying "
+ + "one of these lite topics are delivered.
Leaving this empty "
+ + "subscribes to nothing, so the source
would run idle and "
+ + "emit no record at all.");
public static final ConfigOption<Integer> FETCH_CONCURRENCY =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "fetch-concurrency")
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
index 7111912..ea43e6b 100644
---
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
@@ -32,6 +32,8 @@ 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 org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.time.Duration;
@@ -40,17 +42,25 @@ 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}.
+ * LiteSimpleConsumer} bound to the main lite topic and subscribed to the
configured sub lite topics
+ * 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 static final Logger LOG =
LoggerFactory.getLogger(PopConsumerProvider.class);
+
private PopConsumerProvider() {}
/**
* Build and start the {@link PopConsumer} selected by {@link
RocketMQGrpcSourceOptions#MODE}.
+ *
+ * <p>Every failure to build or subscribe the consumer is propagated to
the caller: the SDK
+ * reports a rejected client as an unchecked exception (its Guava service
wraps the startup
+ * failure in an {@code IllegalStateException}), so callers must not
narrow the failure to
+ * {@link ClientException}. Swallowing it would leave the source running
against a consumer the
+ * server never accepted, which reads no record at all while the job stays
healthy.
*/
public static PopConsumer create(Configuration configuration) throws
ClientException {
final ClientServiceProvider provider =
ClientServiceProvider.loadService();
@@ -85,9 +95,55 @@ public class PopConsumerProvider {
configuration.get(RocketMQGrpcSourceOptions.AWAIT_DURATION))
.bindTopic(configuration.get(RocketMQGrpcSourceOptions.MAIN_TOPIC))
.build();
+ subscribeOrClose(consumer,
configuration.get(RocketMQGrpcSourceOptions.SUB_LITE_TOPICS));
return new LitePopConsumer(consumer);
}
+ /**
+ * Register the sub lite topics on an already started consumer, closing it
again when the
+ * subscription is rejected so that its channels and threads are not
leaked.
+ */
+ @VisibleForTesting
+ static void subscribeOrClose(LiteSimpleConsumer consumer, List<String>
subLiteTopics)
+ throws ClientException {
+ try {
+ subscribeLiteTopics(consumer, subLiteTopics);
+ } catch (RuntimeException | ClientException e) {
+ closeQuietly(consumer);
+ throw e;
+ }
+ }
+
+ private static void closeQuietly(LiteSimpleConsumer consumer) {
+ try {
+ consumer.close();
+ } catch (Exception e) {
+ LOG.warn("Failed to close a LiteSimpleConsumer whose subscription
was rejected", e);
+ }
+ }
+
+ /**
+ * Register the configured sub lite topics on a freshly built {@code
LiteSimpleConsumer}.
+ *
+ * <p>{@code bindTopic} only tells the SDK which parent topic the consumer
belongs to. The
+ * broker delivers a message once its lite topic is part of the consumer's
lite subscription
+ * set, so without this step the source runs idle and emits nothing.
+ */
+ @VisibleForTesting
+ static void subscribeLiteTopics(LiteSimpleConsumer consumer, List<String>
subLiteTopics)
+ throws ClientException {
+ if (subLiteTopics == null || subLiteTopics.isEmpty()) {
+ throw new IllegalArgumentException(
+ "LITE mode requires at least one sub lite topic via '"
+ + RocketMQGrpcSourceOptions.SUB_LITE_TOPICS.key()
+ + "'; binding the main topic alone subscribes to
nothing, so the source"
+ + " would receive no message at all");
+ }
+ for (String subLiteTopic : subLiteTopics) {
+ consumer.subscribeLite(subLiteTopic);
+ }
+ }
+
/** Wrap an SDK {@code LiteSimpleConsumer} into the connector's {@link
PopConsumer} surface. */
@VisibleForTesting
static PopConsumer wrap(LiteSimpleConsumer consumer) {
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 846639b..ea29edf 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
@@ -35,6 +35,10 @@ 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.java.exception.BadRequestException;
+import org.apache.rocketmq.client.java.exception.NotFoundException;
+import org.apache.rocketmq.client.java.exception.UnauthorizedException;
+import org.apache.rocketmq.client.java.exception.UnsupportedException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -49,6 +53,7 @@ import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import static org.apache.flink.util.Preconditions.checkArgument;
@@ -74,6 +79,12 @@ import static
org.apache.flink.util.Preconditions.checkArgument;
* in the internal queue shortly before they would become visible again are
offered to the policy,
* which may extend their invisible duration. Messages already handed to the
record emitter are
* never renewed because renewing refreshes the receipt handle carried
downstream.
+ *
+ * <p>A failure that retrying cannot fix fails the Flink task instead of being
retried. The server
+ * rejects a misconfigured consumer on every request -- a consumer group
registered as LITE and used
+ * by a {@link ConsumerMode#SIMPLE} source, for instance, is answered with
{@code
+ * ILLEGAL_CONSUMER_GROUP} -- and looping on such a rejection would keep the
job RUNNING while it
+ * reads no record at all.
*/
@Internal
public class RocketMQGrpcSourceSplitReader
@@ -95,6 +106,13 @@ public class RocketMQGrpcSourceSplitReader
private final AtomicBoolean wakeup = new AtomicBoolean(false);
private final AtomicBoolean started = new AtomicBoolean(false);
+ /**
+ * A failure that makes further consumption impossible. Set by a receive
worker and surfaced by
+ * the fetcher thread in {@link #fetch()}, because the fetcher is the only
thread whose
+ * exception reaches the Flink task.
+ */
+ private final AtomicReference<Throwable> fatalFailure = new
AtomicReference<>();
+
private volatile boolean closed = false;
private ReceiveWorker[] workers;
@@ -129,6 +147,14 @@ public class RocketMQGrpcSourceSplitReader
@Override
public RecordsWithSplitIds<MessageViewImpl> fetch() {
+ // A worker that hit a failure no retry can fix must fail the task
instead of letting this
+ // reader hand back empty batches forever, which would keep the job
RUNNING while it reads
+ // nothing.
+ final Throwable failure = fatalFailure.get();
+ if (failure != null) {
+ throw new FlinkRuntimeException(
+ "The RocketMQ gRPC consumer failed and the source stopped
consuming", failure);
+ }
if (wakeup.compareAndSet(true, false) || closed) {
return new RecordsBySplits.Builder<MessageViewImpl>().build();
}
@@ -172,9 +198,19 @@ public class RocketMQGrpcSourceSplitReader
}
try {
consumer = PopConsumerProvider.create(configuration);
- } catch (ClientException e) {
+ } catch (Throwable t) {
+ // Catch everything, not only ClientException: the SDK reports a
consumer the server
+ // rejected as an unchecked exception (its Guava service wraps the
startup failure in an
+ // IllegalStateException), and letting that escape would leave
this reader marked as
+ // started with no consumer, so a later split assignment would
silently do nothing.
started.set(false);
- throw new FlinkRuntimeException("Failed to create the RocketMQ
gRPC consumer", e);
+ recordFatalFailure(t);
+ throw new FlinkRuntimeException(
+ String.format(
+ "Failed to create the RocketMQ gRPC consumer for
group '%s' in %s mode",
+
configuration.get(RocketMQGrpcSourceOptions.CONSUMER_GROUP),
+ configuration.get(RocketMQGrpcSourceOptions.MODE)),
+ t);
}
if (renewalPolicy != null) {
renewalExecutor =
@@ -195,14 +231,36 @@ public class RocketMQGrpcSourceSplitReader
}
final boolean simple =
configuration.get(RocketMQGrpcSourceOptions.MODE) ==
ConsumerMode.SIMPLE;
- LOG.info(
- "Started {} receive worker(s) sharing one {} consumer bound to
topic {}",
- fetchConcurrency,
- simple ? "simple" : "lite",
- configuration.get(
- simple
- ? RocketMQGrpcSourceOptions.TOPIC
- : RocketMQGrpcSourceOptions.MAIN_TOPIC));
+ if (simple) {
+ LOG.info(
+ "Started {} receive worker(s) sharing one simple consumer
subscribed to topic {}",
+ fetchConcurrency,
+ configuration.get(RocketMQGrpcSourceOptions.TOPIC));
+ } else {
+ LOG.info(
+ "Started {} receive worker(s) sharing one lite consumer
bound to topic {} with sub lite topics {}",
+ fetchConcurrency,
+ configuration.get(RocketMQGrpcSourceOptions.MAIN_TOPIC),
+
configuration.get(RocketMQGrpcSourceOptions.SUB_LITE_TOPICS));
+ }
+ }
+
+ /**
+ * Start the receive workers against an injected consumer, so that failure
handling can be
+ * tested without a live cluster.
+ */
+ @VisibleForTesting
+ void startWorkersWith(PopConsumer consumer) {
+ if (!started.compareAndSet(false, true)) {
+ return;
+ }
+ this.consumer = consumer;
+ workers = new ReceiveWorker[fetchConcurrency];
+ for (int i = 0; i < fetchConcurrency; i++) {
+ final ReceiveWorker worker = new ReceiveWorker(i);
+ workers[i] = worker;
+ worker.start();
+ }
}
/**
@@ -238,6 +296,43 @@ public class RocketMQGrpcSourceSplitReader
this.consumer = consumer;
}
+ /**
+ * Records a failure that no retry can recover from and stops every
receive worker. The failure
+ * is rethrown by {@link #fetch()} on the fetcher thread, which is the
only thread whose
+ * exception the Flink task observes.
+ */
+ private void recordFatalFailure(Throwable cause) {
+ if (fatalFailure.compareAndSet(null, cause)) {
+ LOG.error(
+ "The RocketMQ gRPC consumer failed permanently; stopping
the source instead of"
+ + " reading no record while the job stays running",
+ cause);
+ }
+ }
+
+ /**
+ * Whether a {@link ClientException} reports a condition that retrying
cannot fix.
+ *
+ * <p>The SDK maps every non-OK proxy status onto a dedicated exception
type. A rejected request
+ * identity -- an illegal consumer group or topic, an unknown topic or
group, wrong credentials,
+ * an unsupported operation -- is a deployment error: for example a
consumer group the server
+ * registered as LITE cannot be used by a SIMPLE mode source, and the
proxy answers every
+ * request with {@code ILLEGAL_CONSUMER_GROUP}. Retrying that forever
keeps the job RUNNING
+ * while it reads nothing, so it fails the task instead.
+ *
+ * <p>Everything else keeps the existing backoff. {@code
ForbiddenException} in particular is
+ * treated as transient on purpose: the proxy also answers {@code
FORBIDDEN} while a lite
+ * consumer holds too many unacknowledged messages, which is backpressure
rather than a
+ * misconfiguration.
+ */
+ @VisibleForTesting
+ static boolean isPermanentFailure(ClientException e) {
+ return e instanceof BadRequestException
+ || e instanceof UnauthorizedException
+ || e instanceof NotFoundException
+ || e instanceof UnsupportedException;
+ }
+
@Override
public void wakeUp() {
wakeup.compareAndSet(false, true);
@@ -338,7 +433,7 @@ public class RocketMQGrpcSourceSplitReader
@Override
public void run() {
long backoffMs = RECEIVE_BACKOFF_INITIAL_MS;
- while (running && !closed) {
+ while (running && !closed && fatalFailure.get() == null) {
try {
final
List<org.apache.rocketmq.client.apis.message.MessageView> messages =
consumer.receive(maxMessageNum, invisibleDuration);
@@ -356,20 +451,30 @@ public class RocketMQGrpcSourceSplitReader
Thread.currentThread().interrupt();
break;
} catch (ClientException e) {
- if (running && !closed) {
- LOG.warn(
- "Receive worker {} failed to receive messages,
retrying in {} ms",
- index,
- backoffMs,
- e);
- try {
- Thread.sleep(backoffMs);
- } catch (InterruptedException ie) {
- Thread.currentThread().interrupt();
- break;
- }
- backoffMs = Math.min(backoffMs * 2,
RECEIVE_BACKOFF_MAX_MS);
+ if (!running || closed) {
+ break;
}
+ if (isPermanentFailure(e)) {
+ recordFatalFailure(e);
+ break;
+ }
+ LOG.warn(
+ "Receive worker {} failed to receive messages,
retrying in {} ms",
+ index,
+ backoffMs,
+ e);
+ try {
+ Thread.sleep(backoffMs);
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ break;
+ }
+ backoffMs = Math.min(backoffMs * 2,
RECEIVE_BACKOFF_MAX_MS);
+ } catch (Throwable t) {
+ // An unchecked failure from the SDK would otherwise kill
this worker thread and
+ // leave the source polling an empty queue forever.
+ recordFatalFailure(t);
+ break;
}
}
}
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 a59ad69..0c43214 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
@@ -132,9 +132,11 @@ public class RocketMQGrpcConnectorOptions {
+ "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.");
+ + "interval plus timeout). 'lite' is only
available through the "
+ + "DataStream API: it defers
acknowledgement to a downstream "
+ + "operator holding the receipt handle,
which a SQL/Table "
+ + "pipeline has no way to express, so the
table source rejects "
+ + "it.");
public static final ConfigOption<String> FILTER_EXPRESSION =
ConfigOptions.key("source.filter-expression")
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 8c0d0b9..ff4377d 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
@@ -43,7 +43,10 @@ import
org.apache.flink.table.factories.SerializationFormatFactory;
import org.apache.flink.table.types.DataType;
import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
import java.util.Set;
+import java.util.stream.Collectors;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -102,6 +105,7 @@ public class RocketMQGrpcDynamicTableFactory
final DecodingFormat<DeserializationSchema<RowData>> decodingFormat =
helper.discoverDecodingFormat(
DeserializationFormatFactory.class,
FactoryUtil.FORMAT);
+ validateRequiredOptions(requiredOptions(),
context.getCatalogTable().getOptions());
helper.validate();
final ReadableConfig options = helper.getOptions();
@@ -111,8 +115,20 @@ public class RocketMQGrpcDynamicTableFactory
RocketMQGrpcConnectorOptions.CONSUMER_GROUP.key());
final DataType physicalDataType = context.getPhysicalRowDataType();
+ final Configuration configuration = buildConfiguration(options);
+ if (configuration.get(RocketMQGrpcSourceOptions.MODE) ==
ConsumerMode.LITE) {
+ throw new ValidationException(
+ String.format(
+ "'%s' = 'lite' is not supported by the RocketMQ
gRPC table source."
+ + " LITE defers acknowledgement to a
downstream operator holding"
+ + " the receipt handle, which the
SQL/Table path strips, so"
+ + " every message would be redelivered
once its invisible"
+ + " duration expires. Use the DataStream
API"
+ + " (RocketMQGrpcSource with
ConsumerMode.LITE) instead.",
+ RocketMQGrpcConnectorOptions.MODE.key()));
+ }
return new RocketMQGrpcDynamicTableSource(
- buildConfiguration(options),
+ configuration,
options.get(RocketMQGrpcConnectorOptions.TOPIC),
Boundedness.CONTINUOUS_UNBOUNDED,
physicalDataType,
@@ -125,6 +141,7 @@ public class RocketMQGrpcDynamicTableFactory
FactoryUtil.createTableFactoryHelper(this, context);
final EncodingFormat<SerializationSchema<RowData>> encodingFormat =
helper.discoverEncodingFormat(SerializationFormatFactory.class,
FactoryUtil.FORMAT);
+ validateRequiredOptions(requiredOptions(),
context.getCatalogTable().getOptions());
helper.validate();
final ReadableConfig options = helper.getOptions();
@@ -142,6 +159,31 @@ public class RocketMQGrpcDynamicTableFactory
encodingFormat);
}
+ /**
+ * Reports the required options the DDL does not set, naming the missing
keys only.
+ *
+ * <p>{@link FactoryUtil.TableFactoryHelper#validate()} still runs
afterwards, so unknown
+ * options are detected exactly as before. This check runs first because a
validation failure
+ * must never repeat an option value: a WITH clause may carry credentials
('access-key',
+ * 'secret-key') and the message of a validation failure is written to the
JobManager log and to
+ * the REST exception history.
+ */
+ private static void validateRequiredOptions(
+ Set<ConfigOption<?>> requiredOptions, Map<String, String>
tableOptions) {
+ final List<String> missing =
+ requiredOptions.stream()
+ .map(ConfigOption::key)
+ .filter(key -> !tableOptions.containsKey(key))
+ .sorted()
+ .collect(Collectors.toList());
+ if (!missing.isEmpty()) {
+ throw new ValidationException(
+ String.format(
+ "Missing required option%s for the RocketMQ gRPC
connector: %s",
+ missing.size() == 1 ? "" : "s", missing));
+ }
+ }
+
/**
* Translates the user-facing SQL DDL options ({@link
RocketMQGrpcConnectorOptions}) into a
* {@link Configuration} keyed by the programmatic/SDK options that the
source and sink builders
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 c4d8d50..20e8b2f 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,10 +25,7 @@ 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;
@@ -122,17 +119,15 @@ public class RocketMQGrpcDynamicTableSource implements
ScanTableSource, Supports
RocketMQGrpcRowDataConverter.forSource(
valueDeserialization, metadataConverters,
producedType);
- final RocketMQGrpcSourceBuilder<RowData> sourceBuilder =
+ // The table factory rejects ConsumerMode.LITE (the SQL path cannot
acknowledge), so this is
+ // always a SIMPLE mode subscription to a normal topic.
+ final RocketMQGrpcSource<RowData> source =
RocketMQGrpcSource.<RowData>builder()
.setConfig(configuration)
.setBoundedness(boundedness)
- .setDeserializer(converter);
- if (configuration.get(RocketMQGrpcSourceOptions.MODE) ==
ConsumerMode.SIMPLE) {
- sourceBuilder.setTopic(topic);
- } else {
- sourceBuilder.setMainTopic(topic);
- }
- final RocketMQGrpcSource<RowData> source = sourceBuilder.build();
+ .setDeserializer(converter)
+ .setTopic(topic)
+ .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/RocketMQGrpcLiteE2EVerify.java
b/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/RocketMQGrpcLiteE2EVerify.java
index 070247b..e855a95 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
@@ -35,6 +35,7 @@ import javax.annotation.Nullable;
import java.time.Duration;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -61,6 +62,9 @@ public class RocketMQGrpcLiteE2EVerify {
System.getProperty("rocketmq.endpoints", "47.98.110.140:8081");
private static final String MAIN_TOPIC = "LiteTest";
private static final String CONSUMER_GROUP = "LiteSimpleConsumerGroup";
+ private static final String E2E_LITE_TOPIC = "flink-e2e-lite";
+ private static final String RENEW_LITE_TOPIC_PREFIX = "flink-renew-lite-";
+ private static final int RENEW_LITE_TOPIC_COUNT = 20;
private static final String RUN_ID =
UUID.randomUUID().toString().substring(0, 8);
@@ -104,7 +108,7 @@ public class RocketMQGrpcLiteE2EVerify {
for (int i = 0; i < 5; i++) {
messages.add(throttlePrefix + i);
}
- runSinkJob(messages, "flink-e2e-lite");
+ runSinkJob(messages, E2E_LITE_TOPIC);
System.out.println("[phase1] sink wrote " + messages.size() + " lite
messages");
final StreamExecutionEnvironment env = localEnv();
@@ -114,6 +118,7 @@ public class RocketMQGrpcLiteE2EVerify {
.setConsumerGroup(CONSUMER_GROUP)
.setMode(ConsumerMode.LITE)
.setMainTopic(MAIN_TOPIC)
+
.setSubLiteTopics(Collections.singletonList(E2E_LITE_TOPIC))
.setValueOnlyDeserializer(new SimpleStringSchema())
.build();
@@ -175,16 +180,18 @@ public class RocketMQGrpcLiteE2EVerify {
// Spread across lite topics: the broker paces event dispatch per lite
topic when un-acked
// messages accumulate, which would starve the backpressure scenario
otherwise.
final StreamExecutionEnvironment sinkEnv = localEnv();
- for (int t = 0; t < 20; t++) {
+ final List<String> renewLiteTopics = new
ArrayList<>(RENEW_LITE_TOPIC_COUNT);
+ for (int t = 0; t < RENEW_LITE_TOPIC_COUNT; t++) {
+ renewLiteTopics.add(RENEW_LITE_TOPIC_PREFIX + t);
final List<String> messages = new ArrayList<>();
- for (int i = t; i < total; i += 20) {
+ for (int i = t; i < total; i += RENEW_LITE_TOPIC_COUNT) {
messages.add(prefix + i);
}
final RocketMQGrpcSink<String> sink =
RocketMQGrpcSink.<String>builder()
.setEndpoints(ENDPOINTS)
.setTopic(MAIN_TOPIC)
- .setLiteTopic("flink-renew-lite-" + t)
+ .setLiteTopic(RENEW_LITE_TOPIC_PREFIX + t)
.setValueOnlySerializer(new SimpleStringSchema())
.build();
sinkEnv.fromData(messages).sinkTo(sink);
@@ -199,6 +206,7 @@ public class RocketMQGrpcLiteE2EVerify {
.setConsumerGroup(CONSUMER_GROUP)
.setMode(ConsumerMode.LITE)
.setMainTopic(MAIN_TOPIC)
+ .setSubLiteTopics(renewLiteTopics)
.setConfig(
RocketMQGrpcSourceOptions.INVISIBLE_DURATION,
Duration.ofSeconds(20))
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 1f78bda..9ed72bd 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
@@ -22,6 +22,9 @@ import
org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.junit.jupiter.api.Test;
+import java.util.Arrays;
+import java.util.Collections;
+
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -36,6 +39,7 @@ class RocketMQGrpcSourceBuilderTest {
.setConsumerGroup("group")
.setMode(ConsumerMode.LITE)
.setMainTopic("topic")
+ .setSubLiteTopics(Arrays.asList("subA", "subB"))
.setValueOnlyDeserializer(new SimpleStringSchema())
.build();
assertThat(source).isNotNull();
@@ -83,6 +87,31 @@ class RocketMQGrpcSourceBuilderTest {
.hasMessageContaining("main topic");
}
+ @Test
+ void buildFailsWithLiteModeWithoutSubLiteTopicsTest() {
+ assertThatThrownBy(
+ () ->
+ RocketMQGrpcSource.<String>builder()
+ .setEndpoints("127.0.0.1:8080")
+ .setConsumerGroup("group")
+ .setMode(ConsumerMode.LITE)
+ .setMainTopic("topic")
+ .setValueOnlyDeserializer(new
SimpleStringSchema())
+ .build())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("sub lite topic");
+ }
+
+ @Test
+ void buildFailsWithBlankSubLiteTopicTest() {
+ assertThatThrownBy(
+ () ->
+ RocketMQGrpcSource.<String>builder()
+
.setSubLiteTopics(Collections.singletonList(" ")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("sub lite topic must not be null or
blank");
+ }
+
@Test
void buildFailsWithoutDeserializerTest() {
assertThatThrownBy(
@@ -92,6 +121,7 @@ class RocketMQGrpcSourceBuilderTest {
.setConsumerGroup("group")
.setMode(ConsumerMode.LITE)
.setMainTopic("topic")
+
.setSubLiteTopics(Collections.singletonList("subA"))
.build())
.isInstanceOf(NullPointerException.class)
.hasMessageContaining("deserializer");
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
index cfa76ad..1bfbade 100644
---
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
@@ -25,9 +25,12 @@ import org.junit.jupiter.api.Test;
import java.lang.reflect.Proxy;
import java.util.ArrayList;
+import java.util.Arrays;
+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 that the {@link PopConsumer} adapters of {@link PopConsumerProvider}
delegate to the
@@ -64,6 +67,65 @@ class PopConsumerProviderTest {
assertThat(invocations.get(0).argument).isSameAs(messageView);
}
+ @Test
+ void subscribeLiteTopicsRegistersEverySubTopicTest() throws Exception {
+ final List<Invocation> invocations = new ArrayList<>();
+ final LiteSimpleConsumer consumer =
recordingProxy(LiteSimpleConsumer.class, invocations);
+
+ PopConsumerProvider.subscribeLiteTopics(consumer,
Arrays.asList("subA", "subB"));
+
+ assertThat(invocations).hasSize(2);
+ assertThat(invocations.get(0).method).isEqualTo("subscribeLite");
+ assertThat(invocations.get(0).argument).isEqualTo("subA");
+ assertThat(invocations.get(1).method).isEqualTo("subscribeLite");
+ assertThat(invocations.get(1).argument).isEqualTo("subB");
+ }
+
+ @Test
+ void subscribeLiteTopicsRejectsEmptySubscriptionTest() {
+ final LiteSimpleConsumer consumer =
+ recordingProxy(LiteSimpleConsumer.class, new ArrayList<>());
+
+ assertThatThrownBy(
+ () ->
+ PopConsumerProvider.subscribeLiteTopics(
+ consumer, Collections.emptyList()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("sub lite topic");
+ }
+
+ /**
+ * A rejected lite subscription must not leak the already started
consumer, otherwise its
+ * channels and threads outlive the failed task.
+ */
+ @Test
+ void subscribeOrCloseClosesConsumerWhenSubscriptionFailsTest() {
+ final List<Invocation> invocations = new ArrayList<>();
+ final LiteSimpleConsumer consumer =
recordingProxy(LiteSimpleConsumer.class, invocations);
+
+ assertThatThrownBy(
+ () ->
+ PopConsumerProvider.subscribeOrClose(
+ consumer, Collections.emptyList()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("sub lite topic");
+
+ assertThat(invocations).hasSize(1);
+ assertThat(invocations.get(0).method).isEqualTo("close");
+ }
+
+ @Test
+ void subscribeOrCloseKeepsConsumerWhenSubscriptionSucceedsTest() throws
Exception {
+ final List<Invocation> invocations = new ArrayList<>();
+ final LiteSimpleConsumer consumer =
recordingProxy(LiteSimpleConsumer.class, invocations);
+
+ PopConsumerProvider.subscribeOrClose(consumer, Arrays.asList("subA",
"subB"));
+
+ assertThat(invocations).hasSize(2);
+ assertThat(invocations.get(0).method).isEqualTo("subscribeLite");
+ assertThat(invocations.get(1).method).isEqualTo("subscribeLite");
+ }
+
private static <T> T recordingProxy(Class<T> type, List<Invocation>
invocations) {
return type.cast(
Proxy.newProxyInstance(
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
index ab635c3..74c5c56 100644
---
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
@@ -19,10 +19,22 @@
package org.apache.flink.connector.rocketmq.grpc.source.reader;
import org.apache.flink.configuration.Configuration;
+import
org.apache.flink.connector.base.source.reader.splitreader.SplitsAddition;
+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.source.split.RocketMQGrpcSourceSplit;
+import org.apache.flink.util.FlinkRuntimeException;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.message.MessageView;
+import org.apache.rocketmq.client.java.exception.BadRequestException;
+import org.apache.rocketmq.client.java.exception.ForbiddenException;
+import org.apache.rocketmq.client.java.exception.InternalErrorException;
+import org.apache.rocketmq.client.java.exception.NotFoundException;
+import org.apache.rocketmq.client.java.exception.ProxyTimeoutException;
+import org.apache.rocketmq.client.java.exception.TooManyRequestsException;
+import org.apache.rocketmq.client.java.exception.UnauthorizedException;
+import org.apache.rocketmq.client.java.exception.UnsupportedException;
import org.apache.rocketmq.client.java.message.MessageIdCodec;
import org.junit.jupiter.api.Test;
@@ -31,6 +43,8 @@ import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -38,6 +52,19 @@ import static
org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for the SIMPLE-mode acknowledgement path of {@link
RocketMQGrpcSourceSplitReader}. */
class RocketMQGrpcSourceSplitReaderTest {
+ /** Status codes of the gRPC protocol, see {@code
apache.rocketmq.v2.Code}. */
+ private static final int ILLEGAL_CONSUMER_GROUP = 40003;
+
+ private static final int UNAUTHORIZED = 40100;
+ private static final int FORBIDDEN = 40300;
+ private static final int CONSUMER_GROUP_NOT_FOUND = 40403;
+ private static final int TOO_MANY_REQUESTS = 42900;
+ private static final int INTERNAL_ERROR = 50000;
+ private static final int PROXY_TIMEOUT = 50400;
+ private static final int UNSUPPORTED = 50500;
+
+ private static final String REQUEST_ID = "request-id";
+
@Test
void ackBeforeStartThrowsTest() throws Exception {
final RocketMQGrpcSourceSplitReader reader =
@@ -88,6 +115,156 @@ class RocketMQGrpcSourceSplitReaderTest {
assertThat(acked.getReceiptHandle()).isEqualTo(handle.getReceiptHandle());
}
+ /**
+ * A consumer group the server registered as LITE cannot be used by a
SIMPLE mode source: the
+ * proxy answers every request with {@code ILLEGAL_CONSUMER_GROUP}. The
reader must fail the
+ * task instead of retrying that rejection forever while the job stays
RUNNING and reads
+ * nothing.
+ */
+ @Test
+ void liteGroupRejectedOnReceiveFailsFetchTest() throws Exception {
+ final RocketMQGrpcSourceSplitReader reader =
+ new RocketMQGrpcSourceSplitReader(new Configuration());
+ final ClientException rejection =
+ new BadRequestException(
+ ILLEGAL_CONSUMER_GROUP,
+ REQUEST_ID,
+ "lite group must use LITE mode: GID-test");
+ reader.startWorkersWith(new FailingPopConsumer(rejection, 1));
+ try {
+ assertThatThrownBy(() -> awaitFetchFailure(reader))
+ .isInstanceOf(FlinkRuntimeException.class)
+ .hasRootCause(rejection)
+ .hasStackTraceContaining("lite group must use LITE mode:
GID-test");
+ } finally {
+ reader.close();
+ }
+ }
+
+ @Test
+ void unexpectedReceiveFailureFailsFetchTest() throws Exception {
+ final RocketMQGrpcSourceSplitReader reader =
+ new RocketMQGrpcSourceSplitReader(new Configuration());
+ final IllegalStateException failure =
+ new IllegalStateException("Stream is already completed, no
further calls allowed");
+ reader.startWorkersWith(new FailingPopConsumer(failure, 1));
+ try {
+ assertThatThrownBy(() -> awaitFetchFailure(reader))
+ .isInstanceOf(FlinkRuntimeException.class)
+ .hasRootCause(failure);
+ } finally {
+ reader.close();
+ }
+ }
+
+ @Test
+ void transientReceiveFailureKeepsRetryingTest() throws Exception {
+ final RocketMQGrpcSourceSplitReader reader =
+ new RocketMQGrpcSourceSplitReader(new Configuration());
+ final FailingPopConsumer consumer =
+ new FailingPopConsumer(
+ new InternalErrorException(INTERNAL_ERROR, REQUEST_ID,
"internal error"),
+ 2);
+ reader.startWorkersWith(consumer);
+ try {
+ awaitReceiveAttempts(consumer, 3);
+ // The reader keeps working: no failure is surfaced to the fetcher
thread.
+ assertThat(reader.fetch().nextSplit()).isNull();
+ assertThat(consumer.attempts.get()).isGreaterThanOrEqualTo(3);
+ } finally {
+ reader.close();
+ }
+ }
+
+ @Test
+ void permanentFailureClassificationTest() {
+ assertThat(
+ RocketMQGrpcSourceSplitReader.isPermanentFailure(
+ new BadRequestException(
+ ILLEGAL_CONSUMER_GROUP, REQUEST_ID,
"lite group")))
+ .isTrue();
+ assertThat(
+ RocketMQGrpcSourceSplitReader.isPermanentFailure(
+ new UnauthorizedException(
+ UNAUTHORIZED, REQUEST_ID, "no
credential")))
+ .isTrue();
+ assertThat(
+ RocketMQGrpcSourceSplitReader.isPermanentFailure(
+ new NotFoundException(
+ CONSUMER_GROUP_NOT_FOUND, REQUEST_ID,
"group not found")))
+ .isTrue();
+ assertThat(
+ RocketMQGrpcSourceSplitReader.isPermanentFailure(
+ new UnsupportedException(UNSUPPORTED,
REQUEST_ID, "unsupported")))
+ .isTrue();
+
+ // FORBIDDEN is also the proxy's backpressure answer while a consumer
holds too many
+ // unacknowledged messages, so it must stay retryable.
+ assertThat(
+ RocketMQGrpcSourceSplitReader.isPermanentFailure(
+ new ForbiddenException(FORBIDDEN, REQUEST_ID,
"too many unacked")))
+ .isFalse();
+ assertThat(
+ RocketMQGrpcSourceSplitReader.isPermanentFailure(
+ new TooManyRequestsException(
+ TOO_MANY_REQUESTS, REQUEST_ID,
"throttled")))
+ .isFalse();
+ assertThat(
+ RocketMQGrpcSourceSplitReader.isPermanentFailure(
+ new InternalErrorException(INTERNAL_ERROR,
REQUEST_ID, "internal")))
+ .isFalse();
+ assertThat(
+ RocketMQGrpcSourceSplitReader.isPermanentFailure(
+ new ProxyTimeoutException(PROXY_TIMEOUT,
REQUEST_ID, "timeout")))
+ .isFalse();
+ assertThat(
+ RocketMQGrpcSourceSplitReader.isPermanentFailure(
+ new ClientException("unmapped failure")))
+ .isFalse();
+ }
+
+ /**
+ * Creating the consumer fails with an unchecked exception when the
endpoints are missing. It
+ * must fail the split assignment, and a later assignment must try again
rather than silently
+ * doing nothing because the reader still believes it is started.
+ */
+ @Test
+ void consumerCreationFailureFailsSplitAssignmentTest() throws Exception {
+ final RocketMQGrpcSourceSplitReader reader =
+ new RocketMQGrpcSourceSplitReader(new Configuration());
+ final SplitsChange<RocketMQGrpcSourceSplit> change =
+ new
SplitsAddition<>(Collections.singletonList(RocketMQGrpcSourceSplit.INSTANCE));
+ try {
+ assertThatThrownBy(() -> reader.handleSplitsChanges(change))
+ .isInstanceOf(FlinkRuntimeException.class)
+ .hasRootCauseInstanceOf(IllegalArgumentException.class)
+ .hasStackTraceContaining("endpoints");
+ assertThatThrownBy(() -> reader.handleSplitsChanges(change))
+ .isInstanceOf(FlinkRuntimeException.class);
+ } finally {
+ reader.close();
+ }
+ }
+
+ /**
+ * Calls {@link RocketMQGrpcSourceSplitReader#fetch()} until it fails or
the deadline passes.
+ */
+ private static void awaitFetchFailure(RocketMQGrpcSourceSplitReader
reader) throws Exception {
+ final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30);
+ while (System.nanoTime() < deadline) {
+ reader.fetch();
+ }
+ throw new AssertionError("fetch() did not fail within 30s");
+ }
+
+ private static void awaitReceiveAttempts(FailingPopConsumer consumer, int
attempts)
+ throws InterruptedException {
+ final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30);
+ while (consumer.attempts.get() < attempts && System.nanoTime() <
deadline) {
+ Thread.sleep(20L);
+ }
+ }
+
private static RocketMQReceiptHandle handle() {
return new RocketMQReceiptHandle(
"127.0.0.1:8081",
@@ -100,6 +277,48 @@ class RocketMQGrpcSourceSplitReaderTest {
1);
}
+ /**
+ * A {@link PopConsumer} whose {@code receive()} fails the first {@code
failures} calls with the
+ * given throwable and returns no message afterwards.
+ */
+ private static final class FailingPopConsumer implements PopConsumer {
+
+ private final Throwable failure;
+ private final int failures;
+ private final AtomicInteger attempts = new AtomicInteger();
+
+ private FailingPopConsumer(Throwable failure, int failures) {
+ this.failure = failure;
+ this.failures = failures;
+ }
+
+ @Override
+ public List<MessageView> receive(int maxMessageNum, Duration
invisibleDuration)
+ throws ClientException {
+ if (attempts.incrementAndGet() <= failures) {
+ if (failure instanceof RuntimeException) {
+ throw (RuntimeException) failure;
+ }
+ throw (ClientException) failure;
+ }
+ return Collections.emptyList();
+ }
+
+ @Override
+ public void ack(MessageView messageView) throws ClientException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void changeInvisibleDuration(MessageView messageView, Duration
invisibleDuration)
+ throws ClientException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void close() throws IOException {}
+ }
+
private static final class RecordingPopConsumer implements PopConsumer {
private final List<MessageView> acked = new ArrayList<>();
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 8db4bc6..120d27f 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
@@ -33,10 +33,14 @@ import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.catchThrowable;
/** Tests for {@link RocketMQGrpcDynamicTableFactory}. */
class RocketMQGrpcDynamicTableFactoryTest {
+ private static final String ACCESS_KEY_VALUE = "AK-TEST-VALUE";
+ private static final String SECRET_KEY_VALUE = "SK-TEST-VALUE";
+
private static final ResolvedSchema SCHEMA =
ResolvedSchema.of(Column.physical("f0", DataTypes.STRING()));
@@ -62,13 +66,15 @@ class RocketMQGrpcDynamicTableFactoryTest {
}
@Test
- void createLiteModeTableSourceTest() {
+ void createLiteModeTableSourceFailsTest() {
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);
+ assertThatThrownBy(() -> FactoryMocks.createTableSource(SCHEMA,
options))
+ .isInstanceOf(ValidationException.class)
+ .hasStackTraceContaining(
+ "lite' is not supported by the RocketMQ gRPC table
source");
}
@Test
@@ -112,4 +118,52 @@ class RocketMQGrpcDynamicTableFactoryTest {
assertThatThrownBy(() -> FactoryMocks.createTableSource(SCHEMA,
options))
.hasStackTraceContaining("bogus");
}
+
+ /**
+ * A missing required option must be reported by its key only. Credentials
set in the WITH
+ * clause end up in the JobManager log and in the REST exception history,
so the message the
+ * connector produces must not repeat any option value.
+ */
+ @Test
+ void missingSourceOptionIsReportedWithoutOptionValuesTest() {
+ final Map<String, String> options = baseOptions();
+ options.remove("endpoints");
+ options.put("source.consumer-group", "test-group");
+ options.put("access-key", ACCESS_KEY_VALUE);
+ options.put("secret-key", SECRET_KEY_VALUE);
+
+ final Throwable thrown =
+ catchThrowable(() -> FactoryMocks.createTableSource(SCHEMA,
options));
+
+ assertThat(thrown).isInstanceOf(ValidationException.class);
+ // Flink wraps every factory failure and lists the table options
itself; assert on the
+ // message the connector produced.
+ final Throwable cause = thrown.getCause();
+ assertThat(cause).isInstanceOf(ValidationException.class);
+ assertThat(cause.getMessage())
+ .startsWith("Missing required option")
+ .contains("endpoints")
+ .doesNotContain(ACCESS_KEY_VALUE)
+ .doesNotContain(SECRET_KEY_VALUE);
+ }
+
+ @Test
+ void missingSinkOptionIsReportedWithoutOptionValuesTest() {
+ final Map<String, String> options = baseOptions();
+ options.remove("topic");
+ options.put("access-key", ACCESS_KEY_VALUE);
+ options.put("secret-key", SECRET_KEY_VALUE);
+
+ final Throwable thrown =
+ catchThrowable(() -> FactoryMocks.createTableSink(SCHEMA,
options));
+
+ assertThat(thrown).isInstanceOf(ValidationException.class);
+ final Throwable cause = thrown.getCause();
+ assertThat(cause).isInstanceOf(ValidationException.class);
+ assertThat(cause.getMessage())
+ .startsWith("Missing required option")
+ .contains("topic")
+ .doesNotContain(ACCESS_KEY_VALUE)
+ .doesNotContain(SECRET_KEY_VALUE);
+ }
}
