SteNicholas commented on code in PR #46: URL: https://github.com/apache/rocketmq-flink/pull/46#discussion_r946305414
##########
src/main/java/org/apache/rocketmq/flink/legacy/RocketMQSourceFunction.java:
##########
@@ -241,91 +243,58 @@ public void run(SourceContext context) throws Exception {
() -> {
while (runningChecker.isRunning()) {
try {
- long offset =
getMessageQueueOffset(mq);
-
- PullResult pullResult = null;
+ getMessageQueueOffset(mq);
if (StringUtils.isEmpty(sql)) {
- pullResult =
-
consumer.pullBlockIfNotFound(
- mq, tag,
offset, pullBatchSize);
+ consumer.subscribe(topic,
tag);
Review Comment:
I agree with @keranbingaa. The `consumer.subscribe(topic, tag)` could be
move to the while loop.
##########
src/main/java/org/apache/rocketmq/flink/legacy/RocketMQSourceFunction.java:
##########
@@ -241,91 +243,58 @@ public void run(SourceContext context) throws Exception {
() -> {
while (runningChecker.isRunning()) {
try {
- long offset =
getMessageQueueOffset(mq);
-
- PullResult pullResult = null;
+ getMessageQueueOffset(mq);
if (StringUtils.isEmpty(sql)) {
- pullResult =
-
consumer.pullBlockIfNotFound(
- mq, tag,
offset, pullBatchSize);
+ consumer.subscribe(topic,
tag);
} else {
// pull with sql do not
support block pull.
- pullResult =
- consumer.pull(
- mq,
-
MessageSelector.bySql(sql),
- offset,
-
pullBatchSize);
+ consumer.subscribe(
+ topic,
MessageSelector.bySql(sql));
}
+
consumer.setPullBatchSize(pullBatchSize);
+ // consumer.seek(mq, offset);
boolean found = false;
- switch
(pullResult.getPullStatus()) {
- case FOUND:
- List<MessageExt>
messages =
-
pullResult.getMsgFoundList();
- for (MessageExt msg :
messages) {
- byte[] key =
-
msg.getKeys() != null
- ?
msg.getKeys()
-
.getBytes(
-
StandardCharsets
-
.UTF_8)
- :
null;
- byte[] value =
msg.getBody();
- OUT data =
-
schema.deserializeKeyAndValue(
-
key, value);
-
- // output and
state update are atomic
- synchronized
(checkPointLock) {
- log.debug(
-
msg.getMsgId()
-
+ "_"
-
+ msg
-
.getBrokerName()
-
+ " "
-
+ msg.getQueueId()
-
+ " "
-
+ msg
-
.getQueueOffset());
-
context.collectWithTimestamp(
- data,
-
msg.getBornTimestamp());
-
- // update max
eventTime per queue
- //
waterMarkPerQueue.extractTimestamp(mq, msg.getBornTimestamp());
-
waterMarkForAll.extractTimestamp(
-
msg.getBornTimestamp());
-
tpsMetric.markEvent();
- }
+ List<MessageExt> messages =
consumer.poll();
Review Comment:
This need to use `poll(timeout)` to avoid blocking the message consumer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
