[incubator-tubemq] 08/49: [TUBEMQ-443] TubemqSourceFunction class prints too many logs problem (#344)

[yuanbo]

This is an automated email from the ASF dual-hosted git repository.

yuanbo pushed a commit to branch TUBEMQ-421
in repository https://gitbox.apache.org/repos/asf/incubator-tubemq.git

commit aec25b501feb8f5a62832d0403e9589593f14bf3
Author: leno1001 <1960700...@qq.com>
AuthorDate: Wed Dec 9 11:15:05 2020 +0800

    [TUBEMQ-443] TubemqSourceFunction class prints too many logs problem (#344)
    
    Co-authored-by: 曹显乐 <xianle....@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());
+                }
             }
         }
     }