wangzzu commented on code in PR #22468:
URL: https://github.com/apache/flink/pull/22468#discussion_r1314383990
##
docs/content/docs/deployment/advanced/failure_enrichers.md:
##
@@ -0,0 +1,93 @@
+---
+title: "Failure Enrichers"
+nav-title: failure-enrichers
+nav-parent_id: advanced
+nav-pos: 3
+---
+
+
+## Custom failure enrichers
+Flink provides a pluggable interface for users to register their custom logic
and enrich failures with extra metadata labels (string kv pairs).
+The goal is to enable developers implement their own failure enrichment
plugins to categorize job failures, expose custom metrics, make calls to
external notification systems, and more.
+
+FailureEnrichers are triggered every time an exception is reported at runtime
by the JobManager.
+Every FailureEnricher may asynchronously return labels associated with the
failure that are then exposed via the JobManager's Rest interface (e.g., a
'type:System' label implying the failure is categorized as a system error).
+
+
+### Implement a plugin for your custom enricher
+
+To implement a custom FailureEnricher plugin, you need to:
+
+- Add your own FailureEnricher by implementing the {{< gh_link
file="/flink-core/src/main/java/org/apache/flink/core/failure/FailureEnricher.java"
name="FailureEnricher" >}} interface.
+
+- Add your own FailureEnricherFactory by implementing the {{< gh_link
file="/flink-core/src/main/java/org/apache/flink/core/failure/FailureEnricherFactory.java"
name="FailureEnricherFactory" >}} interface.
+
+- Add a service entry. Create a file
`META-INF/services/org.apache.flink.core.failure.FailureEnricherFactory` which
contains the class name of your failure enricher factory class (see [Java
Service
Loader](https://docs.oracle.com/javase/8/docs/api/java/util/ServiceLoader.html)
docs for more details).
Review Comment:
maybe use
`https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/util/ServiceLoader.html`
is better, now flink has spported jdk17
##
docs/content/docs/deployment/advanced/failure_enrichers.md:
##
@@ -0,0 +1,93 @@
+---
+title: "Failure Enrichers"
+nav-title: failure-enrichers
+nav-parent_id: advanced
+nav-pos: 3
+---
+
+
+## Custom failure enrichers
+Flink provides a pluggable interface for users to register their custom logic
and enrich failures with extra metadata labels (string kv pairs).
+The goal is to enable developers implement their own failure enrichment
plugins to categorize job failures, expose custom metrics, make calls to
external notification systems, and more.
+
+FailureEnrichers are triggered every time an exception is reported at runtime
by the JobManager.
+Every FailureEnricher may asynchronously return labels associated with the
failure that are then exposed via the JobManager's Rest interface (e.g., a
'type:System' label implying the failure is categorized as a system error).
+
+
+### Implement a plugin for your custom enricher
+
+To implement a custom FailureEnricher plugin, you need to:
+
+- Add your own FailureEnricher by implementing the {{< gh_link
file="/flink-core/src/main/java/org/apache/flink/core/failure/FailureEnricher.java"
name="FailureEnricher" >}} interface.
+
+- Add your own FailureEnricherFactory by implementing the {{< gh_link
file="/flink-core/src/main/java/org/apache/flink/core/failure/FailureEnricherFactory.java"
name="FailureEnricherFactory" >}} interface.
+
+- Add a service entry. Create a file
`META-INF/services/org.apache.flink.core.failure.FailureEnricherFactory` which
contains the class name of your failure enricher factory class (see [Java
Service
Loader](https://docs.oracle.com/javase/8/docs/api/java/util/ServiceLoader.html)
docs for more details).
+
+
+Then, create a jar which includes your `FailureEnricher`,
`FailureEnricherFactory`, `META-INF/services/` and all the external
dependencies.
+Make a directory in `plugins/` of your Flink distribution with an arbitrary
name, e.g. "failure-enrichment", and put the jar into this directory.
+See [Flink Plugin]({% link deployment/filesystems/plugins.md %}) for more
details.
+
+{{< hint warning >}}
+Note that every FailureEnricher should have defined a set of {{< gh_link
file="/flink-core/src/main/java/org/apache/flink/core/failure/FailureEnricher.java"
name="output keys" >}} that may be associated with values. This set of keys
has to be unique across enrichers otherwise may be ignored.
+{{< /hint >}}
+
+FailureEnricherFactory example:
+
+``` java
+public class TestFailureEnricherFactory implements FailureEnricherFactory {
+
+ @Override
+ public FailureEnricher createFailureEnricher(Configuration conf) {
+return new CustomEnricher();
+ }
+}
+```
+
+FailureEnricher example:
+
+``` java
+public class CustomEnricher implements FailureEnricher {
+private final Set outputKeys;
+
+public CustomEnricher() {
+this.outputKeys = Collections.singleton("labelKey");
+}
+
+@Override
+public Set
