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: dev-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org