zentol commented on code in PR #22468: URL: https://github.com/apache/flink/pull/22468#discussion_r1318386858
########## docs/content/docs/deployment/advanced/failure_enrichers.md: ########## @@ -0,0 +1,112 @@ +--- +title: "Failure Enrichers" +nav-title: failure-enrichers +nav-parent_id: advanced +nav-pos: 3 +--- +<!-- +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. +--> + +## 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/en/java/javase/17/docs/api/java.base/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. Review Comment: > otherwise may be ignored Otherwise _what_ may be ignored? ########## docs/content/docs/deployment/advanced/failure_enrichers.md: ########## @@ -0,0 +1,112 @@ +--- +title: "Failure Enrichers" +nav-title: failure-enrichers +nav-parent_id: advanced +nav-pos: 3 +--- +<!-- +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. +--> + +## 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). Review Comment: ```suggestion Flink provides a pluggable interface for users to register their custom logic and enrich failures with extra metadata labels (string key-value pairs). ``` ########## docs/content/docs/deployment/advanced/failure_enrichers.md: ########## @@ -0,0 +1,112 @@ +--- +title: "Failure Enrichers" +nav-title: failure-enrichers +nav-parent_id: advanced +nav-pos: 3 +--- +<!-- +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. +--> + +## 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/en/java/javase/17/docs/api/java.base/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<String> outputKeys; + + public CustomEnricher() { + this.outputKeys = Collections.singleton("labelKey"); + } + + @Override + public Set<String> getOutputKeys() { + return outputKeys; + } + + @Override + public CompletableFuture<Map<String, String>> processFailure( + Throwable cause, Context context) { + return CompletableFuture.completedFuture(Collections.singletonMap("labelKey", "labelValue")); + } +} +``` + +### Configuration + +JobManager loads FailureEnricher plugins at startup. To make sure your FailureEnrichers are loaded: +* All class names should be defined as part of {{< javadoc file="org/apache/flink/configuration/JobManagerOptions.html#FAILURE_ENRICHERS_LIST" name="jobmanager.failure-enrichers">}} configuration. Review Comment: Please link to the config docs instead of the javadocs. ########## docs/content/docs/deployment/advanced/failure_enrichers.md: ########## @@ -0,0 +1,112 @@ +--- +title: "Failure Enrichers" +nav-title: failure-enrichers +nav-parent_id: advanced +nav-pos: 3 +--- +<!-- +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. +--> + +## 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/en/java/javase/17/docs/api/java.base/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<String> outputKeys; + + public CustomEnricher() { + this.outputKeys = Collections.singleton("labelKey"); + } + + @Override + public Set<String> getOutputKeys() { + return outputKeys; + } + + @Override + public CompletableFuture<Map<String, String>> processFailure( + Throwable cause, Context context) { + return CompletableFuture.completedFuture(Collections.singletonMap("labelKey", "labelValue")); + } +} +``` + +### Configuration + +JobManager loads FailureEnricher plugins at startup. To make sure your FailureEnrichers are loaded: Review Comment: ```suggestion The JobManager loads FailureEnricher plugins at startup. To make sure your FailureEnrichers are loaded: ``` ########## docs/content/docs/deployment/advanced/failure_enrichers.md: ########## @@ -0,0 +1,112 @@ +--- +title: "Failure Enrichers" +nav-title: failure-enrichers +nav-parent_id: advanced +nav-pos: 3 +--- +<!-- +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. +--> + +## 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/en/java/javase/17/docs/api/java.base/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<String> outputKeys; + + public CustomEnricher() { + this.outputKeys = Collections.singleton("labelKey"); + } + + @Override + public Set<String> getOutputKeys() { + return outputKeys; + } + + @Override + public CompletableFuture<Map<String, String>> processFailure( + Throwable cause, Context context) { + return CompletableFuture.completedFuture(Collections.singletonMap("labelKey", "labelValue")); + } +} +``` + +### Configuration + +JobManager loads FailureEnricher plugins at startup. To make sure your FailureEnrichers are loaded: +* All class names should be defined as part of {{< javadoc file="org/apache/flink/configuration/JobManagerOptions.html#FAILURE_ENRICHERS_LIST" name="jobmanager.failure-enrichers">}} configuration. + If this configuration is empty, NO enrichers will be started. Example: +``` + jobmanager.failure-enrichers = org.apache.flink.test.plugin.jar.failure.CustomEnricher Review Comment: This would be more complete if the CustomEnricher example included a `package` ########## docs/content/docs/deployment/advanced/failure_enrichers.md: ########## @@ -0,0 +1,112 @@ +--- +title: "Failure Enrichers" +nav-title: failure-enrichers +nav-parent_id: advanced +nav-pos: 3 +--- +<!-- +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. +--> + +## 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. Review Comment: ```suggestion This enables users to implement their own failure enrichment plugins to categorize job failures, expose custom metrics, or make calls to external notification systems. ``` Consistent target audience, clarify that you _can_ do that stuff and that it's not just a "goal" (which _may_ not have been reached). I'd also remove "and more" because it doesn't add any value. You don't want users guessing what this can be used for. ########## docs/content/docs/deployment/advanced/failure_enrichers.md: ########## @@ -0,0 +1,112 @@ +--- +title: "Failure Enrichers" +nav-title: failure-enrichers +nav-parent_id: advanced +nav-pos: 3 +--- +<!-- +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. +--> + +## 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). Review Comment: ```suggestion Every FailureEnricher may asynchronously return labels associated with the failure that are then exposed via the JobManager's REST API (e.g., a 'type:System' label implying the failure is categorized as a system error). ``` ########## docs/content/docs/deployment/advanced/failure_enrichers.md: ########## @@ -0,0 +1,112 @@ +--- +title: "Failure Enrichers" +nav-title: failure-enrichers +nav-parent_id: advanced +nav-pos: 3 +--- +<!-- +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. +--> + +## 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/en/java/javase/17/docs/api/java.base/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. Review Comment: ```suggestion Then, create a jar which includes your `FailureEnricher`, `FailureEnricherFactory`, `META-INF/services/` and all external dependencies. ``` ########## docs/content/docs/deployment/advanced/failure_enrichers.md: ########## @@ -0,0 +1,112 @@ +--- +title: "Failure Enrichers" +nav-title: failure-enrichers +nav-parent_id: advanced +nav-pos: 3 +--- +<!-- +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. +--> + +## 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/en/java/javase/17/docs/api/java.base/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<String> outputKeys; + + public CustomEnricher() { + this.outputKeys = Collections.singleton("labelKey"); + } + + @Override + public Set<String> getOutputKeys() { + return outputKeys; + } + + @Override + public CompletableFuture<Map<String, String>> processFailure( + Throwable cause, Context context) { + return CompletableFuture.completedFuture(Collections.singletonMap("labelKey", "labelValue")); + } +} +``` + +### Configuration + +JobManager loads FailureEnricher plugins at startup. To make sure your FailureEnrichers are loaded: +* All class names should be defined as part of {{< javadoc file="org/apache/flink/configuration/JobManagerOptions.html#FAILURE_ENRICHERS_LIST" name="jobmanager.failure-enrichers">}} configuration. + If this configuration is empty, NO enrichers will be started. Example: +``` + jobmanager.failure-enrichers = org.apache.flink.test.plugin.jar.failure.CustomEnricher +``` + +### Validation + +To validate that your FailureEnricher is loaded, you can check the JobManager logs for the following line: +``` + Found failure enricher org.apache.flink.test.plugin.jar.failure.CustomEnricher at jar:file:/path/to/flink/plugins/failure-enrichment/flink-test-plugin.jar!/org/apache/flink/test/plugin/jar/failure/CustomEnricher.class +``` + +Moreover, you can query the JobManager's Rest API using the following endpoint: +``` + http://localhost:8081/jobs/<jobid>/exceptions Review Comment: You could link to the REST API docs instead. ########## docs/content/docs/deployment/advanced/failure_enrichers.md: ########## @@ -0,0 +1,112 @@ +--- +title: "Failure Enrichers" +nav-title: failure-enrichers +nav-parent_id: advanced +nav-pos: 3 +--- +<!-- +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. +--> + +## 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/en/java/javase/17/docs/api/java.base/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<String> outputKeys; + + public CustomEnricher() { + this.outputKeys = Collections.singleton("labelKey"); + } + + @Override + public Set<String> getOutputKeys() { + return outputKeys; + } + + @Override + public CompletableFuture<Map<String, String>> processFailure( + Throwable cause, Context context) { + return CompletableFuture.completedFuture(Collections.singletonMap("labelKey", "labelValue")); + } +} +``` + +### Configuration + +JobManager loads FailureEnricher plugins at startup. To make sure your FailureEnrichers are loaded: +* All class names should be defined as part of {{< javadoc file="org/apache/flink/configuration/JobManagerOptions.html#FAILURE_ENRICHERS_LIST" name="jobmanager.failure-enrichers">}} configuration. Review Comment: Feels weird to have a list containing a single item. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
