[FLINK-33022][runtime] Log an error when enrichers defined as part of the configuration can not be found/loaded
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/failure/FailureEnricherUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/failure/FailureEnricherUtils.java
index d9b4c22..0e19573 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/failure/FailureEnricherUtils.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/failure/FailureEnricherUtils.java
@@ -75,9 +75,9 @@
@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 @@
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 @@
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);
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/failure/FailureEnricherUtilsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/failure/FailureEnricherUtilsTest.java
index d5b6ef3..7d1cb80 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/failure/FailureEnricherUtilsTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/failure/FailureEnricherUtilsTest.java
@@ -108,7 +108,20 @@
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
