[FLINK-19295][yarn][tests] Exclude more meaningless akka shutdown errors (apache#13439)

Author: rmetzger

flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java

@@ -90,6 +90,7 @@
 import java.util.UUID;
 import java.util.concurrent.ConcurrentMap;
 import java.util.function.Supplier;
+import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
@@ -123,25 +124,26 @@ public abstract class YarnTestBase extends TestLogger {
 	};
 
 	/** These strings are white-listed, overriding the prohibited strings. */
-	protected static final String[] WHITELISTED_STRINGS = {
-		"akka.remote.RemoteTransportExceptionNoStackTrace",
+	protected static final Pattern[] WHITELISTED_STRINGS = {
+		Pattern.compile("akka\\.remote\\.RemoteTransportExceptionNoStackTrace"),
 		// workaround for annoying InterruptedException logging:
 		// https://issues.apache.org/jira/browse/YARN-1022
-		"java.lang.InterruptedException",
-		// very specific on purpose
-		"Remote connection to [null] failed with java.net.ConnectException: Connection refused",
-		"Remote connection to [null] failed with java.nio.channels.NotYetConnectedException",
-		"java.io.IOException: Connection reset by peer",
+		Pattern.compile("java\\.lang\\.InterruptedException"),
+		// very specific on purpose; whitelist meaningless exceptions that occur during akka shutdown:
+		Pattern.compile("Remote connection to \\[null\\] failed with java.net.ConnectException: Connection refused"),
+		Pattern.compile("Remote connection to \\[null\\] failed with java.nio.channels.NotYetConnectedException"),
+		Pattern.compile("java\\.io\\.IOException: Connection reset by peer"),
+		Pattern.compile("Association with remote system \\[akka.tcp://flink@[^]]+\\] has failed, address is now gated for \\[50\\] ms. Reason: \\[Association failed with \\[akka.tcp://flink@[^]]+\\]\\] Caused by: \\[java.net.ConnectException: Connection refused: [^]]+\\]"),
 
 		// filter out expected ResourceManagerException caused by intended shutdown request
-		YarnResourceManager.ERROR_MASSAGE_ON_SHUTDOWN_REQUEST,
+		Pattern.compile(YarnResourceManager.ERROR_MASSAGE_ON_SHUTDOWN_REQUEST),
 
 		// this can happen in Akka 2.4 on shutdown.
-		"java.util.concurrent.RejectedExecutionException: Worker has already been shutdown",
+		Pattern.compile("java\\.util\\.concurrent\\.RejectedExecutionException: Worker has already been shutdown"),
 
-		"org.apache.flink.util.FlinkException: Stopping JobMaster",
-		"org.apache.flink.util.FlinkException: JobManager is shutting down.",
-		"lost the leadership."
+		Pattern.compile("org\\.apache\\.flink.util\\.FlinkException: Stopping JobMaster"),
+		Pattern.compile("org\\.apache\\.flink.util\\.FlinkException: JobManager is shutting down\\."),
+		Pattern.compile("lost the leadership.")
 	};
 
 	// Temp directory which is deleted after the unit test.
@@ -390,7 +392,7 @@ private static void writeHDFSSiteConfigXML(Configuration coreSite, File targetFo
 	 * So always run "mvn clean" before running the tests here.
 	 *
 	 */
-	public static void ensureNoProhibitedStringInLogFiles(final String[] prohibited, final String[] whitelisted) {
+	public static void ensureNoProhibitedStringInLogFiles(final String[] prohibited, final Pattern[] whitelisted) {
 		File cwd = new File("target/" + YARN_CONFIGURATION.get(TEST_CLUSTER_NAME_KEY));
 		Assert.assertTrue("Expecting directory " + cwd.getAbsolutePath() + " to exist", cwd.exists());
 		Assert.assertTrue("Expecting directory " + cwd.getAbsolutePath() + " to be a directory", cwd.isDirectory());
@@ -400,17 +402,18 @@ public static void ensureNoProhibitedStringInLogFiles(final String[] prohibited,
 			@Override
 			public boolean accept(File dir, String name) {
 			// scan each file for prohibited strings.
-			File f = new File(dir.getAbsolutePath() + "/" + name);
+			File logFile = new File(dir.getAbsolutePath() + "/" + name);
 			try {
-				BufferingScanner scanner = new BufferingScanner(new Scanner(f), 10);
+				BufferingScanner scanner = new BufferingScanner(new Scanner(logFile), 10);
 				while (scanner.hasNextLine()) {
 					final String lineFromFile = scanner.nextLine();
 					for (String aProhibited : prohibited) {
 						if (lineFromFile.contains(aProhibited)) {
 
 							boolean whitelistedFound = false;
-							for (String white : whitelisted) {
-								if (lineFromFile.contains(white)) {
+							for (Pattern whitelistPattern : whitelisted) {
+								Matcher whitelistMatch = whitelistPattern.matcher(lineFromFile);
+								if (whitelistMatch.find()) {
 									whitelistedFound = true;
 									break;
 								}
@@ -419,7 +422,7 @@ public boolean accept(File dir, String name) {
 							if (!whitelistedFound) {
 								// logging in FATAL to see the actual message in CI tests.
 								Marker fatal = MarkerFactory.getMarker("FATAL");
-								LOG.error(fatal, "Prohibited String '{}' in '{}:{}'", aProhibited, f.getAbsolutePath(), lineFromFile);
+								LOG.error(fatal, "Prohibited String '{}' in '{}:{}'", aProhibited, logFile.getAbsolutePath(), lineFromFile);
 
 								StringBuilder logExcerpt = new StringBuilder();
 
@@ -456,7 +459,7 @@ public boolean accept(File dir, String name) {
 
 				}
 			} catch (FileNotFoundException e) {
-				LOG.warn("Unable to locate file: " + e.getMessage() + " file: " + f.getAbsolutePath());
+				LOG.warn("Unable to locate file: " + e.getMessage() + " file: " + logFile.getAbsolutePath());
 			}
 
 			return false;

flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBaseTest.java

@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.yarn;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.regex.Pattern;
+
+/**
+ * Tests for {@link YarnTestBase}.
+ */
+public class YarnTestBaseTest {
+
+	@Test
+	public void ensureWhitelistEntryMatches() {
+		ensureWhitelistEntryMatch("465 java.lang.InterruptedException: sleep interrupted");
+		ensureWhitelistEntryMatch("2020-09-19 22:06:19,458 WARN  akka.remote.ReliableDeliverySupervisor                       [] - Association with remote system [akka.tcp://flink@e466f3e261f3:42352] has failed, address is now gated for [50] ms. Reason: [Association failed with [akka.tcp://flink@e466f3e261f3:42352]] Caused by: [java.net.ConnectException: Connection refused: e466f3e261f3/192.168.224.2:42352]");
+	}
+
+	private void ensureWhitelistEntryMatch(String probe) {
+		for (Pattern pattern : YarnTestBase.WHITELISTED_STRINGS) {
+			if (pattern.matcher(probe).find()) {
+				return;
+			}
+		}
+		Assert.fail("The following string didn't match any whitelisted patterns '" + probe + "'");
+	}
+}