[flink] branch master updated: [FLINK-29859][e2e] Running TPC-DS with adaptive batch scheduler supports custom errors check.

[zh...@apache.org]

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

zhuzh pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/master by this push:
     new 28e75620db7 [FLINK-29859][e2e] Running TPC-DS with adaptive batch scheduler supports custom errors check.
28e75620db7 is described below

commit 28e75620db7b0794ff1c19f8928ce7d33516ba64
Author: JunRuiLee <jr...@gmail.com>
AuthorDate: Mon Feb 27 16:42:45 2023 +0800

    [FLINK-29859][e2e] Running TPC-DS with adaptive batch scheduler supports custom errors check.
    
    This closes #22029.
---
 flink-end-to-end-tests/test-scripts/common.sh     | 79 +++++++++++++++--------
 flink-end-to-end-tests/test-scripts/test_tpcds.sh |  8 ++-
 2 files changed, 60 insertions(+), 27 deletions(-)

diff --git a/flink-end-to-end-tests/test-scripts/common.sh b/flink-end-to-end-tests/test-scripts/common.sh
index 44f0ed4c5df..1bef68bbefb 100644
--- a/flink-end-to-end-tests/test-scripts/common.sh
+++ b/flink-end-to-end-tests/test-scripts/common.sh
@@ -355,33 +355,60 @@ function wait_for_number_of_running_tms {
 }
 
 function check_logs_for_errors {
+  internal_check_logs_for_errors
+}
+
+# check logs for errors, the arguments are the additional allowed errors
+function internal_check_logs_for_errors {
   echo "Checking for errors..."
-  error_count=$(grep -rv "GroupCoordinatorNotAvailableException" $FLINK_LOG_DIR \
-      | grep -v "RetriableCommitFailedException" \
-      | grep -v "NoAvailableBrokersException" \
-      | grep -v "Async Kafka commit failed" \
-      | grep -v "DisconnectException" \
-      | grep -v "Cannot connect to ResourceManager right now" \
-      | grep -v "AskTimeoutException" \
-      | grep -v "Error while loading kafka-version.properties" \
-      | grep -v "WARN  akka.remote.transport.netty.NettyTransport" \
-      | grep -v "WARN  org.jboss.netty.channel.DefaultChannelPipeline" \
-      | grep -v "jvm-exit-on-fatal-error" \
-      | grep -v 'INFO.*AWSErrorCode' \
-      | grep -v "RejectedExecutionException" \
-      | grep -v "An exception was thrown by an exception handler" \
-      | grep -v "java.lang.NoClassDefFoundError: org/apache/hadoop/yarn/exceptions/YarnException" \
-      | grep -v "java.lang.NoClassDefFoundError: org/apache/hadoop/conf/Configuration" \
-      | grep -v "org.apache.commons.beanutils.FluentPropertyBeanIntrospector.*Error when creating PropertyDescriptor.*org.apache.commons.configuration2.AbstractConfiguration.setProperty(java.lang.String,java.lang.Object)! Ignoring this property." \
-      | grep -v "Error while loading kafka-version.properties :null" \
-      | grep -v "[Terror] modules" \
-      | grep -v "HeapDumpOnOutOfMemoryError" \
-      | grep -v "error_prone_annotations" \
-      | grep -v "Error sending fetch request" \
-      | grep -v "WARN  akka.remote.ReliableDeliverySupervisor" \
-      | grep -v "Options.*error_*" \
-      | grep -v "not packaged with this application" \
-      | grep -ic "error" || true)
+
+  local additional_allowed_errors=()
+  local index=0
+  for error in "$@"; do
+    additional_allowed_errors[index]="$error"
+    index=$index+1
+  done
+
+  local default_allowed_errors=("GroupCoordinatorNotAvailableException" \
+  "RetriableCommitFailedException" \
+  "NoAvailableBrokersException" \
+  "Async Kafka commit failed" \
+  "DisconnectException" \
+  "Cannot connect to ResourceManager right now" \
+  "AskTimeoutException" \
+  "Error while loading kafka-version.properties" \
+  "WARN  akka.remote.transport.netty.NettyTransport" \
+  "WARN  org.jboss.netty.channel.DefaultChannelPipeline" \
+  "jvm-exit-on-fatal-error" \
+  'INFO.*AWSErrorCode' \
+  "RejectedExecutionException" \
+  "An exception was thrown by an exception handler" \
+  "java.lang.NoClassDefFoundError: org/apache/hadoop/yarn/exceptions/YarnException" \
+  "java.lang.NoClassDefFoundError: org/apache/hadoop/conf/Configuration" \
+  "org.apache.commons.beanutils.FluentPropertyBeanIntrospector.*Error when creating PropertyDescriptor.*org.apache.commons.configuration2.AbstractConfiguration.setProperty(java.lang.String,java.lang.Object)! Ignoring this property." \
+  "Error while loading kafka-version.properties :null" \
+  "[Terror] modules" \
+  "HeapDumpOnOutOfMemoryError" \
+  "error_prone_annotations" \
+  "Error sending fetch request" \
+  "WARN  akka.remote.ReliableDeliverySupervisor" \
+  "Options.*error_*" \
+  "not packaged with this application")
+
+  local all_allowed_errors=("${default_allowed_errors[@]}" "${additional_allowed_errors[@]}")
+
+  # generate the grep command
+  local grep_command=""
+  for error in "${all_allowed_errors[@]}"; do
+    if [[ $grep_command == "" ]]; then
+      grep_command="grep -rv \"$error\" $FLINK_LOG_DIR"
+    else
+      grep_command="$grep_command | grep -v \"$error\""
+    fi
+  done
+  grep_command="$grep_command | grep -ic \"error\" || true"
+
+  error_count=$(eval "$grep_command")
   if [[ ${error_count} -gt 0 ]]; then
     echo "Found error in log files; printing first 500 lines; see full logs for details:"
     find $FLINK_LOG_DIR/ -type f -exec head -n 500 {} \;
diff --git a/flink-end-to-end-tests/test-scripts/test_tpcds.sh b/flink-end-to-end-tests/test-scripts/test_tpcds.sh
index 8797f7a2e0f..c46db4b4396 100755
--- a/flink-end-to-end-tests/test-scripts/test_tpcds.sh
+++ b/flink-end-to-end-tests/test-scripts/test_tpcds.sh
@@ -120,6 +120,12 @@ function check_logs_for_exceptions_for_adaptive_batch_scheduler {
     internal_check_logs_for_exceptions "${additional_allowed_exceptions[@]}"
 }
 
+function check_logs_for_errors_for_adaptive_batch_scheduler {
+    local additional_allowed_errors=("The handler of the request-complete-callback threw an exception: java.nio.channels.ClosedChannelException")
+
+    internal_check_logs_for_errors "${additional_allowed_errors[@]}"
+}
+
 SCHEDULER="${1:-Default}"
 ACTION="${2:-run_test}"
 
@@ -131,7 +137,7 @@ elif [ "${ACTION}" == "check_exceptions" ]; then
         exit 1
     fi
 
-    check_logs_for_errors
+    check_logs_for_errors_for_adaptive_batch_scheduler
     check_logs_for_exceptions_for_adaptive_batch_scheduler
     check_logs_for_non_empty_out_files
 else