You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by ni...@apache.org on 2022/10/10 07:09:18 UTC
[rocketmq-flink] branch main updated: [ISSUE #59] Support rich initialization modes of RocketMQ source (#62)
This is an automated email from the ASF dual-hosted git repository.
nicholasjiang 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 e8d5350 [ISSUE #59] Support rich initialization modes of RocketMQ source (#62)
e8d5350 is described below
commit e8d535099ccc1c999e13b120d5de119df999a724
Author: 高思伟 <48...@users.noreply.github.com>
AuthorDate: Mon Oct 10 15:09:14 2022 +0800
[ISSUE #59] Support rich initialization modes of RocketMQ source (#62)
1.Add rich initialization modes those are similar to kafka-connector
2.Add metrics in source phase
Co-authored-by: 高思伟 <si...@amh-group.com>
---
README.md | 38 +++-
pom.xml | 14 ++
.../flink/legacy/RocketMQSourceFunction.java | 232 +++++++++++++++------
.../OffsetResetStrategy.java} | 19 +-
.../TestUtils.java => config/StartupMode.java} | 20 +-
.../SimpleStringDeserializationSchema.java | 39 ++++
.../flink/legacy/common/util/MetricUtils.java | 18 ++
.../flink/legacy/common/util/TestUtils.java | 7 +
.../sourceFunction/RocketMQSourceFunctionTest.java | 88 ++++++++
9 files changed, 387 insertions(+), 88 deletions(-)
diff --git a/README.md b/README.md
index fa53dd7..be889eb 100644
--- a/README.md
+++ b/README.md
@@ -73,7 +73,12 @@ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironm
Properties producerProps = new Properties();
producerProps.setProperty(RocketMQConfig.NAME_SERVER_ADDR, "localhost:9876");
- env.addSource(new RocketMQSourceFunction(new SimpleKeyValueDeserializationSchema("id", "address"), consumerProps))
+ RocketMQSourceFunction<Map<Object,Object>> source = new RocketMQSourceFunction(
+ new SimpleKeyValueDeserializationSchema("id", "address"), consumerProps);
+ // use group offsets.
+ // If there is no committed offset,consumer would start from the latest offset.
+ source.setStartFromGroupOffsets(OffsetResetStrategy.LATEST);
+ env.addSource()
.name("rocketmq-source")
.setParallelism(2)
.process(new ProcessFunction<Map, Map>() {
@@ -123,13 +128,40 @@ The following configurations are all from the class `org.apache.rocketmq.flink.l
| consumer.group | consumer group *Required* | null |
| consumer.topic | consumer topic *Required* | null |
| consumer.tag | consumer topic tag | * |
-| consumer.offset.reset.to | what to do when there is no initial offset on the server | latest/earliest/timestamp |
-| consumer.offset.from.timestamp | the timestamp when `consumer.offset.reset.to=timestamp` was set | `System.currentTimeMillis()` |
| consumer.offset.persist.interval | auto commit offset interval | 5000 |
| consumer.pull.thread.pool.size | consumer pull thread pool size | 20 |
| consumer.batch.size | consumer messages batch size | 32 |
| consumer.delay.when.message.not.found | the delay time when messages were not found | 10 |
+### Consumer Strategy
+
+```java
+RocketMQSourceFunction<String> source = new RocketMQSourceFunction<>(
+ new SimpleStringDeserializationSchema(), props);
+HashMap<MessageQueue, Long> brokerMap = new HashMap<>();
+brokerMap.put(new MessageQueue("tp_driver_tag_sync_back", "broker-a", 1), 201L);
+brokerMap.put(new MessageQueue("tp_driver_tag_sync_back", "broker-c", 3), 123L);
+source.setStartFromSpecificOffsets(brokerMap);
+```
+RocketMQSourceFunction offer five initialization policies
+* setStartFromEarliest
+* setStartFromLatest
+* setStartFromTimeStamp with timestamp
+* setStartFromGroupOffsets with `OffsetResetStrategy`
+* setStartFromSpecificOffsets
+
+| STRATEGY | DESCRIPTION |
+| --------------------------- | ------------------------------------------------------------ |
+| EARLIEST | consume from the earliest offset after restart with no state |
+| LATEST | consume from the latest offset after restart with no state |
+| TIMESTAMP | consume from the closest timestamp of data in each broker's queue |
+| GROUP_OFFSETS with LATEST | If broker has the committed offset then consume from the next else consume from the latest offset |
+| GROUP_OFFSETS with EARLIEST | If broker has the committed offset ,consume from the next ,otherwise consume from the earlist offset.It's useful when server expand broker |
+| SPECIFIC_OFFSETS | consumer from the specificOffsets in broker's queues.Group offsets will be returned from those broker's queues whose didn't be specified |
+
+**Attention**
+
+Only if Flink job starts with none state, these strategies are effective. If the job recovers from the checkpoint, the offset would intialize from the stored data.
## RocketMQ SQL Connector
diff --git a/pom.xml b/pom.xml
index 1b725e1..8617bd2 100644
--- a/pom.xml
+++ b/pom.xml
@@ -315,6 +315,20 @@
</execution>
</executions>
</plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-source-plugin</artifactId>
+ <version>3.0.1</version>
+ <executions>
+ <execution>
+ <id>attach-sources</id>
+ <phase>compile</phase>
+ <goals>
+ <goal>jar-no-fork</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
</plugins>
</build>
</project>
diff --git a/src/main/java/org/apache/rocketmq/flink/legacy/RocketMQSourceFunction.java b/src/main/java/org/apache/rocketmq/flink/legacy/RocketMQSourceFunction.java
index db2ab15..b078056 100644
--- a/src/main/java/org/apache/rocketmq/flink/legacy/RocketMQSourceFunction.java
+++ b/src/main/java/org/apache/rocketmq/flink/legacy/RocketMQSourceFunction.java
@@ -23,6 +23,8 @@ import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.flink.legacy.common.config.OffsetResetStrategy;
+import org.apache.rocketmq.flink.legacy.common.config.StartupMode;
import org.apache.rocketmq.flink.legacy.common.serialization.KeyValueDeserializationSchema;
import org.apache.rocketmq.flink.legacy.common.util.MetricUtils;
import org.apache.rocketmq.flink.legacy.common.util.RetryUtil;
@@ -76,13 +78,8 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
import static org.apache.rocketmq.flink.legacy.RocketMQConfig.CONSUMER_BATCH_SIZE;
-import static org.apache.rocketmq.flink.legacy.RocketMQConfig.CONSUMER_OFFSET_EARLIEST;
-import static org.apache.rocketmq.flink.legacy.RocketMQConfig.CONSUMER_OFFSET_LATEST;
-import static org.apache.rocketmq.flink.legacy.RocketMQConfig.CONSUMER_OFFSET_TIMESTAMP;
import static org.apache.rocketmq.flink.legacy.RocketMQConfig.DEFAULT_CONSUMER_BATCH_SIZE;
-import static org.apache.rocketmq.flink.legacy.RocketMQConfig.DEFAULT_START_MESSAGE_OFFSET;
import static org.apache.rocketmq.flink.legacy.common.util.RocketMQUtils.getInteger;
-import static org.apache.rocketmq.flink.legacy.common.util.RocketMQUtils.getLong;
/**
* The RocketMQSource is based on RocketMQ pull consumer mode, and provides exactly once reliability
@@ -116,12 +113,33 @@ public class RocketMQSourceFunction<OUT> extends RichParallelSourceFunction<OUT>
private Properties props;
private String topic;
private String group;
- private long startMessageOffset;
private transient volatile boolean restored;
private transient boolean enableCheckpoint;
private volatile Object checkPointLock;
private Meter tpsMetric;
+ private MetricUtils.TimestampGauge fetchDelay = new MetricUtils.TimestampGauge();
+ private MetricUtils.TimestampGauge emitDelay = new MetricUtils.TimestampGauge();
+
+ /** The startup mode for the consumer (default is {@link StartupMode#GROUP_OFFSETS}). */
+ private StartupMode startMode = StartupMode.GROUP_OFFSETS;
+
+ /**
+ * If StartupMode#GROUP_OFFSETS has no commit offset.OffsetResetStrategy would offer init
+ * strategy.
+ */
+ private OffsetResetStrategy offsetResetStrategy = OffsetResetStrategy.LATEST;
+
+ /**
+ * Specific startup offsets; only relevant when startup mode is {@link
+ * StartupMode#SPECIFIC_OFFSETS}.
+ */
+ private Map<MessageQueue, Long> specificStartupOffsets;
+
+ /**
+ * Specific startup offsets; only relevant when startup mode is {@link StartupMode#TIMESTAMP}.
+ */
+ private long specificTimeStamp;
public RocketMQSourceFunction(KeyValueDeserializationSchema<OUT> schema, Properties props) {
this.schema = schema;
@@ -135,11 +153,6 @@ public class RocketMQSourceFunction<OUT> extends RichParallelSourceFunction<OUT>
this.topic = props.getProperty(RocketMQConfig.CONSUMER_TOPIC);
this.group = props.getProperty(RocketMQConfig.CONSUMER_GROUP);
- this.startMessageOffset =
- props.containsKey(RocketMQConfig.CONSUMER_START_MESSAGE_OFFSET)
- ? Long.parseLong(
- props.getProperty(RocketMQConfig.CONSUMER_START_MESSAGE_OFFSET))
- : DEFAULT_START_MESSAGE_OFFSET;
Validate.notEmpty(topic, "Consumer topic can not be empty");
Validate.notEmpty(group, "Consumer group can not be empty");
@@ -214,14 +227,13 @@ public class RocketMQSourceFunction<OUT> extends RichParallelSourceFunction<OUT>
getRuntimeContext()
.getMetricGroup()
.meter(MetricUtils.METRICS_TPS, new MeterView(outputCounter, 60));
- }
- @Override
- public void run(SourceContext context) throws Exception {
- String sql = props.getProperty(RocketMQConfig.CONSUMER_SQL);
- String tag =
- props.getProperty(RocketMQConfig.CONSUMER_TAG, RocketMQConfig.DEFAULT_CONSUMER_TAG);
- int pullBatchSize = getInteger(props, CONSUMER_BATCH_SIZE, DEFAULT_CONSUMER_BATCH_SIZE);
+ getRuntimeContext()
+ .getMetricGroup()
+ .gauge(MetricUtils.CURRENT_FETCH_EVENT_TIME_LAG, fetchDelay);
+ getRuntimeContext()
+ .getMetricGroup()
+ .gauge(MetricUtils.CURRENT_EMIT_EVENT_TIME_LAG, emitDelay);
final RuntimeContext ctx = getRuntimeContext();
// The lock that guarantees that record emission and state updates are atomic,
@@ -229,7 +241,22 @@ public class RocketMQSourceFunction<OUT> extends RichParallelSourceFunction<OUT>
int taskNumber = ctx.getNumberOfParallelSubtasks();
int taskIndex = ctx.getIndexOfThisSubtask();
log.info("Source run, NumberOfTotalTask={}, IndexOfThisSubTask={}", taskNumber, taskIndex);
+ Collection<MessageQueue> totalQueues = consumer.fetchSubscribeMessageQueues(topic);
+ messageQueues =
+ RocketMQUtils.allocate(totalQueues, taskNumber, ctx.getIndexOfThisSubtask());
+ // If the job recovers from the state, the state has already contained the offsets of last
+ // commit.
+ if (!restored) {
+ initOffsets(messageQueues);
+ }
+ }
+ @Override
+ public void run(SourceContext context) throws Exception {
+ String sql = props.getProperty(RocketMQConfig.CONSUMER_SQL);
+ String tag =
+ props.getProperty(RocketMQConfig.CONSUMER_TAG, RocketMQConfig.DEFAULT_CONSUMER_TAG);
+ int pullBatchSize = getInteger(props, CONSUMER_BATCH_SIZE, DEFAULT_CONSUMER_BATCH_SIZE);
timer.scheduleAtFixedRate(
() -> {
// context.emitWatermark(waterMarkPerQueue.getCurrentWatermark());
@@ -238,10 +265,6 @@ public class RocketMQSourceFunction<OUT> extends RichParallelSourceFunction<OUT>
5,
5,
TimeUnit.SECONDS);
-
- Collection<MessageQueue> totalQueues = consumer.fetchSubscribeMessageQueues(topic);
- messageQueues =
- RocketMQUtils.allocate(totalQueues, taskNumber, ctx.getIndexOfThisSubtask());
for (MessageQueue mq : messageQueues) {
this.executor.execute(
() ->
@@ -249,8 +272,7 @@ public class RocketMQSourceFunction<OUT> extends RichParallelSourceFunction<OUT>
() -> {
while (runningChecker.isRunning()) {
try {
- long offset = getMessageQueueOffset(mq);
-
+ long offset = offsetTable.get(mq);
PullResult pullResult;
if (StringUtils.isEmpty(sql)) {
pullResult =
@@ -271,6 +293,7 @@ public class RocketMQSourceFunction<OUT> extends RichParallelSourceFunction<OUT>
case FOUND:
List<MessageExt> messages =
pullResult.getMsgFoundList();
+ long fetchTime = System.currentTimeMillis();
for (MessageExt msg : messages) {
byte[] key =
msg.getKeys() != null
@@ -299,12 +322,24 @@ public class RocketMQSourceFunction<OUT> extends RichParallelSourceFunction<OUT>
context.collectWithTimestamp(
data,
msg.getBornTimestamp());
+ long emitTime =
+ System.currentTimeMillis();
// update max eventTime per queue
// waterMarkPerQueue.extractTimestamp(mq, msg.getBornTimestamp());
waterMarkForAll.extractTimestamp(
msg.getBornTimestamp());
tpsMetric.markEvent();
+ long eventTime =
+ msg.getStoreTimestamp();
+ fetchDelay.report(
+ Math.abs(
+ fetchTime
+ - eventTime));
+ emitDelay.report(
+ Math.abs(
+ emitTime
+ - eventTime));
}
}
found = true;
@@ -359,46 +394,121 @@ public class RocketMQSourceFunction<OUT> extends RichParallelSourceFunction<OUT>
}
}
- private long getMessageQueueOffset(MessageQueue mq) throws MQClientException {
- Long offset = offsetTable.get(mq);
- if (offset != null) {
- return offset;
- }
- // restoredOffsets(unionOffsetStates) is the restored global union state;
- // should only snapshot mqs that actually belong to us
- if (startMessageOffset == DEFAULT_START_MESSAGE_OFFSET) {
- // fetchConsumeOffset from broker
- offset = consumer.fetchConsumeOffset(mq, false);
- if (!restored && offset < 0) {
- String initialOffset =
- props.getProperty(
- RocketMQConfig.CONSUMER_OFFSET_RESET_TO, CONSUMER_OFFSET_LATEST);
- switch (initialOffset) {
- case CONSUMER_OFFSET_EARLIEST:
- offset = consumer.minOffset(mq);
- break;
- case CONSUMER_OFFSET_LATEST:
- offset = consumer.maxOffset(mq);
- break;
- case CONSUMER_OFFSET_TIMESTAMP:
- offset =
- consumer.searchOffset(
+ /**
+ * only flink job start with no state can init offsets from broker
+ *
+ * @param messageQueues
+ * @throws MQClientException
+ */
+ private void initOffsets(List<MessageQueue> messageQueues) throws MQClientException {
+ for (MessageQueue mq : messageQueues) {
+ long offset;
+ switch (startMode) {
+ case LATEST:
+ offset = consumer.maxOffset(mq);
+ break;
+ case EARLIEST:
+ offset = consumer.minOffset(mq);
+ break;
+ case GROUP_OFFSETS:
+ offset = consumer.fetchConsumeOffset(mq, false);
+ // the min offset return if consumer group first join,return a negative number
+ // if
+ // catch exception when fetch from broker.
+ // If you want consumer from earliest,please use OffsetResetStrategy.EARLIEST
+ if (offset <= 0) {
+ switch (offsetResetStrategy) {
+ case LATEST:
+ offset = consumer.maxOffset(mq);
+ log.info(
+ "current consumer thread:{} has no committed offset,use Strategy:{} instead",
mq,
- getLong(
- props,
- RocketMQConfig.CONSUMER_OFFSET_FROM_TIMESTAMP,
- System.currentTimeMillis()));
- break;
- default:
- throw new IllegalArgumentException(
- "Unknown value for CONSUMER_OFFSET_RESET_TO.");
- }
+ offsetResetStrategy);
+ break;
+ case EARLIEST:
+ log.info(
+ "current consumer thread:{} has no committed offset,use Strategy:{} instead",
+ mq,
+ offsetResetStrategy);
+ offset = consumer.minOffset(mq);
+ break;
+ default:
+ break;
+ }
+ }
+ break;
+ case TIMESTAMP:
+ offset = consumer.searchOffset(mq, specificTimeStamp);
+ break;
+ case SPECIFIC_OFFSETS:
+ if (specificStartupOffsets == null) {
+ throw new RuntimeException(
+ "StartMode is specific_offsets.But none offsets has been specified");
+ }
+ Long specificOffset = specificStartupOffsets.get(mq);
+ if (specificOffset != null) {
+ offset = specificOffset;
+ } else {
+ offset = consumer.fetchConsumeOffset(mq, false);
+ }
+ break;
+ default:
+ throw new IllegalArgumentException(
+ "current startMode is not supported" + startMode);
}
- } else {
- offset = startMessageOffset;
+ log.info(
+ "current consumer queue:{} start from offset of: {}",
+ mq.getBrokerName() + "-" + mq.getQueueId(),
+ offset);
+ offsetTable.put(mq, offset);
}
- offsetTable.put(mq, offset);
- return offsetTable.get(mq);
+ }
+
+ /** consume from the min offset at every restart with no state */
+ public RocketMQSourceFunction<OUT> setStartFromEarliest() {
+ this.startMode = StartupMode.EARLIEST;
+ return this;
+ }
+
+ /** consume from the max offset of each broker's queue at every restart with no state */
+ public RocketMQSourceFunction<OUT> setStartFromLatest() {
+ this.startMode = StartupMode.LATEST;
+ return this;
+ }
+
+ /** consume from the closest offset */
+ public RocketMQSourceFunction<OUT> setStartFromTimeStamp(long timeStamp) {
+ this.startMode = StartupMode.TIMESTAMP;
+ this.specificTimeStamp = timeStamp;
+ return this;
+ }
+
+ /** consume from the group offsets those was stored in brokers. */
+ public RocketMQSourceFunction<OUT> setStartFromGroupOffsets() {
+ this.startMode = StartupMode.GROUP_OFFSETS;
+ return this;
+ }
+
+ /**
+ * consume from the group offsets those was stored in brokers. If there is no committed
+ * offset,#{@link OffsetResetStrategy} would provide initialization policy.
+ */
+ public RocketMQSourceFunction<OUT> setStartFromGroupOffsets(
+ OffsetResetStrategy offsetResetStrategy) {
+ this.startMode = StartupMode.GROUP_OFFSETS;
+ this.offsetResetStrategy = offsetResetStrategy;
+ return this;
+ }
+
+ /**
+ * consume from the specific offset. Group offsets is enable while the broker didn't specify
+ * offset.
+ */
+ public RocketMQSourceFunction<OUT> setStartFromSpecificOffsets(
+ Map<MessageQueue, Long> specificOffsets) {
+ this.specificStartupOffsets = specificOffsets;
+ this.startMode = StartupMode.SPECIFIC_OFFSETS;
+ return this;
}
private void updateMessageQueueOffset(MessageQueue mq, long offset) throws MQClientException {
diff --git a/src/main/java/org/apache/rocketmq/flink/legacy/common/util/TestUtils.java b/src/main/java/org/apache/rocketmq/flink/legacy/common/config/OffsetResetStrategy.java
similarity index 65%
copy from src/main/java/org/apache/rocketmq/flink/legacy/common/util/TestUtils.java
copy to src/main/java/org/apache/rocketmq/flink/legacy/common/config/OffsetResetStrategy.java
index 70c26d9..a48e6d4 100644
--- a/src/main/java/org/apache/rocketmq/flink/legacy/common/util/TestUtils.java
+++ b/src/main/java/org/apache/rocketmq/flink/legacy/common/config/OffsetResetStrategy.java
@@ -14,18 +14,13 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.rocketmq.flink.legacy.common.util;
-import java.lang.reflect.Field;
+package org.apache.rocketmq.flink.legacy.common.config;
-public class TestUtils {
- public static void setFieldValue(Object obj, String fieldName, Object value) {
- try {
- Field field = obj.getClass().getDeclaredField(fieldName);
- field.setAccessible(true);
- field.set(obj, value);
- } catch (Exception e) {
- e.printStackTrace();
- }
- }
+/** Config for #{@link StartupMode#GROUP_OFFSETS}. */
+public enum OffsetResetStrategy {
+ /** If group offsets is not found,the latest offset would be set to start consumer */
+ LATEST,
+ /** If group offsets is not found,the earliest offset would be set to start consumer */
+ EARLIEST
}
diff --git a/src/main/java/org/apache/rocketmq/flink/legacy/common/util/TestUtils.java b/src/main/java/org/apache/rocketmq/flink/legacy/common/config/StartupMode.java
similarity index 65%
copy from src/main/java/org/apache/rocketmq/flink/legacy/common/util/TestUtils.java
copy to src/main/java/org/apache/rocketmq/flink/legacy/common/config/StartupMode.java
index 70c26d9..163dae4 100644
--- a/src/main/java/org/apache/rocketmq/flink/legacy/common/util/TestUtils.java
+++ b/src/main/java/org/apache/rocketmq/flink/legacy/common/config/StartupMode.java
@@ -14,18 +14,14 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.rocketmq.flink.legacy.common.util;
-import java.lang.reflect.Field;
+package org.apache.rocketmq.flink.legacy.common.config;
-public class TestUtils {
- public static void setFieldValue(Object obj, String fieldName, Object value) {
- try {
- Field field = obj.getClass().getDeclaredField(fieldName);
- field.setAccessible(true);
- field.set(obj, value);
- } catch (Exception e) {
- e.printStackTrace();
- }
- }
+/** RocketMQ startup mode. */
+public enum StartupMode {
+ EARLIEST,
+ LATEST,
+ GROUP_OFFSETS,
+ TIMESTAMP,
+ SPECIFIC_OFFSETS
}
diff --git a/src/main/java/org/apache/rocketmq/flink/legacy/common/serialization/SimpleStringDeserializationSchema.java b/src/main/java/org/apache/rocketmq/flink/legacy/common/serialization/SimpleStringDeserializationSchema.java
new file mode 100644
index 0000000..f59b961
--- /dev/null
+++ b/src/main/java/org/apache/rocketmq/flink/legacy/common/serialization/SimpleStringDeserializationSchema.java
@@ -0,0 +1,39 @@
+/*
+ * 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.rocketmq.flink.legacy.common.serialization;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+
+import java.nio.charset.StandardCharsets;
+
+/** deserialize the message body to string */
+public class SimpleStringDeserializationSchema implements KeyValueDeserializationSchema<String> {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public String deserializeKeyAndValue(byte[] key, byte[] value) {
+ String v = value != null ? new String(value, StandardCharsets.UTF_8) : null;
+ return v;
+ }
+
+ @Override
+ public TypeInformation<String> getProducedType() {
+ return TypeInformation.of(String.class);
+ }
+}
diff --git a/src/main/java/org/apache/rocketmq/flink/legacy/common/util/MetricUtils.java b/src/main/java/org/apache/rocketmq/flink/legacy/common/util/MetricUtils.java
index 6c62cc4..17b4dda 100644
--- a/src/main/java/org/apache/rocketmq/flink/legacy/common/util/MetricUtils.java
+++ b/src/main/java/org/apache/rocketmq/flink/legacy/common/util/MetricUtils.java
@@ -23,10 +23,15 @@ import org.apache.flink.metrics.Meter;
import org.apache.flink.metrics.MeterView;
import org.apache.flink.metrics.SimpleCounter;
+import java.io.Serializable;
+
/** RocketMQ connector metrics. */
public class MetricUtils {
public static final String METRICS_TPS = "tps";
+ // https://cwiki.apache.org/confluence/display/FLINK/FLIP-33%3A+Standardize+Connector+Metrics
+ public static final String CURRENT_FETCH_EVENT_TIME_LAG = "currentFetchEventTimeLag";
+ public static final String CURRENT_EMIT_EVENT_TIME_LAG = "currentEmitEventTimeLag";
private static final String METRIC_GROUP_SINK = "sink";
private static final String METRICS_SINK_IN_TPS = "inTps";
@@ -84,4 +89,17 @@ public class MetricUtils {
return value;
}
}
+
+ public static class TimestampGauge implements Gauge<Long>, Serializable {
+ private Long value;
+
+ public void report(long delay) {
+ this.value = delay;
+ }
+
+ @Override
+ public Long getValue() {
+ return value;
+ }
+ }
}
diff --git a/src/main/java/org/apache/rocketmq/flink/legacy/common/util/TestUtils.java b/src/main/java/org/apache/rocketmq/flink/legacy/common/util/TestUtils.java
index 70c26d9..00a24c9 100644
--- a/src/main/java/org/apache/rocketmq/flink/legacy/common/util/TestUtils.java
+++ b/src/main/java/org/apache/rocketmq/flink/legacy/common/util/TestUtils.java
@@ -28,4 +28,11 @@ public class TestUtils {
e.printStackTrace();
}
}
+
+ public static Object getFieldValue(Object obj, String fieldName)
+ throws NoSuchFieldException, IllegalAccessException {
+ Field field = obj.getClass().getDeclaredField(fieldName);
+ field.setAccessible(true);
+ return field.get(obj);
+ }
}
diff --git a/src/test/java/org/apache/rocketmq/flink/legacy/sourceFunction/RocketMQSourceFunctionTest.java b/src/test/java/org/apache/rocketmq/flink/legacy/sourceFunction/RocketMQSourceFunctionTest.java
new file mode 100644
index 0000000..8b7b44f
--- /dev/null
+++ b/src/test/java/org/apache/rocketmq/flink/legacy/sourceFunction/RocketMQSourceFunctionTest.java
@@ -0,0 +1,88 @@
+/*
+ * 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.rocketmq.flink.legacy.sourceFunction;
+
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.flink.legacy.RocketMQConfig;
+import org.apache.rocketmq.flink.legacy.RocketMQSourceFunction;
+import org.apache.rocketmq.flink.legacy.common.config.OffsetResetStrategy;
+import org.apache.rocketmq.flink.legacy.common.config.StartupMode;
+import org.apache.rocketmq.flink.legacy.common.serialization.SimpleStringDeserializationSchema;
+
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ConcurrentHashMap;
+
+import static org.apache.rocketmq.flink.legacy.RocketMQConfig.DEFAULT_CONSUMER_TAG;
+import static org.apache.rocketmq.flink.legacy.common.util.TestUtils.getFieldValue;
+import static org.apache.rocketmq.flink.legacy.common.util.TestUtils.setFieldValue;
+import static org.junit.Assert.assertEquals;
+
+/** Tests for {@link RocketMQSourceFunction}. */
+public class RocketMQSourceFunctionTest {
+
+ @Test
+ public void testSetStartupMode() throws NoSuchFieldException, IllegalAccessException {
+ RocketMQSourceFunction<String> source =
+ new RocketMQSourceFunction<>(
+ new SimpleStringDeserializationSchema(), new Properties());
+ assertEquals(StartupMode.GROUP_OFFSETS, getFieldValue(source, "startMode"));
+ source.setStartFromEarliest();
+ assertEquals(StartupMode.EARLIEST, getFieldValue(source, "startMode"));
+ source.setStartFromLatest();
+ assertEquals(StartupMode.LATEST, getFieldValue(source, "startMode"));
+ source.setStartFromTimeStamp(0L);
+ assertEquals(StartupMode.TIMESTAMP, getFieldValue(source, "startMode"));
+ source.setStartFromSpecificOffsets(null);
+ assertEquals(StartupMode.SPECIFIC_OFFSETS, getFieldValue(source, "startMode"));
+ source.setStartFromGroupOffsets();
+ assertEquals(StartupMode.GROUP_OFFSETS, getFieldValue(source, "startMode"));
+ assertEquals(OffsetResetStrategy.LATEST, getFieldValue(source, "offsetResetStrategy"));
+ source.setStartFromGroupOffsets(OffsetResetStrategy.EARLIEST);
+ assertEquals(OffsetResetStrategy.EARLIEST, getFieldValue(source, "offsetResetStrategy"));
+ }
+
+ @Test
+ public void testRestartFromCheckpoint() throws Exception {
+ Properties properties = new Properties();
+ properties.setProperty(RocketMQConfig.CONSUMER_GROUP, "${ConsumerGroup}");
+ properties.setProperty(RocketMQConfig.CONSUMER_TOPIC, "${SourceTopic}");
+ properties.setProperty(RocketMQConfig.CONSUMER_TAG, DEFAULT_CONSUMER_TAG);
+ RocketMQSourceFunction<String> source =
+ new RocketMQSourceFunction<>(new SimpleStringDeserializationSchema(), properties);
+ source.setStartFromLatest();
+ setFieldValue(source, "restored", true);
+ HashMap<MessageQueue, Long> map = new HashMap<>();
+ map.put(new MessageQueue("tpc", "broker-0", 0), 20L);
+ map.put(new MessageQueue("tpc", "broker-0", 1), 21L);
+ map.put(new MessageQueue("tpc", "broker-1", 0), 30L);
+ map.put(new MessageQueue("tpc", "broker-1", 1), 31L);
+ setFieldValue(source, "restoredOffsets", map);
+ setFieldValue(source, "offsetTable", new ConcurrentHashMap<>());
+ source.initOffsetTableFromRestoredOffsets();
+ Map<MessageQueue, Long> offsetTable = (Map) getFieldValue(source, "offsetTable");
+ for (Map.Entry<MessageQueue, Long> entry : map.entrySet()) {
+ assertEquals(offsetTable.containsKey(entry.getKey()), true);
+ assertEquals(offsetTable.containsValue(entry.getValue()), true);
+ }
+ }
+}