[FLINK-33022][runtime] Log an error when enrichers defined as part of the configuration can not be found/loaded

Author: pgaref

flink-runtime/src/main/java/org/apache/flink/runtime/failure/FailureEnricherUtils.java

@@ -75,9 +75,9 @@ public static Collection<FailureEnricher> getFailureEnrichers(
     @VisibleForTesting
     static Collection<FailureEnricher> getFailureEnrichers(
             final Configuration configuration, final PluginManager pluginManager) {
-        Set<String> includedEnrichers = getIncludedFailureEnrichers(configuration);
+        final Set<String> enrichersToLoad = getIncludedFailureEnrichers(configuration);
         //  When empty, NO enrichers will be started.
-        if (includedEnrichers.isEmpty()) {
+        if (enrichersToLoad.isEmpty()) {
             return Collections.emptySet();
         }
         final Iterator<FailureEnricherFactory> factoryIterator =
@@ -87,7 +87,7 @@ static Collection<FailureEnricher> getFailureEnrichers(
             final FailureEnricherFactory failureEnricherFactory = factoryIterator.next();
             final FailureEnricher failureEnricher =
                     failureEnricherFactory.createFailureEnricher(configuration);
-            if (includedEnrichers.contains(failureEnricher.getClass().getName())) {
+            if (enrichersToLoad.remove(failureEnricher.getClass().getName())) {
                 failureEnrichers.add(failureEnricher);
                 LOG.info(
                         "Found failure enricher {} at {}.",
@@ -102,10 +102,16 @@ static Collection<FailureEnricher> getFailureEnrichers(
                 LOG.debug(
                         "Excluding failure enricher {}, not configured in enricher list ({}).",
                         failureEnricherFactory.getClass().getName(),
-                        includedEnrichers);
+                        enrichersToLoad);
             }
         }
 
+        if (!enrichersToLoad.isEmpty()) {
+            LOG.error(
+                    "The following failure enrichers were configured but not found on the classpath: {}.",
+                    enrichersToLoad);
+        }
+
         return filterInvalidEnrichers(failureEnrichers);
     }
 

flink-runtime/src/test/java/org/apache/flink/runtime/failure/FailureEnricherUtilsTest.java

@@ -108,7 +108,20 @@ public void testGetFailureEnrichers() {
                 FailureEnricherUtils.getFailureEnrichers(configuration, createPluginManager());
         assertThat(enrichers).hasSize(1);
         // verify that the failure enricher was created and returned
-        assertThat(enrichers.iterator().next()).isInstanceOf(TestEnricher.class);
+        assertThat(enrichers)
+                .satisfiesExactly(
+                        enricher -> assertThat(enricher).isInstanceOf(TestEnricher.class));
+
+        // Valid plus Invalid Name combination
+        configuration.set(
+                JobManagerOptions.FAILURE_ENRICHERS_LIST,
+                FailureEnricherUtilsTest.class.getName() + "," + TestEnricher.class.getName());
+        final Collection<FailureEnricher> validInvalidEnrichers =
+                FailureEnricherUtils.getFailureEnrichers(configuration, createPluginManager());
+        assertThat(validInvalidEnrichers).hasSize(1);
+        assertThat(validInvalidEnrichers)
+                .satisfiesExactly(
+                        enricher -> assertThat(enricher).isInstanceOf(TestEnricher.class));
     }
 
     @Test