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 fad67bf [ISSUE #141] fix(remoting): stop delivering the wrong data
silently (#145)
fad67bf is described below
commit fad67bf45d4d5f2f59acf16b22acb9e2d383d6b4
Author: lizhimins <[email protected]>
AuthorDate: Mon Oct 26 14:47:03 2026 +0800
[ISSUE #141] fix(remoting): stop delivering the wrong data silently (#145)
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.
---
.../rocketmq/common/config/RocketMQOptions.java | 1 -
.../rocketmq/common/filter/Sql92MessageFilter.java | 97 ++++++++++
.../connector/rocketmq/sink/InnerProducer.java | 11 ++
.../connector/rocketmq/sink/InnerProducerImpl.java | 109 +++++++++--
.../rocketmq/sink/RocketMQSinkOptions.java | 5 -
.../rocketmq/sink/committer/RocketMQCommitter.java | 3 +-
.../rocketmq/sink/writer/RocketMQWriter.java | 49 +++--
.../rocketmq/source/InnerConsumerImpl.java | 32 ++--
.../rocketmq/source/RocketMQSourceOptions.java | 14 +-
.../metrics/RocketMQSourceReaderMetrics.java | 137 +++++++++++++-
.../rocketmq/source/reader/MessageViewExt.java | 7 +-
.../source/reader/RocketMQRecordEmitter.java | 17 +-
.../source/reader/RocketMQSplitReader.java | 152 ++++++++++-----
.../deserializer/RowDeserializationSchema.java | 86 ++++-----
.../connectors/rocketmq/RocketMQSink.java | 26 +++
.../rocketmq/RocketMQSourceFunction.java | 22 ++-
.../RowKeyValueDeserializationSchema.java | 82 ++++----
.../rocketmq/table/RequiredOptionValidator.java | 56 ++++++
.../table/RocketMQDynamicTableSinkFactory.java | 1 +
.../table/RocketMQDynamicTableSourceFactory.java | 2 +-
.../rocketmq/table/RocketMQScanTableSource.java | 10 +-
.../common/filter/Sql92MessageFilterTest.java | 103 ++++++++++
.../writer/RocketMQWriterQueueSelectorTest.java | 158 ++++++++++++++++
.../metrics/RocketMQSourceReaderMetricsTest.java | 208 +++++++++++++++++++++
.../RowDeserializationSchemaOptionsTest.java | 144 ++++++++++++++
.../RocketMQDynamicTableSourceFactoryTest.java | 65 +++++++
.../rocketmq/table/RocketMQParameterChainTest.java | 6 -
27 files changed, 1386 insertions(+), 217 deletions(-)
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/common/config/RocketMQOptions.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/common/config/RocketMQOptions.java
index 3df6b41..9e7eef7 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/common/config/RocketMQOptions.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/common/config/RocketMQOptions.java
@@ -64,7 +64,6 @@ public class RocketMQOptions {
.noDefaultValue()
.withDescription("RocketMQ instance namespace");
- /** 这里不知道对轨迹功能有没有影响, 待验证 */
public static final ConfigOption<AccessChannel> OPTIONAL_ACCESS_CHANNEL =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "channel")
.enumType(AccessChannel.class)
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/common/filter/Sql92MessageFilter.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/common/filter/Sql92MessageFilter.java
new file mode 100644
index 0000000..3020d8f
--- /dev/null
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/common/filter/Sql92MessageFilter.java
@@ -0,0 +1,97 @@
+/*
+ * 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.common.filter;
+
+import org.apache.rocketmq.common.filter.ExpressionType;
+import org.apache.rocketmq.filter.FilterFactory;
+import org.apache.rocketmq.filter.expression.EvaluationContext;
+import org.apache.rocketmq.filter.expression.Expression;
+import org.apache.rocketmq.filter.expression.MQFilterException;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Evaluates a SQL92 expression against the properties of a single message.
+ *
+ * <p>{@link org.apache.rocketmq.client.consumer.DefaultLitePullConsumer} can
only carry a tag
+ * sub-expression while queues are assigned explicitly, and assigning queues
explicitly is what
+ * keeps split assignment under the control of the Flink enumerator. A SQL92
filter therefore has to
+ * be evaluated here rather than on the broker.
+ *
+ * <p>The expression is compiled once by the reader that owns this filter.
Compilation uses the same
+ * {@code rocketmq-filter} implementation the broker uses, so predicate
semantics, including the
+ * coercion between string properties and numeric literals, match server side
SQL92 filtering.
+ */
+public final class Sql92MessageFilter {
+
+ private final String expressionString;
+ private final Expression expression;
+
+ public Sql92MessageFilter(String expressionString) {
+ this.expressionString = expressionString;
+ try {
+ this.expression =
+
FilterFactory.INSTANCE.get(ExpressionType.SQL92).compile(expressionString);
+ } catch (MQFilterException e) {
+ throw new IllegalArgumentException(
+ "Invalid SQL92 filter expression '" + expressionString +
"'", e);
+ }
+ }
+
+ /**
+ * Returns whether the message satisfies the expression.
+ *
+ * <p>A message whose properties cannot be evaluated, for instance because
the expression
+ * dereferences a missing property in an unsupported way, does not satisfy
the predicate. This
+ * mirrors the broker, which drops such messages instead of failing the
pull request.
+ */
+ public boolean matches(Map<String, String> properties) {
+ try {
+ return Boolean.TRUE.equals(expression.evaluate(new
PropertiesContext(properties)));
+ } catch (Exception e) {
+ return false;
+ }
+ }
+
+ public String getExpressionString() {
+ return expressionString;
+ }
+
+ /** Adapts the immutable string properties of a message to the filter
evaluation context. */
+ private static final class PropertiesContext implements EvaluationContext {
+
+ private final Map<String, String> properties;
+
+ private PropertiesContext(Map<String, String> properties) {
+ this.properties = properties == null ? Collections.emptyMap() :
properties;
+ }
+
+ @Override
+ public Object get(String name) {
+ return properties.get(name);
+ }
+
+ @Override
+ public Map<String, Object> keyValues() {
+ return Collections.unmodifiableMap(new HashMap<String,
Object>(properties));
+ }
+ }
+}
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/InnerProducer.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/InnerProducer.java
index 9d1669b..aef6614 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/InnerProducer.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/InnerProducer.java
@@ -18,6 +18,7 @@
package org.apache.flink.connector.rocketmq.sink;
import org.apache.flink.connector.rocketmq.sink.committer.SendCommittable;
+import
org.apache.flink.streaming.connectors.rocketmq.common.selector.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
@@ -52,6 +53,16 @@ public interface InnerProducer extends AutoCloseable {
*/
CompletableFuture<SendResult> send(Message message);
+ /**
+ * Sends the message to a queue chosen by the given selector.
+ *
+ * @param message the message to be sent
+ * @param selector picks the queue among the queues of the topic
+ * @param arg the sharding argument handed to the selector, usually the
record being written
+ * @return a Future for the send operation that returns a SendResult object
+ */
+ CompletableFuture<SendResult> send(Message message, MessageQueueSelector
selector, Object arg);
+
/**
* Sends the message to the messaging system and returns a Future for the
send operation.
*
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/InnerProducerImpl.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/InnerProducerImpl.java
index 84581b2..6595f7a 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/InnerProducerImpl.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/InnerProducerImpl.java
@@ -20,6 +20,8 @@ package org.apache.flink.connector.rocketmq.sink;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.rocketmq.common.config.RocketMQOptions;
import org.apache.flink.connector.rocketmq.sink.committer.SendCommittable;
+import
org.apache.flink.streaming.connectors.rocketmq.common.selector.MessageQueueSelector;
+import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.StringUtils;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
@@ -40,6 +42,7 @@ import
org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.remoting.protocol.NamespaceUtil;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RequestCode;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
import
org.apache.rocketmq.remoting.protocol.header.EndTransactionRequestHeader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -209,7 +212,7 @@ public class InnerProducerImpl implements InnerProducer {
() -> {
try {
SendResult sendResult = producer.send(message);
- LOG.info(
+ LOG.debug(
"Send message successfully, topic={},
messageId={}",
message.getTopic(),
sendResult.getMsgId());
@@ -222,6 +225,27 @@ public class InnerProducerImpl implements InnerProducer {
Runnable::run);
}
+ @Override
+ public CompletableFuture<SendResult> send(
+ Message message, MessageQueueSelector selector, Object arg) {
+ return CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ SendResult sendResult = producer.send(message,
selector, arg);
+ LOG.debug(
+ "Send message successfully, topic={},
messageId={}, queue={}",
+ message.getTopic(),
+ sendResult.getMsgId(),
+ sendResult.getMessageQueue());
+ return sendResult;
+ } catch (Exception e) {
+ LOG.error("Failed to send message, topic={}",
message.getTopic(), e);
+ throw new RuntimeException(e);
+ }
+ },
+ Runnable::run);
+ }
+
@Override
public CompletableFuture<SendResult> sendMessageInTransaction(Message
message) {
return CompletableFuture.supplyAsync(
@@ -278,8 +302,28 @@ public class InnerProducerImpl implements InnerProducer {
final String brokerName =
this.mqClientInstance.getBrokerNameFromMessageQueue(
producer.queueWithNamespace(sendCommittable.getMessageQueue()));
- final String brokerAddress =
-
this.mqClientInstance.findBrokerAddressInPublish(brokerName);
+ String brokerAddress =
this.mqClientInstance.findBrokerAddressInPublish(brokerName);
+ if (brokerAddress == null) {
+ // The Committer owns a producer that never sends anything, so
its publish route
+ // table stays empty and findBrokerAddressInPublish() returns
null. invokeSync()
+ // then falls back to a name server channel, which cannot
commit the half message,
+ // and the reply is not a SUCCESS the caller would notice --
every record of the
+ // checkpoint is lost without a single error. Refresh the
route for the topic the
+ // half message belongs to before resolving again.
+ LOG.warn(
+ "Publish route for {} is not loaded yet, refreshing
before ending transaction",
+ brokerName);
+ this.mqClientInstance.updateTopicRouteInfoFromNameServer(
+ sendCommittable.getTopic());
+ brokerAddress =
this.mqClientInstance.findBrokerAddressInPublish(brokerName);
+ }
+ if (brokerAddress == null) {
+ throw new IllegalStateException(
+ "Cannot resolve the address of broker "
+ + brokerName
+ + " for topic "
+ + sendCommittable.getTopic());
+ }
EndTransactionRequestHeader requestHeader = new
EndTransactionRequestHeader();
requestHeader.setTransactionId(sendCommittable.getTransactionId());
@@ -304,16 +348,38 @@ public class InnerProducerImpl implements InnerProducer {
break;
}
- if (sendCommittable.getMessageOffset() != -1L) {
- this.endTransaction(
- brokerAddress, requestHeader, "",
this.producer.getSendMsgTimeout());
- } else {
- LOG.error(
- "Convert message physical offset error, msgId={}",
- sendCommittable.getMsgId());
+ if (sendCommittable.getMessageOffset() == -1L) {
+ throw new IllegalStateException(
+ "Convert message physical offset error, msgId="
+ + sendCommittable.getMsgId()
+ + ", offsetMsgId="
+ + sendCommittable.getOffsetMsgId());
}
+ LOG.debug(
+ "End transaction, state={}, brokerName={},
brokerAddress={}, transId={}, "
+ + "commitLogOffset={}, tranStateTableOffset={}",
+ transactionResult,
+ brokerName,
+ brokerAddress,
+ sendCommittable.getTransactionId(),
+ sendCommittable.getMessageOffset(),
+ sendCommittable.getQueueOffset());
+ this.endTransaction(
+ brokerAddress, requestHeader, "",
this.producer.getSendMsgTimeout());
} catch (Exception e) {
- LOG.error("Try end transaction error", e);
+ // Swallowing the failure here lets the CompletableFuture complete
normally, so the
+ // Committer reports success and Flink drops the committable: the
half message is
+ // then never committed and the record is lost without any error
anywhere.
+ throw new FlinkRuntimeException(
+ "End transaction failed, state="
+ + transactionResult
+ + ", topic="
+ + sendCommittable.getTopic()
+ + ", msgId="
+ + sendCommittable.getMsgId()
+ + ", transId="
+ + sendCommittable.getTransactionId(),
+ e);
}
}
@@ -326,10 +392,23 @@ public class InnerProducerImpl implements InnerProducer {
RemotingCommand request =
RemotingCommand.createRequestCommand(RequestCode.END_TRANSACTION,
requestHeader);
request.setRemark(remark);
- this.mqClientInstance
- .getMQClientAPIImpl()
- .getRemotingClient()
- .invokeSync(address, request, timeoutMillis);
+ RemotingCommand response =
+ this.mqClientInstance
+ .getMQClientAPIImpl()
+ .getRemotingClient()
+ .invokeSync(address, request, timeoutMillis);
+ if (response == null) {
+ throw new RemotingException("END_TRANSACTION got a null response
from " + address);
+ }
+ if (response.getCode() != ResponseCode.SUCCESS) {
+ throw new RemotingException(
+ "END_TRANSACTION rejected by "
+ + address
+ + ", code="
+ + response.getCode()
+ + ", remark="
+ + response.getRemark());
+ }
}
@Override
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/RocketMQSinkOptions.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/RocketMQSinkOptions.java
index a220e50..99bb5d0 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/RocketMQSinkOptions.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/RocketMQSinkOptions.java
@@ -40,11 +40,6 @@ public class RocketMQSinkOptions extends RocketMQOptions {
.noDefaultValue()
.withDescription("The name of the persist topic");
- public static final ConfigOption<String> SERIALIZE_FORMAT =
- ConfigOptions.key(PRODUCER_PREFIX + "serialize.format")
- .stringType()
- .defaultValue("json");
-
public static final ConfigOption<String> DELIVERY_GUARANTEE =
ConfigOptions.key(PRODUCER_PREFIX + "delivery.guarantee")
.stringType()
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/committer/RocketMQCommitter.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/committer/RocketMQCommitter.java
index 4262b7b..77d3327 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/committer/RocketMQCommitter.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/committer/RocketMQCommitter.java
@@ -63,7 +63,8 @@ public class RocketMQCommitter implements
Committer<SendCommittable>, Cloneable
throwable -> {
LOG.error(
"Commit message error,
committable={}",
- committable);
+ committable,
+ throwable);
request.signalFailedWithKnownReason(throwable);
return null;
});
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/writer/RocketMQWriter.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/writer/RocketMQWriter.java
index ae1b4e6..59ad65c 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/writer/RocketMQWriter.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/writer/RocketMQWriter.java
@@ -17,6 +17,7 @@
package org.apache.flink.connector.rocketmq.sink.writer;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.connector.sink2.Sink;
import org.apache.flink.api.connector.sink2.TwoPhaseCommittingSink;
@@ -37,12 +38,15 @@ import org.apache.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
@@ -69,6 +73,17 @@ public class RocketMQWriter<IN>
MessageQueueSelector messageQueueSelector,
RocketMQSerializationSchema<IN> serializationSchema,
Sink.InitContext initContext) {
+ this(configuration, messageQueueSelector, serializationSchema,
initContext, null);
+ }
+
+ /** @param producer the producer to send with, or null to create and start
a real one */
+ @VisibleForTesting
+ RocketMQWriter(
+ Configuration configuration,
+ MessageQueueSelector messageQueueSelector,
+ RocketMQSerializationSchema<IN> serializationSchema,
+ Sink.InitContext initContext,
+ @Nullable InnerProducer producer) {
this.deliveryGuarantee =
DeliveryGuarantee.valueOf(
@@ -86,8 +101,12 @@ public class RocketMQWriter<IN>
} catch (Exception e) {
throw new FlinkRuntimeException("Failed to open serialization
schema", e);
}
- this.producer = new InnerProducerImpl(configuration);
- this.producer.start();
+ if (producer == null) {
+ this.producer = new InnerProducerImpl(configuration);
+ this.producer.start();
+ } else {
+ this.producer = producer;
+ }
}
@Override
@@ -99,17 +118,23 @@ public class RocketMQWriter<IN>
if (deliveryGuarantee == DeliveryGuarantee.EXACTLY_ONCE) {
writeInTransaction(message);
- } else {
- pendingRecords.incrementAndGet();
- producer.send(message)
- .whenComplete(
- (sendResult, throwable) -> {
- if (throwable != null) {
- asyncSendException.compareAndSet(null,
throwable);
- }
- pendingRecords.decrementAndGet();
- });
+ return;
}
+
+ pendingRecords.incrementAndGet();
+ // The element is handed to the selector as its sharding argument,
which is what lets a
+ // HashMessageQueueSelector route records that share a key to the same
queue.
+ CompletableFuture<SendResult> sendFuture =
+ messageQueueSelector == null
+ ? producer.send(message)
+ : producer.send(message, messageQueueSelector,
element);
+ sendFuture.whenComplete(
+ (sendResult, throwable) -> {
+ if (throwable != null) {
+ asyncSendException.compareAndSet(null, throwable);
+ }
+ pendingRecords.decrementAndGet();
+ });
}
private void writeInTransaction(Message message) throws IOException {
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/InnerConsumerImpl.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/InnerConsumerImpl.java
index 1149de6..ef621cc 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/InnerConsumerImpl.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/InnerConsumerImpl.java
@@ -89,7 +89,6 @@ public class InnerConsumerImpl implements InnerConsumer {
this.consumer = new DefaultLitePullConsumer();
}
- // Enable message trace if configured (P1 fix — previously a ghost
parameter)
this.consumer.setEnableMsgTrace(enableTrace);
if (traceTopic != null) {
this.consumer.setCustomizedTraceTopic(traceTopic);
@@ -100,7 +99,6 @@ public class InnerConsumerImpl implements InnerConsumer {
this.consumer.setNamesrvAddr(endPoints);
this.consumer.setConsumerGroup(groupId);
- this.consumer.setAutoCommit(false);
this.consumer.setVipChannelEnabled(false);
this.consumer.setInstanceName(
String.join(
@@ -109,29 +107,33 @@ public class InnerConsumerImpl implements InnerConsumer {
groupId,
UUID.randomUUID().toString()));
- // Apply pull batch size from config (P1 fix — previously a ghost
parameter)
long pullBatchSize =
configuration.getLong(RocketMQSourceOptions.PULL_BATCH_SIZE);
if (pullBatchSize > 0) {
this.consumer.setPullBatchSize((int) pullBatchSize);
}
+ long pullTimeoutMillis =
configuration.getLong(RocketMQSourceOptions.PULL_TIMEOUT_MILLIS);
+ if (pullTimeoutMillis > 0) {
+ this.consumer.setConsumerPullTimeoutMillis(pullTimeoutMillis);
+ }
- // Apply namespace from config (P2 fix)
String namespace = configuration.getString(RocketMQOptions.NAMESPACE);
if (namespace != null && !namespace.isEmpty()) {
this.consumer.setNamespace(namespace);
}
- // Log auto-commit configuration for observability (P1 fix)
+ // Offsets are driven by Flink checkpoints; the client side
auto-commit below only affects
+ // what broker side consumer progress reports.
boolean autoCommit =
configuration.getBoolean(RocketMQSourceOptions.AUTO_COMMIT_OFFSET);
long commitInterval =
configuration.getLong(RocketMQSourceOptions.AUTO_COMMIT_OFFSET_INTERVAL);
+ this.consumer.setAutoCommit(autoCommit);
+ this.consumer.setAutoCommitIntervalMillis(commitInterval);
LOG.info(
"Consumer offset commit config: autoCommit={},
commitInterval={}ms, checkpointCommit={}",
autoCommit,
commitInterval,
configuration.getBoolean(RocketMQSourceOptions.COMMIT_OFFSETS_ON_CHECKPOINT));
- // Apply common client parameters (P2 fix)
if (configuration.contains(RocketMQOptions.OPTIONAL_ACCESS_CHANNEL)) {
this.consumer.setAccessChannel(
configuration.get(RocketMQOptions.OPTIONAL_ACCESS_CHANNEL));
@@ -285,16 +287,14 @@ public class InnerConsumerImpl implements InnerConsumer {
@Override
public void wakeup() {
- // wakeup long polling
- try {
- Set<MessageQueue> assignment = this.consumer.assignment();
- if (assignment != null) {
- this.consumer.pause(assignment);
- }
- } catch (MQClientException e) {
- LOG.warn("Consume wakeup long polling failed", e);
- throw new RuntimeException(e);
- }
+ // Deliberately does not pause the assigned queues.
DefaultLitePullConsumer offers no
+ // interruptible poll, but poll(Duration) is already bounded by a
short timeout, so the
+ // fetch loop never blocks long enough to need forcing. Pausing here
without a matching
+ // resume() would permanently stop delivering records while the client
keeps pulling and
+ // advancing offsets in the background -- silent message loss with
nothing logged.
+ // Use pause(Collection)/resume(Collection) as a pair if queue pausing
is ever needed.
+ LOG.debug(
+ "Wakeup requested; relying on the bounded poll timeout instead
of pausing queues");
}
@Override
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/RocketMQSourceOptions.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/RocketMQSourceOptions.java
index 4683184..eb5f06a 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/RocketMQSourceOptions.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/RocketMQSourceOptions.java
@@ -73,11 +73,6 @@ public class RocketMQSourceOptions extends RocketMQOptions {
.longType()
.defaultValue(RocketMQConfig.DEFAULT_START_MESSAGE_OFFSET);
- public static final ConfigOption<String> OPTIONAL_STARTUP_OFFSET_STRATEGY =
- ConfigOptions.key(CONSUMER_PREFIX + "startup.offset.strategy")
- .stringType()
- .noDefaultValue();
-
public static final ConfigOption<String> OPTIONAL_STARTUP_OFFSET_DATE =
ConfigOptions.key(CONSUMER_PREFIX + "startup.offset.date")
.stringType()
@@ -152,11 +147,16 @@ public class RocketMQSourceOptions extends
RocketMQOptions {
"The maximum wait time for a response from the
broker "
+ "in long polling by the client");
- /** for auto commit offset to rocketmq server */
+ /**
+ * Lets the RocketMQ client commit the offset of everything it has polled
on its own timer. Off
+ * by default: Flink commits the offsets of checkpointed state instead,
see {@link
+ * #COMMIT_OFFSETS_ON_CHECKPOINT}. Broker side offsets never decide where
a restored job starts
+ * reading, so enabling this only changes what consumer progress
monitoring reports.
+ */
public static final ConfigOption<Boolean> AUTO_COMMIT_OFFSET =
ConfigOptions.key(CONSUMER_PREFIX + "offset.commit.auto")
.booleanType()
- .defaultValue(true)
+ .defaultValue(false)
.withDescription("The setting for automatic commit of
offset");
public static final ConfigOption<Long> AUTO_COMMIT_OFFSET_INTERVAL =
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/metrics/RocketMQSourceReaderMetrics.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/metrics/RocketMQSourceReaderMetrics.java
index f21e12a..7c62f58 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/metrics/RocketMQSourceReaderMetrics.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/metrics/RocketMQSourceReaderMetrics.java
@@ -1,10 +1,11 @@
/*
- * 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
+ * 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
*
@@ -18,10 +19,29 @@
package org.apache.flink.connector.rocketmq.source.metrics;
import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Gauge;
+import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.metrics.groups.SourceReaderMetricGroup;
import org.apache.rocketmq.common.message.MessageQueue;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Metrics of the RocketMQ source reader.
+ *
+ * <p>Offsets are exposed per message queue under the {@code
RocketmqSourceReader} group, keyed by
+ * topic and queue id, so a reader that owns several queues can be inspected
queue by queue. The lag
+ * of every queue is additionally summed into {@code pendingRecords}, which is
what Flink's
+ * backpressure and idleness tooling reads.
+ *
+ * <p>Lag needs the maximum offset of a queue, which only the broker knows.
The reader therefore
+ * pushes the values in with {@link #updateMaxOffset} whenever it refreshes
them, and the gauges
+ * here only ever report the last known value; reading a metric never triggers
a remote call.
+ */
@PublicEvolving
public class RocketMQSourceReaderMetrics {
@@ -38,9 +58,108 @@ public class RocketMQSourceReaderMetrics {
public static final String BYTES_CONSUMED_TOTAL = "bytes-consumed-total";
public static final String RECORDS_LAG = "records-lag";
- public RocketMQSourceReaderMetrics(SourceReaderMetricGroup
sourceReaderMetricGroup) {}
+ private final MetricGroup readerGroup;
+ private final Counter commitsSucceeded;
+ private final Counter commitsFailed;
+ private final Counter bytesConsumedTotal;
+ private final Map<MessageQueue, QueueMetrics> queues = new
ConcurrentHashMap<>();
+
+ public RocketMQSourceReaderMetrics(SourceReaderMetricGroup
sourceReaderMetricGroup) {
+ this.readerGroup =
sourceReaderMetricGroup.addGroup(ROCKETMQ_SOURCE_READER_METRIC_GROUP);
+ this.commitsSucceeded =
readerGroup.counter(COMMITS_SUCCEEDED_METRIC_COUNTER);
+ this.commitsFailed =
readerGroup.counter(COMMITS_FAILED_METRIC_COUNTER);
+
+ MetricGroup fetchManagerGroup =
+ sourceReaderMetricGroup
+ .addGroup(ROCKETMQ_CONSUMER_METRIC_GROUP)
+ .addGroup(CONSUMER_FETCH_MANAGER_GROUP);
+ this.bytesConsumedTotal =
fetchManagerGroup.counter(BYTES_CONSUMED_TOTAL);
+ fetchManagerGroup.gauge(RECORDS_LAG, (Gauge<Long>) this::getTotalLag);
+
+ sourceReaderMetricGroup.setPendingRecordsGauge((Gauge<Long>)
this::getTotalLag);
+ }
+
+ /** Starts reporting the offsets of a newly assigned queue. */
+ public void registerNewMessageQueue(MessageQueue messageQueue) {
+ queues.computeIfAbsent(messageQueue, this::createQueueMetrics);
+ }
+
+ /**
+ * Stops aggregating a queue that is no longer assigned to this reader.
+ *
+ * <p>{@link MetricGroup} offers no way to deregister from a connector, so
the gauges of the
+ * queue stay visible and keep reporting their last known values. The
queue count of a topic is
+ * finite, which bounds how many groups a reader can accumulate.
+ */
+ public void unregisterMessageQueue(MessageQueue messageQueue) {
+ queues.remove(messageQueue);
+ }
+
+ /** Records how far the reader has consumed a queue. */
+ public void setCurrentOffset(MessageQueue messageQueue, long offset) {
+ QueueMetrics metrics = queues.get(messageQueue);
+ if (metrics != null) {
+ metrics.currentOffset.set(offset);
+ }
+ }
+
+ /** Records the offset that was successfully committed to the broker for a
queue. */
+ public void setCommittedOffset(MessageQueue messageQueue, long offset) {
+ QueueMetrics metrics = queues.get(messageQueue);
+ if (metrics != null) {
+ metrics.committedOffset.set(offset);
+ }
+ }
+
+ /**
+ * Records the maximum offset the broker holds for a queue and derives its
lag from the offset
+ * the reader has consumed so far.
+ */
+ public void updateMaxOffset(MessageQueue messageQueue, long maxOffset) {
+ QueueMetrics metrics = queues.get(messageQueue);
+ if (metrics != null) {
+ long currentOffset = metrics.currentOffset.get();
+ metrics.lag.set(currentOffset < 0L ? 0L : Math.max(0L, maxOffset -
currentOffset));
+ }
+ }
+
+ public void recordBytesConsumed(long bytes) {
+ bytesConsumedTotal.inc(bytes);
+ }
+
+ public void recordCommitSucceeded() {
+ commitsSucceeded.inc();
+ }
+
+ public void recordCommitFailed() {
+ commitsFailed.inc();
+ }
+
+ private QueueMetrics createQueueMetrics(MessageQueue messageQueue) {
+ MetricGroup queueGroup =
+ readerGroup
+ .addGroup(TOPIC_GROUP, messageQueue.getTopic())
+ .addGroup(QUEUE_GROUP,
String.valueOf(messageQueue.getQueueId()));
+ QueueMetrics metrics = new QueueMetrics();
+ queueGroup.gauge(CURRENT_OFFSET_METRIC_GAUGE, (Gauge<Long>)
metrics.currentOffset::get);
+ queueGroup.gauge(COMMITTED_OFFSET_METRIC_GAUGE, (Gauge<Long>)
metrics.committedOffset::get);
+ queueGroup.gauge(RECORDS_LAG, (Gauge<Long>) metrics.lag::get);
+ return metrics;
+ }
+
+ private long getTotalLag() {
+ long total = 0L;
+ for (QueueMetrics metrics : queues.values()) {
+ total += metrics.lag.get();
+ }
+ return total;
+ }
- public void registerNewMessageQueue(MessageQueue messageQueue) {}
+ /** The mutable state behind the gauges of a single queue. */
+ private static final class QueueMetrics {
- public void unregisterMessageQueue(MessageQueue messageQueue) {}
+ private final AtomicLong currentOffset = new AtomicLong(-1L);
+ private final AtomicLong committedOffset = new AtomicLong(-1L);
+ private final AtomicLong lag = new AtomicLong(0L);
+ }
}
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/reader/MessageViewExt.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/reader/MessageViewExt.java
index b289f20..e48ca5e 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/reader/MessageViewExt.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/reader/MessageViewExt.java
@@ -17,6 +17,7 @@
package org.apache.flink.connector.rocketmq.source.reader;
+import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.ArrayList;
@@ -26,7 +27,11 @@ import java.util.Map;
public class MessageViewExt implements MessageView {
- private static final String KEY_SEPARATE = "";
+ // Message.getKeys() returns the keys joined by MessageConst.KEY_SEPARATOR
(a single space) --
+ // that is what Message.setKeys(Collection) produces via String.join.
Splitting on anything
+ // else is wrong: splitting on "" yields one element per character, so a
message whose key is
+ // "smoke-11" would report eight keys ("s","m","o","k","e","-","1","1")
instead of one.
+ private static final String KEY_SEPARATE = MessageConst.KEY_SEPARATOR;
private final String messageId;
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/reader/RocketMQRecordEmitter.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/reader/RocketMQRecordEmitter.java
index 7ba42d6..12afce2 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/reader/RocketMQRecordEmitter.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/reader/RocketMQRecordEmitter.java
@@ -42,14 +42,23 @@ public class RocketMQRecordEmitter<T>
MessageView element, SourceOutput<T> output,
RocketMQSourceSplitState splitState)
throws IOException {
+ sourceOutputWrapper.setSourceOutput(output);
+ sourceOutputWrapper.setTimestamp(element.getEventTime());
try {
- sourceOutputWrapper.setSourceOutput(output);
- sourceOutputWrapper.setTimestamp(element.getEventTime());
deserializationSchema.deserialize(element, sourceOutputWrapper);
- splitState.setCurrentOffset(element.getQueueOffset() + 1);
} catch (Exception e) {
- throw new IOException("Failed to deserialize message due to", e);
+ // Naming the record makes a poison message identifiable from the
exception alone;
+ // otherwise the only way to find it is to replay the whole queue.
+ throw new IOException(
+ String.format(
+ "Failed to deserialize message %s from topic %s,
queue %d, offset %d",
+ element.getMessageId(),
+ element.getTopic(),
+ element.getQueueId(),
+ element.getQueueOffset()),
+ e);
}
+ splitState.setCurrentOffset(element.getQueueOffset() + 1);
}
private static class SourceOutputWrapper<T> implements Collector<T> {
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/reader/RocketMQSplitReader.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/reader/RocketMQSplitReader.java
index b8349e5..17c7e82 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/reader/RocketMQSplitReader.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/reader/RocketMQSplitReader.java
@@ -28,6 +28,7 @@ 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.common.config.RocketMQOptions;
import org.apache.flink.connector.rocketmq.common.event.SourceInitAssignEvent;
+import org.apache.flink.connector.rocketmq.common.filter.Sql92MessageFilter;
import org.apache.flink.connector.rocketmq.common.lock.SpinLock;
import org.apache.flink.connector.rocketmq.source.InnerConsumer;
import org.apache.flink.connector.rocketmq.source.InnerConsumerImpl;
@@ -48,7 +49,9 @@ import javax.annotation.Nullable;
import java.io.IOException;
import java.time.Duration;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
@@ -59,6 +62,7 @@ import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.stream.Collectors;
/**
* A {@link SplitReader} implementation that reads records from RocketMQ
partitions.
@@ -70,6 +74,10 @@ public class RocketMQSplitReader<T> implements
SplitReader<MessageView, RocketMQ
private static final Logger LOG =
LoggerFactory.getLogger(RocketMQSplitReader.class);
+ // The maximum offset of a queue is only known to the broker, so the lag
is refreshed on a
+ // coarse interval instead of on every fetch.
+ private static final long LAG_REFRESH_INTERVAL_MS = 10_000L;
+
private volatile boolean wakeup = false;
private final InnerConsumer consumer;
@@ -87,12 +95,15 @@ public class RocketMQSplitReader<T> implements
SplitReader<MessageView, RocketMQ
// Init status : true-finish init; false-not finish init
private boolean initFinish;
+ private long lastLagRefreshTime;
private final RocketMQRecordsWithSplitIds<MessageView> recordsWithSplitIds;
private final SpinLock lock;
- // Client-side filter configuration (P0 fix — tag/sql were dead in new API
path)
- private final String filterTag;
- private final String filterSql;
+ // Tag and SQL92 filters are evaluated here rather than on the broker: the
lite-pull consumer
+ // can only carry a tag sub-expression while queues are assigned
explicitly, and explicit
+ // assignment is what lets the enumerator own split allocation. See
Sql92MessageFilter.
+ private final Set<String> filterTags;
+ private final Sql92MessageFilter sqlFilter;
public RocketMQSplitReader(
Configuration configuration,
@@ -116,14 +127,18 @@ public class RocketMQSplitReader<T> implements
SplitReader<MessageView, RocketMQ
this.commitOffsetsOnCheckpoint =
configuration.getBoolean(RocketMQSourceOptions.COMMIT_OFFSETS_ON_CHECKPOINT);
- // Read filter configuration for client-side filtering (P0 fix)
- this.filterTag =
configuration.getString(RocketMQSourceOptions.OPTIONAL_TAG);
- this.filterSql = configuration.get(RocketMQSourceOptions.OPTIONAL_SQL);
- if (filterSql != null) {
- LOG.warn(
- "SQL filter expression '{}' is set but client-side SQL
evaluation "
- + "is not supported with assign() mode. Only
tag-based filtering will be applied.",
- filterSql);
+ this.filterTags =
+
parseFilterTags(configuration.getString(RocketMQSourceOptions.OPTIONAL_TAG));
+ String sqlExpression =
configuration.get(RocketMQSourceOptions.OPTIONAL_SQL);
+ this.sqlFilter =
+ sqlExpression == null || sqlExpression.trim().isEmpty()
+ ? null
+ : new Sql92MessageFilter(sqlExpression);
+ if (sqlFilter != null) {
+ LOG.info(
+ "Evaluating SQL92 filter '{}' on the client side; assign
mode cannot push it "
+ + "down to the broker",
+ sqlFilter.getExpressionString());
}
}
@@ -145,16 +160,20 @@ public class RocketMQSplitReader<T> implements
SplitReader<MessageView, RocketMQ
Duration.ofMillis(this.configuration.getLong(RocketMQOptions.POLL_TIMEOUT));
List<MessageView> messageExtList = consumer.poll(duration);
for (MessageView messageView : messageExtList) {
- // Client-side tag filtering (P0 fix)
+ MessageQueue messageQueue =
+ new MessageQueue(
+ messageView.getTopic(),
+ messageView.getBrokerName(),
+ messageView.getQueueId());
+ // Records dropped by a filter still advance the queue, so
they are counted here
+ // rather than after the filter.
+ rocketmqSourceReaderMetrics.setCurrentOffset(
+ messageQueue, messageView.getQueueOffset() + 1);
+
rocketmqSourceReaderMetrics.recordBytesConsumed(messageView.getStoreSize());
if (!matchesFilter(messageView)) {
continue;
}
- String splitId =
- UtilAll.getSplitId(
- new MessageQueue(
- messageView.getTopic(),
- messageView.getBrokerName(),
- messageView.getQueueId()));
+ String splitId = UtilAll.getSplitId(messageQueue);
recordsWithSplitIds.recordsForSplit(splitId).add(messageView);
if
(this.configuration.getBoolean(RocketMQSourceOptions.GLOBAL_DEBUG_MODE)) {
LOG.info(
@@ -164,6 +183,7 @@ public class RocketMQSplitReader<T> implements
SplitReader<MessageView, RocketMQ
}
}
recordsWithSplitIds.prepareForRead();
+ refreshLagIfDue();
} catch (Exception e) {
if (wakeup) {
LOG.debug("Reader fetch interrupted by wakeup", e);
@@ -257,37 +277,83 @@ public class RocketMQSplitReader<T> implements
SplitReader<MessageView, RocketMQ
}
}
+ /** Returns whether the message satisfies the configured tag and SQL92
filters. */
+ private boolean matchesFilter(MessageView messageView) {
+ if (!filterTags.isEmpty() &&
!filterTags.contains(messageView.getTag())) {
+ return false;
+ }
+ return sqlFilter == null ||
sqlFilter.matches(messageView.getProperties());
+ }
+
/**
- * Client-side message filtering. Applies tag-based filtering for
consumers using assign() mode
- * where server-side MessageSelector is not available.
+ * Splits a {@code "tagA||tagB"} sub-expression into the set of accepted
tags. An absent or
+ * {@code "*"} expression accepts every tag and yields an empty set.
*/
- private boolean matchesFilter(MessageView messageView) {
- if (filterTag != null && !"*".equals(filterTag) &&
!filterTag.isEmpty()) {
- String messageTag = messageView.getTag();
- if (messageTag == null) {
- return false;
- }
- // Support "||" separated multi-tag filter expression
- String[] filterTags = filterTag.split("\\|\\|");
- boolean matched = false;
- for (String ft : filterTags) {
- if (ft.trim().equals(messageTag)) {
- matched = true;
- break;
- }
- }
- if (!matched) {
- return false;
- }
+ private static Set<String> parseFilterTags(@Nullable String tagExpression)
{
+ if (tagExpression == null) {
+ return Collections.emptySet();
}
- return true;
+ String trimmed = tagExpression.trim();
+ if (trimmed.isEmpty() || "*".equals(trimmed)) {
+ return Collections.emptySet();
+ }
+ return Arrays.stream(trimmed.split("\\|\\|"))
+ .map(String::trim)
+ .filter(tag -> !tag.isEmpty())
+ .collect(Collectors.toSet());
}
public void notifyCheckpointComplete(Map<MessageQueue, Long>
offsetsToCommit) {
- if (offsetsToCommit != null) {
- for (Map.Entry<MessageQueue, Long> entry :
offsetsToCommit.entrySet()) {
- consumer.commitOffset(entry.getKey(), entry.getValue());
- }
+ if (offsetsToCommit == null) {
+ return;
+ }
+ for (Map.Entry<MessageQueue, Long> entry : offsetsToCommit.entrySet())
{
+ MessageQueue messageQueue = entry.getKey();
+ long offset = entry.getValue();
+ consumer.commitOffset(messageQueue, offset)
+ .whenComplete(
+ (ignored, error) -> {
+ if (error == null) {
+
rocketmqSourceReaderMetrics.recordCommitSucceeded();
+
rocketmqSourceReaderMetrics.setCommittedOffset(
+ messageQueue, offset);
+ } else {
+
rocketmqSourceReaderMetrics.recordCommitFailed();
+ LOG.error(
+ "Failed to commit offset {} for mq
{}",
+ offset,
+
UtilAll.getQueueDescription(messageQueue),
+ error);
+ }
+ });
+ }
+ }
+
+ /**
+ * Asks the broker for the maximum offset of every assigned queue, at most
once per {@link
+ * #LAG_REFRESH_INTERVAL_MS}. The lookup is asynchronous so the fetch loop
never blocks on a
+ * remote call; a queue whose lookup fails simply keeps its previous lag.
+ */
+ private void refreshLagIfDue() {
+ long now = System.currentTimeMillis();
+ if (now - lastLagRefreshTime < LAG_REFRESH_INTERVAL_MS) {
+ return;
+ }
+ lastLagRefreshTime = now;
+ for (MessageQueue messageQueue : currentOffsetTable.keySet()) {
+ consumer.seekMaxOffset(messageQueue)
+ .whenComplete(
+ (maxOffset, error) -> {
+ if (error != null) {
+ LOG.debug(
+ "Failed to fetch the max offset of
mq {} for the lag metric",
+
UtilAll.getQueueDescription(messageQueue),
+ error);
+ } else if (maxOffset != null) {
+
rocketmqSourceReaderMetrics.updateMaxOffset(
+ messageQueue, maxOffset);
+ }
+ });
}
}
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/reader/deserializer/RowDeserializationSchema.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/reader/deserializer/RowDeserializationSchema.java
index 98baef7..24d9c71 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/reader/deserializer/RowDeserializationSchema.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/reader/deserializer/RowDeserializationSchema.java
@@ -20,9 +20,9 @@ package
org.apache.flink.connector.rocketmq.source.reader.deserializer;
import
org.apache.flink.api.common.serialization.DeserializationSchema.InitializationContext;
import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.configuration.ConfigOption;
-import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.rocketmq.common.config.RocketMQOptions;
+import org.apache.flink.connector.rocketmq.source.RocketMQSourceOptions;
import org.apache.flink.connector.rocketmq.source.util.ByteSerializer;
import org.apache.flink.connector.rocketmq.source.util.StringSerializer;
import org.apache.flink.table.api.TableSchema;
@@ -528,46 +528,54 @@ public class RowDeserializationSchema implements
QueryableSchema<List<BytesMessa
return this;
}
Configuration configuration = new Configuration();
- for (String key : properties.keySet()) {
- configuration.setString(key, properties.get(key));
+ for (Map.Entry<String, String> entry : properties.entrySet()) {
+ configuration.setString(entry.getKey(), entry.getValue());
}
- String lengthCheck =
configuration.get(CollectorOption.LENGTH_CHECK);
+
applyLengthCheck(configuration.get(RocketMQOptions.OPTIONAL_LENGTH_CHECK));
+
this.setEncoding(configuration.get(RocketMQOptions.OPTIONAL_ENCODING));
+
this.setFieldDelimiter(configuration.get(RocketMQOptions.OPTIONAL_FIELD_DELIMITER));
+
this.setLineDelimiter(configuration.get(RocketMQOptions.OPTIONAL_LINE_DELIMITER));
+ this.setColumnErrorDebug(
+
configuration.get(RocketMQSourceOptions.OPTIONAL_COLUMN_ERROR_DEBUG));
+ return this;
+ }
+
+ /**
+ * Maps {@code rocketmq.client.message.length.check} onto the dirty
data strategies. An
+ * unknown value is rejected rather than ignored: silently keeping the
defaults is what
+ * turns a misspelled option into a stream of dropped records.
+ */
+ private void applyLengthCheck(String lengthCheck) {
switch (lengthCheck.toUpperCase()) {
+ case "NONE":
+ break;
case "SKIP":
- {
- this.setFormatErrorStrategy(DirtyDataStrategy.SKIP);
- this.setFieldMissingStrategy(DirtyDataStrategy.SKIP);
- this.setFieldIncrementStrategy(DirtyDataStrategy.SKIP);
- }
+ this.setFormatErrorStrategy(DirtyDataStrategy.SKIP);
+ this.setFieldMissingStrategy(DirtyDataStrategy.SKIP);
+ this.setFieldIncrementStrategy(DirtyDataStrategy.SKIP);
break;
case "PAD":
- {
- this.setFormatErrorStrategy(DirtyDataStrategy.SKIP);
- this.setFieldMissingStrategy(DirtyDataStrategy.PAD);
- this.setFieldIncrementStrategy(DirtyDataStrategy.CUT);
- }
- break;
- case "EXCEPTION":
- {
-
this.setFormatErrorStrategy(DirtyDataStrategy.EXCEPTION);
-
this.setFieldMissingStrategy(DirtyDataStrategy.EXCEPTION);
-
this.setFieldIncrementStrategy(DirtyDataStrategy.EXCEPTION);
- }
+ this.setFormatErrorStrategy(DirtyDataStrategy.SKIP);
+ this.setFieldMissingStrategy(DirtyDataStrategy.PAD);
+ this.setFieldIncrementStrategy(DirtyDataStrategy.CUT);
break;
case "SKIP_SILENT":
- {
-
this.setFormatErrorStrategy(DirtyDataStrategy.SKIP_SILENT);
-
this.setFieldMissingStrategy(DirtyDataStrategy.SKIP_SILENT);
-
this.setFieldIncrementStrategy(DirtyDataStrategy.SKIP_SILENT);
- }
+ this.setFormatErrorStrategy(DirtyDataStrategy.SKIP_SILENT);
+
this.setFieldMissingStrategy(DirtyDataStrategy.SKIP_SILENT);
+
this.setFieldIncrementStrategy(DirtyDataStrategy.SKIP_SILENT);
+ break;
+ case "EXCEPTION":
+ this.setFormatErrorStrategy(DirtyDataStrategy.EXCEPTION);
+ this.setFieldMissingStrategy(DirtyDataStrategy.EXCEPTION);
+
this.setFieldIncrementStrategy(DirtyDataStrategy.EXCEPTION);
break;
default:
+ throw new IllegalArgumentException(
+ String.format(
+ "Unsupported value '%s' for %s. Expected
one of NONE, SKIP, "
+ + "SKIP_SILENT, PAD, EXCEPTION.",
+ lengthCheck,
RocketMQOptions.OPTIONAL_LENGTH_CHECK.key()));
}
-
this.setEncoding(configuration.getString(CollectorOption.ENCODING));
-
this.setFieldDelimiter(configuration.getString(CollectorOption.FIELD_DELIMITER));
-
this.setLineDelimiter(configuration.getString(CollectorOption.LINE_DELIMITER));
-
this.setColumnErrorDebug(configuration.getBoolean(CollectorOption.COLUMN_ERROR_DEBUG));
- return this;
}
public RowDeserializationSchema build() {
@@ -586,18 +594,4 @@ public class RowDeserializationSchema implements
QueryableSchema<List<BytesMessa
properties);
}
}
-
- /** Options for {@link RowDeserializationSchema}. */
- public static class CollectorOption {
- public static final ConfigOption<String> ENCODING =
-
ConfigOptions.key("encoding".toLowerCase()).defaultValue("UTF-8");
- public static final ConfigOption<String> FIELD_DELIMITER =
-
ConfigOptions.key("fieldDelimiter".toLowerCase()).defaultValue("\u0001");
- public static final ConfigOption<String> LINE_DELIMITER =
-
ConfigOptions.key("lineDelimiter".toLowerCase()).defaultValue("\n");
- public static final ConfigOption<Boolean> COLUMN_ERROR_DEBUG =
-
ConfigOptions.key("columnErrorDebug".toLowerCase()).defaultValue(true);
- public static final ConfigOption<String> LENGTH_CHECK =
-
ConfigOptions.key("lengthCheck".toLowerCase()).defaultValue("NONE");
- }
}
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/RocketMQSink.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/RocketMQSink.java
index bc3ef5c..f6cae6e 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/RocketMQSink.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/RocketMQSink.java
@@ -38,10 +38,12 @@ import
org.apache.rocketmq.remoting.exception.RemotingException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.Properties;
import java.util.UUID;
+import java.util.concurrent.atomic.AtomicReference;
/**
* The RocketMQSink provides at-least-once reliability guarantees when
checkpoints are enabled and
@@ -106,8 +108,17 @@ public class RocketMQSink extends
RichSinkFunction<Message> implements Checkpoin
latencyGauge = MetricUtils.registerOutLatency(getRuntimeContext());
}
+ /**
+ * The first asynchronous send failure that has not been reported yet. An
async callback runs on
+ * a client thread, so it cannot fail the task directly; the failure is
parked here and rethrown
+ * from the next {@link #invoke} or from {@link #finish}. Without this the
job keeps running and
+ * checkpointing while every record is being dropped.
+ */
+ private final AtomicReference<Throwable> asyncSendException = new
AtomicReference<>();
+
@Override
public void invoke(Message input, Context context) throws Exception {
+ checkAsyncSendException();
sinkInTps.markEvent();
if (batchFlushOnCheckpoint) {
@@ -138,6 +149,7 @@ public class RocketMQSink extends RichSinkFunction<Message>
implements Checkpoin
public void onException(Throwable throwable) {
if (throwable != null) {
LOG.error("Async send message failure!",
throwable);
+ asyncSendException.compareAndSet(null,
throwable);
}
}
};
@@ -152,6 +164,7 @@ public class RocketMQSink extends RichSinkFunction<Message>
implements Checkpoin
}
} catch (Exception e) {
LOG.error("Async send message failure!", e);
+ asyncSendException.compareAndSet(null, e);
}
} else {
try {
@@ -180,6 +193,19 @@ public class RocketMQSink extends
RichSinkFunction<Message> implements Checkpoin
}
}
+ @Override
+ public void finish() throws Exception {
+ checkAsyncSendException();
+ }
+
+ /** Rethrows the first asynchronous send failure recorded since the last
call. */
+ private void checkAsyncSendException() throws IOException {
+ Throwable throwable = asyncSendException.getAndSet(null);
+ if (throwable != null) {
+ throw new IOException("Async send message failed", throwable);
+ }
+ }
+
public RocketMQSink withAsync(boolean async) {
this.async = async;
return this;
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/RocketMQSourceFunction.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/RocketMQSourceFunction.java
index 42dc415..2fe7f98 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/RocketMQSourceFunction.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/RocketMQSourceFunction.java
@@ -25,6 +25,7 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.rocketmq.common.filter.Sql92MessageFilter;
import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.Meter;
import org.apache.flink.metrics.MeterView;
@@ -48,7 +49,6 @@ import org.apache.commons.collections.map.LinkedMap;
import org.apache.commons.lang.Validate;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
-import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
@@ -88,6 +88,7 @@ public class RocketMQSourceFunction<OUT> extends
RichParallelSourceFunction<OUT>
private RunningChecker runningChecker;
private transient DefaultLitePullConsumer consumer;
+ private transient Sql92MessageFilter sqlFilter;
private KeyValueDeserializationSchema<OUT> schema;
private transient ListState<Tuple2<MessageQueue, Long>> unionOffsetStates;
@@ -227,6 +228,11 @@ public class RocketMQSourceFunction<OUT> extends
RichParallelSourceFunction<OUT>
Collection<MessageQueue> totalQueues =
consumer.fetchMessageQueues(topic);
messageQueues =
RocketMQUtils.allocate(totalQueues, taskNumber,
ctx.getIndexOfThisSubtask());
+ // Assign the queues this subtask was allocated instead of
subscribing. subscribe() leaves
+ // allocation to the client's own rebalance, which runs
asynchronously, so the assigned list
+ // is still empty when initOffsets() seeks below and every seek fails
with "The message
+ // queue is not in assigned list".
+ consumer.assign(messageQueues);
// If the job recovers from the state, the state has already contained
the offsets of last
// commit.
if (restored) {
@@ -249,12 +255,13 @@ public class RocketMQSourceFunction<OUT> extends
RichParallelSourceFunction<OUT>
RocketMQConfig.DEFAULT_CONSUMER_BATCH_SIZE);
// Watermark emission removed: legacy watermark assigners are no
longer available.
// Use WatermarkStrategy with the new Source API instead.
- if (StringUtils.isEmpty(sql)) {
- consumer.subscribe(topic, tag);
- } else {
- // pull with sql do not support block pull.
- consumer.subscribe(topic, MessageSelector.bySql(sql));
+ // The queues are assigned in open(); only the filter is set up here.
A tag sub-expression
+ // is honoured by the broker in assign mode, while SQL92 cannot be
pushed down and is
+ // evaluated against the message properties on the way out.
+ if (!StringUtils.isEmpty(tag) &&
!RocketMQConfig.DEFAULT_CONSUMER_TAG.equals(tag)) {
+ consumer.setSubExpressionForAssign(topic, tag);
}
+ this.sqlFilter = StringUtils.isEmpty(sql) ? null : new
Sql92MessageFilter(sql);
for (MessageQueue mq : messageQueues) {
this.executor.execute(() -> consumeMessages(mq, context,
pullBatchSize));
}
@@ -317,6 +324,9 @@ public class RocketMQSourceFunction<OUT> extends
RichParallelSourceFunction<OUT>
private void emitMessages(List<MessageExt> messages, SourceContext<OUT>
context) {
long fetchTime = System.currentTimeMillis();
for (MessageExt msg : messages) {
+ if (sqlFilter != null && !sqlFilter.matches(msg.getProperties())) {
+ continue;
+ }
byte[] key =
msg.getKeys() != null ?
msg.getKeys().getBytes(StandardCharsets.UTF_8) : null;
byte[] value = msg.getBody();
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/common/serialization/RowKeyValueDeserializationSchema.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/common/serialization/RowKeyValueDeserializationSchema.java
index bf7d58d..69a2fa2 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/common/serialization/RowKeyValueDeserializationSchema.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/common/serialization/RowKeyValueDeserializationSchema.java
@@ -17,9 +17,9 @@
package org.apache.flink.streaming.connectors.rocketmq.common.serialization;
import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.configuration.ConfigOption;
-import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.rocketmq.common.config.RocketMQOptions;
+import org.apache.flink.connector.rocketmq.source.RocketMQSourceOptions;
import
org.apache.flink.connector.rocketmq.source.reader.deserializer.DirtyDataStrategy;
import org.apache.flink.connector.rocketmq.source.util.ByteSerializer;
import org.apache.flink.connector.rocketmq.source.util.StringSerializer;
@@ -337,45 +337,53 @@ public class RowKeyValueDeserializationSchema implements
KeyValueDeserialization
return this;
}
Configuration configuration = new Configuration();
- for (String key : properties.keySet()) {
- configuration.setString(key, properties.get(key));
+ for (Map.Entry<String, String> entry : properties.entrySet()) {
+ configuration.setString(entry.getKey(), entry.getValue());
}
- String lengthCheck =
configuration.get(CollectorOption.LENGTH_CHECK);
+
applyLengthCheck(configuration.get(RocketMQOptions.OPTIONAL_LENGTH_CHECK));
+
this.setEncoding(configuration.get(RocketMQOptions.OPTIONAL_ENCODING));
+
this.setFieldDelimiter(configuration.get(RocketMQOptions.OPTIONAL_FIELD_DELIMITER));
+ this.setColumnErrorDebug(
+
configuration.get(RocketMQSourceOptions.OPTIONAL_COLUMN_ERROR_DEBUG));
+ return this;
+ }
+
+ /**
+ * Maps {@code rocketmq.client.message.length.check} onto the dirty
data strategies. An
+ * unknown value is rejected rather than ignored: silently keeping the
defaults is what
+ * turns a misspelled option into a stream of dropped records.
+ */
+ private void applyLengthCheck(String lengthCheck) {
switch (lengthCheck.toUpperCase()) {
+ case "NONE":
+ break;
case "SKIP":
- {
- this.setFormatErrorStrategy(DirtyDataStrategy.SKIP);
- this.setFieldMissingStrategy(DirtyDataStrategy.SKIP);
- this.setFieldIncrementStrategy(DirtyDataStrategy.SKIP);
- }
+ this.setFormatErrorStrategy(DirtyDataStrategy.SKIP);
+ this.setFieldMissingStrategy(DirtyDataStrategy.SKIP);
+ this.setFieldIncrementStrategy(DirtyDataStrategy.SKIP);
break;
case "PAD":
- {
- this.setFormatErrorStrategy(DirtyDataStrategy.SKIP);
- this.setFieldMissingStrategy(DirtyDataStrategy.PAD);
- this.setFieldIncrementStrategy(DirtyDataStrategy.CUT);
- }
- break;
- case "EXCEPTION":
- {
-
this.setFormatErrorStrategy(DirtyDataStrategy.EXCEPTION);
-
this.setFieldMissingStrategy(DirtyDataStrategy.EXCEPTION);
-
this.setFieldIncrementStrategy(DirtyDataStrategy.EXCEPTION);
- }
+ this.setFormatErrorStrategy(DirtyDataStrategy.SKIP);
+ this.setFieldMissingStrategy(DirtyDataStrategy.PAD);
+ this.setFieldIncrementStrategy(DirtyDataStrategy.CUT);
break;
case "SKIP_SILENT":
- {
-
this.setFormatErrorStrategy(DirtyDataStrategy.SKIP_SILENT);
-
this.setFieldMissingStrategy(DirtyDataStrategy.SKIP_SILENT);
-
this.setFieldIncrementStrategy(DirtyDataStrategy.SKIP_SILENT);
- }
+ this.setFormatErrorStrategy(DirtyDataStrategy.SKIP_SILENT);
+
this.setFieldMissingStrategy(DirtyDataStrategy.SKIP_SILENT);
+
this.setFieldIncrementStrategy(DirtyDataStrategy.SKIP_SILENT);
+ break;
+ case "EXCEPTION":
+ this.setFormatErrorStrategy(DirtyDataStrategy.EXCEPTION);
+ this.setFieldMissingStrategy(DirtyDataStrategy.EXCEPTION);
+
this.setFieldIncrementStrategy(DirtyDataStrategy.EXCEPTION);
break;
default:
+ throw new IllegalArgumentException(
+ String.format(
+ "Unsupported value '%s' for %s. Expected
one of NONE, SKIP, "
+ + "SKIP_SILENT, PAD, EXCEPTION.",
+ lengthCheck,
RocketMQOptions.OPTIONAL_LENGTH_CHECK.key()));
}
-
this.setEncoding(configuration.getString(CollectorOption.ENCODING));
-
this.setFieldDelimiter(configuration.getString(CollectorOption.FIELD_DELIMITER));
-
this.setColumnErrorDebug(configuration.getBoolean(CollectorOption.COLUMN_ERROR_DEBUG));
- return this;
}
public RowKeyValueDeserializationSchema build() {
@@ -390,16 +398,4 @@ public class RowKeyValueDeserializationSchema implements
KeyValueDeserialization
properties);
}
}
-
- /** Options for {@link RowKeyValueDeserializationSchema}. */
- public static class CollectorOption {
- public static final ConfigOption<String> ENCODING =
-
ConfigOptions.key("encoding".toLowerCase()).defaultValue("UTF-8");
- public static final ConfigOption<String> FIELD_DELIMITER =
-
ConfigOptions.key("fieldDelimiter".toLowerCase()).defaultValue("\u0001");
- public static final ConfigOption<Boolean> COLUMN_ERROR_DEBUG =
-
ConfigOptions.key("columnErrorDebug".toLowerCase()).defaultValue(true);
- public static final ConfigOption<String> LENGTH_CHECK =
-
ConfigOptions.key("lengthCheck".toLowerCase()).defaultValue("NONE");
- }
}
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/table/RequiredOptionValidator.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/table/RequiredOptionValidator.java
new file mode 100644
index 0000000..01c3398
--- /dev/null
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/table/RequiredOptionValidator.java
@@ -0,0 +1,56 @@
+/*
+ * 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.streaming.connectors.rocketmq.table;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.table.api.ValidationException;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Reports the required table options a DDL does not set, naming the missing
keys only.
+ *
+ * <p>{@link
org.apache.flink.table.factories.FactoryUtil.TableFactoryHelper#validate()}
repeats the
+ * table options in its failure message, and a WITH clause normally carries
{@code accessKey} and
+ * {@code secretKey}. That message ends up in the JobManager log and in the
REST exception history,
+ * so a typo would turn the credentials into plaintext in both. Factories
therefore run this check
+ * first and still call the helper afterwards, which keeps unknown option
detection unchanged.
+ */
+final class RequiredOptionValidator {
+
+ private RequiredOptionValidator() {}
+
+ static void validate(Set<ConfigOption<?>> requiredOptions, Map<String,
String> tableOptions) {
+ 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
connector: %s",
+ missing.size() == 1 ? "" : "s", missing));
+ }
+ }
+}
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQDynamicTableSinkFactory.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQDynamicTableSinkFactory.java
index 1210268..3c5f240 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQDynamicTableSinkFactory.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQDynamicTableSinkFactory.java
@@ -98,6 +98,7 @@ public class RocketMQDynamicTableSinkFactory implements
DynamicTableSinkFactory
@Override
public DynamicTableSink createDynamicTableSink(Context context) {
FactoryUtil.TableFactoryHelper helper = createTableFactoryHelper(this,
context);
+ RequiredOptionValidator.validate(requiredOptions(),
context.getCatalogTable().getOptions());
helper.validate();
Map<String, String> rawProperties =
context.getCatalogTable().getOptions();
Configuration properties = Configuration.fromMap(rawProperties);
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQDynamicTableSourceFactory.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQDynamicTableSourceFactory.java
index c5754c7..cd4f3f9 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQDynamicTableSourceFactory.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQDynamicTableSourceFactory.java
@@ -83,7 +83,6 @@ public class RocketMQDynamicTableSourceFactory implements
DynamicTableSourceFact
optionalOptions.add(RocketMQSourceOptions.OPTIONAL_SECRET_KEY);
optionalOptions.add(RocketMQSourceOptions.OPTIONAL_STARTUP_SCAN_MODE);
optionalOptions.add(RocketMQSourceOptions.OPTIONAL_STARTUP_OFFSET_DATE);
-
optionalOptions.add(RocketMQSourceOptions.OPTIONAL_STARTUP_OFFSET_STRATEGY);
optionalOptions.add(RocketMQSourceOptions.OPTIONAL_USE_NEW_API);
optionalOptions.add(RocketMQSourceOptions.PULL_BATCH_SIZE);
optionalOptions.add(RocketMQSourceOptions.ALLOCATE_MESSAGE_QUEUE_STRATEGY);
@@ -99,6 +98,7 @@ public class RocketMQDynamicTableSourceFactory implements
DynamicTableSourceFact
@Override
public DynamicTableSource createDynamicTableSource(Context context) {
FactoryUtil.TableFactoryHelper helper = createTableFactoryHelper(this,
context);
+ RequiredOptionValidator.validate(requiredOptions(),
context.getCatalogTable().getOptions());
helper.validate();
Map<String, String> rawProperties =
context.getCatalogTable().getOptions();
Configuration configuration = Configuration.fromMap(rawProperties);
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQScanTableSource.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQScanTableSource.java
index 6d12961..43864f9 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQScanTableSource.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQScanTableSource.java
@@ -238,8 +238,16 @@ public class RocketMQScanTableSource implements
ScanTableSource, SupportsReading
case "committed":
return OffsetsSelector.committedOffsets();
case "latest":
- default:
return OffsetsSelector.latest();
+ default:
+ // Silently falling back to a default is what let a misspelled
scan mode look like
+ // a healthy job that started reading from the wrong position.
+ throw new IllegalArgumentException(
+ String.format(
+ "Unsupported value '%s' for %s. Expected one
of earliest, latest, "
+ + "timestamp, committed.",
+ consumerOffsetMode,
+
RocketMQSourceOptions.OPTIONAL_STARTUP_SCAN_MODE.key()));
}
}
diff --git
a/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/common/filter/Sql92MessageFilterTest.java
b/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/common/filter/Sql92MessageFilterTest.java
new file mode 100644
index 0000000..b2009ef
--- /dev/null
+++
b/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/common/filter/Sql92MessageFilterTest.java
@@ -0,0 +1,103 @@
+/*
+ * 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.common.filter;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Tests for {@link Sql92MessageFilter}. Message properties always arrive as
strings, so the cases
+ * below also pin down how literals are coerced, which is what decides whether
a filter silently
+ * accepts every message.
+ */
+class Sql92MessageFilterTest {
+
+ @Test
+ void numericComparisonKeepsOnlyMatchingMessagesTest() {
+ Sql92MessageFilter filter = new Sql92MessageFilter("id > 50");
+
+ List<Integer> matched = new ArrayList<>();
+ for (int id = 0; id < 100; id++) {
+ if (filter.matches(properties("id", String.valueOf(id)))) {
+ matched.add(id);
+ }
+ }
+
+ assertThat(matched).hasSize(49).startsWith(51).endsWith(99);
+ }
+
+ @Test
+ void stringEqualityAndLogicalOperatorsTest() {
+ Sql92MessageFilter filter = new Sql92MessageFilter("region =
'hangzhou' AND retries >= 2");
+
+ assertThat(filter.matches(row("hangzhou", "3"))).isTrue();
+ assertThat(filter.matches(row("hangzhou", "1"))).isFalse();
+ assertThat(filter.matches(row("beijing", "3"))).isFalse();
+ }
+
+ @Test
+ void inBetweenAndNotOperatorsTest() {
+ Sql92MessageFilter inFilter = new Sql92MessageFilter("level IN ('a',
'b')");
+ assertThat(inFilter.matches(properties("level", "a"))).isTrue();
+ assertThat(inFilter.matches(properties("level", "c"))).isFalse();
+
+ Sql92MessageFilter betweenFilter = new Sql92MessageFilter("seq BETWEEN
10 AND 20");
+ assertThat(betweenFilter.matches(properties("seq", "10"))).isTrue();
+ assertThat(betweenFilter.matches(properties("seq", "21"))).isFalse();
+
+ Sql92MessageFilter notFilter = new Sql92MessageFilter("NOT (seq > 5)");
+ assertThat(notFilter.matches(properties("seq", "1"))).isTrue();
+ assertThat(notFilter.matches(properties("seq", "9"))).isFalse();
+ }
+
+ @Test
+ void missingPropertyDoesNotMatchTest() {
+ Sql92MessageFilter filter = new Sql92MessageFilter("id > 50");
+
+ assertThat(filter.matches(Collections.emptyMap())).isFalse();
+ assertThat(filter.matches(properties("other", "99"))).isFalse();
+ assertThat(filter.matches(null)).isFalse();
+ }
+
+ @Test
+ void malformedExpressionFailsAtConstructionTest() {
+ assertThatThrownBy(() -> new Sql92MessageFilter("id > > 50"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("id > > 50");
+ }
+
+ private static Map<String, String> row(String region, String retries) {
+ Map<String, String> properties = new HashMap<>();
+ properties.put("region", region);
+ properties.put("retries", retries);
+ return properties;
+ }
+
+ private static Map<String, String> properties(String key, String value) {
+ return Collections.singletonMap(key, value);
+ }
+}
diff --git
a/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/sink/writer/RocketMQWriterQueueSelectorTest.java
b/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/sink/writer/RocketMQWriterQueueSelectorTest.java
new file mode 100644
index 0000000..1828201
--- /dev/null
+++
b/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/sink/writer/RocketMQWriterQueueSelectorTest.java
@@ -0,0 +1,158 @@
+/*
+ * 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.sink.writer;
+
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.base.DeliveryGuarantee;
+import org.apache.flink.connector.rocketmq.sink.InnerProducer;
+import org.apache.flink.connector.rocketmq.sink.RocketMQSinkOptions;
+import org.apache.flink.connector.rocketmq.sink.committer.SendCommittable;
+import
org.apache.flink.connector.rocketmq.sink.writer.context.RocketMQSinkContext;
+import
org.apache.flink.connector.rocketmq.sink.writer.serializer.RocketMQSerializationSchema;
+import
org.apache.flink.streaming.connectors.rocketmq.common.selector.HashMessageQueueSelector;
+import
org.apache.flink.streaming.connectors.rocketmq.common.selector.MessageQueueSelector;
+
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.common.message.Message;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests that a configured {@link MessageQueueSelector} actually reaches the
producer. Accepting the
+ * selector on the builder without ever passing it to the send call silently
turns ordered writes
+ * into round robin writes, which no amount of configuration can reveal at
runtime.
+ */
+class RocketMQWriterQueueSelectorTest {
+
+ private static final String TOPIC = "TopicWriterSelector";
+
+ @Test
+ void configuredSelectorReachesTheProducerTest() throws Exception {
+ RecordingProducer producer = new RecordingProducer();
+ MessageQueueSelector selector = new HashMessageQueueSelector();
+
+ newWriter(selector, producer).write("sharding-key", null);
+
+ assertThat(producer.plainSends).isEmpty();
+ assertThat(producer.selectorSends).hasSize(1);
+ assertThat(producer.selectorSends.get(0).selector).isSameAs(selector);
+
assertThat(producer.selectorSends.get(0).arg).isEqualTo("sharding-key");
+
assertThat(producer.selectorSends.get(0).message.getTopic()).isEqualTo(TOPIC);
+ }
+
+ @Test
+ void withoutSelectorThePlainSendIsUsedTest() throws Exception {
+ RecordingProducer producer = new RecordingProducer();
+
+ newWriter(null, producer).write("sharding-key", null);
+
+ assertThat(producer.plainSends).hasSize(1);
+ assertThat(producer.selectorSends).isEmpty();
+ }
+
+ private static RocketMQWriter<String> newWriter(
+ MessageQueueSelector selector, InnerProducer producer) {
+ Configuration configuration = new Configuration();
+ configuration.setString(
+ RocketMQSinkOptions.DELIVERY_GUARANTEE,
DeliveryGuarantee.AT_LEAST_ONCE.name());
+ return new RocketMQWriter<>(
+ configuration,
+ selector,
+ new StringToMessageSchema(),
+ Mockito.mock(Sink.InitContext.class),
+ producer);
+ }
+
+ /** Wraps the record into a message body so the send call can be
inspected. */
+ private static class StringToMessageSchema implements
RocketMQSerializationSchema<String> {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Message serialize(String element, RocketMQSinkContext context,
Long timestamp) {
+ return new Message(TOPIC,
element.getBytes(StandardCharsets.UTF_8));
+ }
+ }
+
+ /** Records which send overload the writer chose. */
+ private static class RecordingProducer implements InnerProducer {
+
+ private final List<Message> plainSends = new ArrayList<>();
+ private final List<SelectorSend> selectorSends = new ArrayList<>();
+
+ @Override
+ public void start() {}
+
+ @Override
+ public String getProducerGroup() {
+ return "recording-producer";
+ }
+
+ @Override
+ public CompletableFuture<SendResult> send(Message message) {
+ plainSends.add(message);
+ return CompletableFuture.completedFuture(new SendResult());
+ }
+
+ @Override
+ public CompletableFuture<SendResult> send(
+ Message message, MessageQueueSelector selector, Object arg) {
+ selectorSends.add(new SelectorSend(message, selector, arg));
+ return CompletableFuture.completedFuture(new SendResult());
+ }
+
+ @Override
+ public CompletableFuture<SendResult> sendMessageInTransaction(Message
message) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CompletableFuture<Void> commit(SendCommittable sendCommittable)
{
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CompletableFuture<Void> rollback(SendCommittable
sendCommittable) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void close() {}
+
+ private static class SelectorSend {
+ private final Message message;
+ private final MessageQueueSelector selector;
+ private final Object arg;
+
+ private SelectorSend(Message message, MessageQueueSelector
selector, Object arg) {
+ this.message = message;
+ this.selector = selector;
+ this.arg = arg;
+ }
+ }
+ }
+}
diff --git
a/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/source/metrics/RocketMQSourceReaderMetricsTest.java
b/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/source/metrics/RocketMQSourceReaderMetricsTest.java
new file mode 100644
index 0000000..8ea85f7
--- /dev/null
+++
b/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/source/metrics/RocketMQSourceReaderMetricsTest.java
@@ -0,0 +1,208 @@
+/*
+ * 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.source.metrics;
+
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Gauge;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.SimpleCounter;
+import org.apache.flink.metrics.groups.OperatorIOMetricGroup;
+import org.apache.flink.metrics.groups.SourceReaderMetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for {@link RocketMQSourceReaderMetrics}. The metric groups are
replaced by a recording fake
+ * so the gauges and counters can be read back the way a metric reporter would.
+ */
+class RocketMQSourceReaderMetricsTest {
+
+ private static final MessageQueue QUEUE_0 = new MessageQueue("TopicA",
"broker-a", 0);
+ private static final MessageQueue QUEUE_1 = new MessageQueue("TopicA",
"broker-a", 1);
+
+ private final RecordingSourceReaderMetricGroup metricGroup =
+ new RecordingSourceReaderMetricGroup();
+ private final RocketMQSourceReaderMetrics metrics =
+ new RocketMQSourceReaderMetrics(metricGroup);
+
+ @Test
+ void registeredQueueExposesOffsetGaugesTest() {
+ metrics.registerNewMessageQueue(QUEUE_0);
+ metrics.setCurrentOffset(QUEUE_0, 42L);
+ metrics.setCommittedOffset(QUEUE_0, 40L);
+
+ RecordingMetricGroup queueGroup = queueGroup("TopicA", "0");
+ assertThat(gaugeValue(queueGroup, "currentOffset")).isEqualTo(42L);
+ assertThat(gaugeValue(queueGroup, "committedOffset")).isEqualTo(40L);
+ }
+
+ @Test
+ void lagIsDerivedFromTheBrokerMaxOffsetTest() {
+ metrics.registerNewMessageQueue(QUEUE_0);
+ metrics.setCurrentOffset(QUEUE_0, 100L);
+ metrics.updateMaxOffset(QUEUE_0, 250L);
+
+ assertThat(gaugeValue(queueGroup("TopicA", "0"),
"records-lag")).isEqualTo(150L);
+ assertThat(metricGroup.pendingRecords.getValue()).isEqualTo(150L);
+ }
+
+ @Test
+ void lagNeverGoesNegativeAndStartsAtZeroTest() {
+ metrics.registerNewMessageQueue(QUEUE_0);
+
+ assertThat(metricGroup.pendingRecords.getValue()).isZero();
+
+ metrics.setCurrentOffset(QUEUE_0, 300L);
+ metrics.updateMaxOffset(QUEUE_0, 120L);
+
+ assertThat(metricGroup.pendingRecords.getValue()).isZero();
+ }
+
+ @Test
+ void pendingRecordsSumsEveryAssignedQueueTest() {
+ metrics.registerNewMessageQueue(QUEUE_0);
+ metrics.registerNewMessageQueue(QUEUE_1);
+ metrics.setCurrentOffset(QUEUE_0, 10L);
+ metrics.updateMaxOffset(QUEUE_0, 30L);
+ metrics.setCurrentOffset(QUEUE_1, 5L);
+ metrics.updateMaxOffset(QUEUE_1, 100L);
+
+ assertThat(metricGroup.pendingRecords.getValue()).isEqualTo(115L);
+ }
+
+ @Test
+ void unregisteredQueueStopsContributingToLagTest() {
+ metrics.registerNewMessageQueue(QUEUE_0);
+ metrics.registerNewMessageQueue(QUEUE_1);
+ metrics.setCurrentOffset(QUEUE_0, 10L);
+ metrics.updateMaxOffset(QUEUE_0, 30L);
+ metrics.setCurrentOffset(QUEUE_1, 5L);
+ metrics.updateMaxOffset(QUEUE_1, 100L);
+
+ metrics.unregisterMessageQueue(QUEUE_1);
+
+ assertThat(metricGroup.pendingRecords.getValue()).isEqualTo(20L);
+ // Updates for a queue that is no longer assigned must not resurrect
it.
+ metrics.updateMaxOffset(QUEUE_1, 500L);
+ assertThat(metricGroup.pendingRecords.getValue()).isEqualTo(20L);
+ }
+
+ @Test
+ void commitCountersTrackSuccessAndFailureTest() {
+ metrics.recordCommitSucceeded();
+ metrics.recordCommitSucceeded();
+ metrics.recordCommitFailed();
+
+ RecordingMetricGroup readerGroup =
metricGroup.group("RocketmqSourceReader");
+
assertThat(readerGroup.counters.get("commitsSucceeded").getCount()).isEqualTo(2L);
+
assertThat(readerGroup.counters.get("commitsFailed").getCount()).isEqualTo(1L);
+ }
+
+ @Test
+ void bytesConsumedCounterAccumulatesTest() {
+ metrics.recordBytesConsumed(1024L);
+ metrics.recordBytesConsumed(512L);
+
+ RecordingMetricGroup fetchManager =
+
metricGroup.group("RocketMQConsumer").group("consumer-fetch-manager-metrics");
+
assertThat(fetchManager.counters.get("bytes-consumed-total").getCount()).isEqualTo(1536L);
+ }
+
+ private RecordingMetricGroup queueGroup(String topic, String queueId) {
+ return metricGroup
+ .group("RocketmqSourceReader")
+ .group("topic=" + topic)
+ .group("queue=" + queueId);
+ }
+
+ private static long gaugeValue(RecordingMetricGroup group, String name) {
+ Gauge<?> gauge = group.gauges.get(name);
+ assertThat(gauge).as("gauge %s", name).isNotNull();
+ return ((Number) gauge.getValue()).longValue();
+ }
+
+ /** Records every metric registered on it so tests can read the values
back. */
+ private static class RecordingMetricGroup extends UnregisteredMetricsGroup
{
+
+ final Map<String, Gauge<?>> gauges = new HashMap<>();
+ final Map<String, Counter> counters = new HashMap<>();
+ final Map<String, RecordingMetricGroup> groups = new HashMap<>();
+
+ @Override
+ public <T, G extends Gauge<T>> G gauge(String name, G gauge) {
+ gauges.put(name, gauge);
+ return gauge;
+ }
+
+ @Override
+ public Counter counter(String name) {
+ SimpleCounter counter = new SimpleCounter();
+ counters.put(name, counter);
+ return counter;
+ }
+
+ @Override
+ public MetricGroup addGroup(String name) {
+ return groups.computeIfAbsent(name, ignored -> new
RecordingMetricGroup());
+ }
+
+ @Override
+ public MetricGroup addGroup(String key, String value) {
+ return groups.computeIfAbsent(key + "=" + value, ignored -> new
RecordingMetricGroup());
+ }
+
+ RecordingMetricGroup group(String name) {
+ RecordingMetricGroup group = groups.get(name);
+ assertThat(group).as("metric group %s", name).isNotNull();
+ return group;
+ }
+ }
+
+ /** The recording group plus the handful of methods only a source reader
group declares. */
+ private static class RecordingSourceReaderMetricGroup extends
RecordingMetricGroup
+ implements SourceReaderMetricGroup {
+
+ Gauge<Long> pendingRecords;
+
+ @Override
+ public OperatorIOMetricGroup getIOMetricGroup() {
+ return UnregisteredMetricsGroup.createOperatorIOMetricGroup();
+ }
+
+ @Override
+ public Counter getNumRecordsInErrorsCounter() {
+ return new SimpleCounter();
+ }
+
+ @Override
+ public void setPendingBytesGauge(Gauge<Long> pendingBytesGauge) {}
+
+ @Override
+ public void setPendingRecordsGauge(Gauge<Long> pendingRecordsGauge) {
+ this.pendingRecords = pendingRecordsGauge;
+ }
+ }
+}
diff --git
a/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/source/reader/deserializer/RowDeserializationSchemaOptionsTest.java
b/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/source/reader/deserializer/RowDeserializationSchemaOptionsTest.java
new file mode 100644
index 0000000..dcc749d
--- /dev/null
+++
b/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/source/reader/deserializer/RowDeserializationSchemaOptionsTest.java
@@ -0,0 +1,144 @@
+/*
+ * 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.source.reader.deserializer;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.descriptors.DescriptorProperties;
+import org.apache.flink.util.Collector;
+
+import org.junit.jupiter.api.Test;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Tests that the format options a user writes in the DDL actually reach {@link
+ * RowDeserializationSchema}. Reading them under a different key than the one
the table factory
+ * registers makes every option silently ineffective, and with a multi column
schema the default
+ * field delimiter then drops every record instead of failing.
+ */
+class RowDeserializationSchemaOptionsTest {
+
+ private static final String FIELD_DELIMITER_KEY =
"rocketmq.client.message.field.delimiter";
+ private static final String LENGTH_CHECK_KEY =
"rocketmq.client.message.length.check";
+
+ @Test
+ void fieldDelimiterFromDdlSplitsRowTest() {
+ List<RowData> rows = deserialize(options(FIELD_DELIMITER_KEY, "|"),
"SMOKE1|a|b|c|d|e");
+
+ assertThat(rows).hasSize(1);
+ assertThat(rows.get(0).getArity()).isEqualTo(6);
+ assertThat(rows.get(0).getString(0).toString()).isEqualTo("SMOKE1");
+ assertThat(rows.get(0).getString(5).toString()).isEqualTo("e");
+ }
+
+ @Test
+ void defaultDelimiterDoesNotSplitPipeSeparatedBodyTest() {
+ // Documents the trap the option above avoids: with the default
delimiter the body is a
+ // single field, so a six column row has too few fields and is dropped.
+ assertThat(deserialize(ddlOptions(), "SMOKE1|a|b|c|d|e")).isEmpty();
+ }
+
+ @Test
+ void lineDelimiterFromDdlSplitsBodyIntoRowsTest() {
+ Map<String, String> options = options(FIELD_DELIMITER_KEY, "|");
+ options.put("rocketmq.client.message.line.delimiter", "#");
+
+ List<RowData> rows = deserialize(options,
"SMOKE1|a|b|c|d|e#SMOKE1|f|g|h|i|j");
+
+ assertThat(rows).hasSize(2);
+ assertThat(rows.get(1).getString(1).toString()).isEqualTo("f");
+ }
+
+ @Test
+ void unsupportedLengthCheckIsRejectedTest() {
+ assertThatThrownBy(() -> deserialize(options(LENGTH_CHECK_KEY,
"BOGUS"), "SMOKE1|a"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(LENGTH_CHECK_KEY)
+ .hasMessageContaining("BOGUS");
+ }
+
+ @Test
+ void lengthCheckExceptionFailsInsteadOfSkippingTest() {
+ Map<String, String> options = options(FIELD_DELIMITER_KEY, "|");
+ options.put(LENGTH_CHECK_KEY, "EXCEPTION");
+
+ assertThatThrownBy(() -> deserialize(options, "only-one-field"))
+ .isInstanceOf(RuntimeException.class);
+ }
+
+ private static Map<String, String> options(String key, String value) {
+ Map<String, String> options = ddlOptions();
+ options.put(key, value);
+ return options;
+ }
+
+ /** Mirrors what the table factory hands over: the DDL options plus the
physical schema. */
+ private static Map<String, String> ddlOptions() {
+ DescriptorProperties descriptorProperties = new DescriptorProperties();
+ descriptorProperties.putTableSchema("schema", tableSchema());
+ return new HashMap<>(descriptorProperties.asMap());
+ }
+
+ private static TableSchema tableSchema() {
+ return TableSchema.builder()
+ .field("f0", DataTypes.STRING())
+ .field("f1", DataTypes.STRING())
+ .field("f2", DataTypes.STRING())
+ .field("f3", DataTypes.STRING())
+ .field("f4", DataTypes.STRING())
+ .field("f5", DataTypes.STRING())
+ .build();
+ }
+
+ private static List<RowData> deserialize(Map<String, String> options,
String body) {
+ RowDeserializationSchema schema =
+ new RowDeserializationSchema.Builder()
+ .setTableSchema(tableSchema())
+ .setProperties(options)
+ .build();
+ schema.open(null);
+
+ BytesMessage message = new BytesMessage();
+ message.setData(body.getBytes(StandardCharsets.UTF_8));
+
+ List<RowData> rows = new ArrayList<>();
+ schema.deserialize(
+ Collections.singletonList(message),
+ new Collector<RowData>() {
+ @Override
+ public void collect(RowData record) {
+ rows.add(record);
+ }
+
+ @Override
+ public void close() {}
+ });
+ return rows;
+ }
+}
diff --git
a/flink-connector-rocketmq/src/test/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQDynamicTableSourceFactoryTest.java
b/flink-connector-rocketmq/src/test/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQDynamicTableSourceFactoryTest.java
index 5f4eff7..ddb18bf 100644
---
a/flink-connector-rocketmq/src/test/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQDynamicTableSourceFactoryTest.java
+++
b/flink-connector-rocketmq/src/test/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQDynamicTableSourceFactoryTest.java
@@ -28,6 +28,7 @@ import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ResolvedCatalogTable;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.factories.DynamicTableFactory;
import org.apache.flink.table.factories.FactoryUtil;
import org.junit.Test;
@@ -39,7 +40,9 @@ import java.util.Map;
import static org.apache.flink.table.api.DataTypes.STRING;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
/** Tests for {@link RocketMQDynamicTableSourceFactory}. */
public class RocketMQDynamicTableSourceFactoryTest {
@@ -55,6 +58,68 @@ public class RocketMQDynamicTableSourceFactoryTest {
private static final String CONSUMER_GROUP = "test_consumer";
private static final String NAME_SERVER_ADDRESS = "127.0.0.1:9876";
+ /**
+ * A missing required option must be reported by key only. Note that this
drives the factory
+ * directly: going through {@link FactoryUtil#createTableSource} lets
Flink run its own required
+ * option check first, and that message repeats every table option,
including the access key in
+ * plaintext.
+ */
+ @Test
+ public void testMissingRequiredOptionDoesNotEchoOptionValues() {
+ final Map<String, String> options = new HashMap<>();
+ options.put("connector", IDENTIFIER);
+ options.put(RocketMQSourceOptions.TOPIC.key(), TOPIC);
+ options.put(RocketMQSourceOptions.CONSUMER_GROUP.key(),
CONSUMER_GROUP);
+ options.put(RocketMQSourceOptions.OPTIONAL_ACCESS_KEY.key(),
"ak-value-must-not-leak");
+ options.put(RocketMQSourceOptions.OPTIONAL_SECRET_KEY.key(),
"sk-value-must-not-leak");
+
+ final ResolvedCatalogTable table =
+ new ResolvedCatalogTable(
+ CatalogTable.of(
+
Schema.newBuilder().fromResolvedSchema(SCHEMA).build(),
+ "mock source",
+ Collections.emptyList(),
+ options),
+ SCHEMA);
+ final DynamicTableFactory.Context context =
+ new DynamicTableFactory.Context() {
+ @Override
+ public ObjectIdentifier getObjectIdentifier() {
+ return ObjectIdentifier.of("default", "default",
IDENTIFIER);
+ }
+
+ @Override
+ public ResolvedCatalogTable getCatalogTable() {
+ return table;
+ }
+
+ @Override
+ public Configuration getConfiguration() {
+ return new Configuration();
+ }
+
+ @Override
+ public ClassLoader getClassLoader() {
+ return
RocketMQDynamicTableSourceFactoryTest.class.getClassLoader();
+ }
+
+ @Override
+ public boolean isTemporary() {
+ return false;
+ }
+ };
+
+ try {
+ new
RocketMQDynamicTableSourceFactory().createDynamicTableSource(context);
+ fail("expected the missing endpoints option to be reported");
+ } catch (ValidationException e) {
+ final String message = e.getMessage();
+ assertTrue(message,
message.contains(RocketMQSourceOptions.ENDPOINTS.key()));
+ assertFalse(message, message.contains("ak-value-must-not-leak"));
+ assertFalse(message, message.contains("sk-value-must-not-leak"));
+ }
+ }
+
@Test
public void testRocketMQDynamicTableSourceWithLegalOption() {
final Map<String, String> options = new HashMap<>();
diff --git
a/flink-connector-rocketmq/src/test/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQParameterChainTest.java
b/flink-connector-rocketmq/src/test/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQParameterChainTest.java
index f6759c5..28b4c27 100644
---
a/flink-connector-rocketmq/src/test/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQParameterChainTest.java
+++
b/flink-connector-rocketmq/src/test/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQParameterChainTest.java
@@ -58,12 +58,6 @@ class RocketMQParameterChainTest {
RocketMQSinkOptions.TRANSACTION_TIMEOUT);
}
- @Test
- void sinkFactoryShouldNotDeclareSerializeFormat() {
- RocketMQDynamicTableSinkFactory factory = new
RocketMQDynamicTableSinkFactory();
-
assertThat(factory.optionalOptions()).doesNotContain(RocketMQSinkOptions.SERIALIZE_FORMAT);
- }
-
@Test
void sourceConfigurationShouldPreserveUserValues() {
Configuration config = new Configuration();
