[FLINK-10669][e2e] Fixed error & exception checking in logs

Previously the 'q' flag of grep make the last element of pipeline exit
as soon as the condition was satisfied. This resulted that if there were
multiple log files and the condition was met for the first file,
operations on subsequent files tried to write to closed pipe which
resulted in pipefail code and the if condition to be evaluated
incorrectly.

flink-end-to-end-tests/test-scripts/common.sh

@@ -300,7 +300,7 @@ function start_and_wait_for_tm {
 }
 
 function check_logs_for_errors {
- if grep -rv "GroupCoordinatorNotAvailableException" $FLINK_DIR/log \
+ error_count=$(grep -rv "GroupCoordinatorNotAvailableException" $FLINK_DIR/log \
  | grep -v "RetriableCommitFailedException" \
  | grep -v "NoAvailableBrokersException" \
  | grep -v "Async Kafka commit failed" \
@@ -315,32 +315,34 @@ function check_logs_for_errors {
  | 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.flink.fs.shaded.hadoop3.org.apache.commons.beanutils.FluentPropertyBeanIntrospector - Error when creating PropertyDescriptor for public final void org.apache.flink.fs.shaded.hadoop3.org.apache.commons.configuration2.AbstractConfiguration.setProperty(java.lang.String,java.lang.Object)! Ignoring this property." \
- | grep -iq "error"; then
+ | grep -ic "error")
+ if [[ ${error_count} -gt 0 ]]; then
  echo "Found error in log files:"
  cat $FLINK_DIR/log/*
  EXIT_CODE=1
  fi
 }
 
 function check_logs_for_exceptions {
- if grep -rv "GroupCoordinatorNotAvailableException" $FLINK_DIR/log \
- | grep -v "RetriableCommitFailedException" \
- | grep -v "NoAvailableBrokersException" \
- | grep -v "Async Kafka commit failed" \
- | grep -v "DisconnectException" \
- | grep -v "AskTimeoutException" \
- | grep -v "WARN akka.remote.transport.netty.NettyTransport" \
- | grep -v "WARN org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline" \
- | grep -v '^INFO:.*AWSErrorCode=\[400 Bad Request\].*ServiceEndpoint=\[https://.*\.s3\.amazonaws\.com\].*RequestType=\[HeadBucketRequest\]' \
- | grep -v "RejectedExecutionException" \
- | grep -v "An exception was thrown by an exception handler" \
- | grep -v "Caused by: java.lang.ClassNotFoundException: org.apache.hadoop.yarn.exceptions.YarnException" \
- | grep -v "Caused by: java.lang.ClassNotFoundException: org.apache.hadoop.conf.Configuration" \
- | grep -v "java.lang.NoClassDefFoundError: org/apache/hadoop/yarn/exceptions/YarnException" \
- | grep -v "java.lang.NoClassDefFoundError: org/apache/hadoop/conf/Configuration" \
- | grep -v "java.lang.Exception: Execution was suspended" \
- | grep -v "Caused by: java.lang.Exception: JobManager is shutting down" \
- | grep -iq "exception"; then
+ exception_count=$(grep -rv "GroupCoordinatorNotAvailableException" $FLINK_DIR/log \
+ | grep -v "RetriableCommitFailedException" \
+ | grep -v "NoAvailableBrokersException" \
+ | grep -v "Async Kafka commit failed" \
+ | grep -v "DisconnectException" \
+ | grep -v "AskTimeoutException" \
+ | grep -v "WARN akka.remote.transport.netty.NettyTransport" \
+ | grep -v "WARN org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline" \
+ | grep -v '^INFO:.*AWSErrorCode=\[400 Bad Request\].*ServiceEndpoint=\[https://.*\.s3\.amazonaws\.com\].*RequestType=\[HeadBucketRequest\]' \
+ | grep -v "RejectedExecutionException" \
+ | grep -v "An exception was thrown by an exception handler" \
+ | grep -v "Caused by: java.lang.ClassNotFoundException: org.apache.hadoop.yarn.exceptions.YarnException" \
+ | grep -v "Caused by: java.lang.ClassNotFoundException: org.apache.hadoop.conf.Configuration" \
+ | grep -v "java.lang.NoClassDefFoundError: org/apache/hadoop/yarn/exceptions/YarnException" \
+ | grep -v "java.lang.NoClassDefFoundError: org/apache/hadoop/conf/Configuration" \
+ | grep -v "java.lang.Exception: Execution was suspended" \
+ | grep -v "Caused by: java.lang.Exception: JobManager is shutting down" \
+ | grep -ic "exception")
+ if [[ ${exception_count} -gt 0 ]]; then
  echo "Found exception in log files:"
  cat $FLINK_DIR/log/*
  EXIT_CODE=1