You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tubemq.apache.org by go...@apache.org on 2020/12/09 03:15:15 UTC
[incubator-tubemq] branch master updated: [TUBEMQ-443]
TubemqSourceFunction class prints too many logs problem (#344)
This is an automated email from the ASF dual-hosted git repository.
gosonzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-tubemq.git
The following commit(s) were added to refs/heads/master by this push:
new 0d3f9c9 [TUBEMQ-443] TubemqSourceFunction class prints too many logs problem (#344)
0d3f9c9 is described below
commit 0d3f9c91828ecc745f53154d78bb6e52381f6dc4
Author: leno1001 <19...@qq.com>
AuthorDate: Wed Dec 9 11:15:05 2020 +0800
[TUBEMQ-443] TubemqSourceFunction class prints too many logs problem (#344)
Co-authored-by: 曹显乐 <xi...@vivo.com>
---
.../flink/connectors/tubemq/TubemqSourceFunction.java | 18 ++++++++++++------
1 file changed, 12 insertions(+), 6 deletions(-)
diff --git a/tubemq-connectors/tubemq-connector-flink/src/main/java/org/apache/flink/connectors/tubemq/TubemqSourceFunction.java b/tubemq-connectors/tubemq-connector-flink/src/main/java/org/apache/flink/connectors/tubemq/TubemqSourceFunction.java
index 2f375a6..c93e057 100644
--- a/tubemq-connectors/tubemq-connector-flink/src/main/java/org/apache/flink/connectors/tubemq/TubemqSourceFunction.java
+++ b/tubemq-connectors/tubemq-connector-flink/src/main/java/org/apache/flink/connectors/tubemq/TubemqSourceFunction.java
@@ -254,9 +254,12 @@ public class TubemqSourceFunction<T>
ConsumerResult consumeResult = messagePullConsumer.getMessage();
if (!consumeResult.isSuccess()) {
- LOG.info("Could not consume messages from tubemq (errcode: {}, " +
- "errmsg: {}).", consumeResult.getErrCode(),
- consumeResult.getErrMsg());
+ if (!(consumeResult.getErrCode() == 400 || consumeResult.getErrCode() == 404 ||
+ consumeResult.getErrCode() == 405 || consumeResult.getErrCode() == 406 ||
+ consumeResult.getErrCode() == 407 || consumeResult.getErrCode() == 408)) {
+ LOG.info("Could not consume messages from tubemq (errcode: {}, " + "errmsg: {}).",
+ consumeResult.getErrCode(), consumeResult.getErrMsg());
+ }
Duration idleTime =
Duration.between(lastConsumeInstant, Instant.now());
@@ -297,9 +300,12 @@ public class TubemqSourceFunction<T>
messagePullConsumer
.confirmConsume(consumeResult.getConfirmContext(), true);
if (!confirmResult.isSuccess()) {
- LOG.warn("Could not confirm messages to tubemq (errcode: {}, " +
- "errmsg: {}).", confirmResult.getErrCode(),
- confirmResult.getErrMsg());
+ if (!(confirmResult.getErrCode() == 400 || confirmResult.getErrCode() == 404 ||
+ confirmResult.getErrCode() == 405 || confirmResult.getErrCode() == 406 ||
+ confirmResult.getErrCode() == 407 || confirmResult.getErrCode() == 408)) {
+ LOG.warn("Could not confirm messages to tubemq (errcode: {}, " + "errmsg: {}).",
+ confirmResult.getErrCode(), confirmResult.getErrMsg());
+ }
}
}
}