[FLINK-27206][metrics] Deprecate reflection-based reporter instantiation

docs/content.zh/docs/deployment/metric_reporters.md

@@ -36,7 +36,7 @@ Below is a list of parameters that are generally applicable to all reporters. Al
 
 {{< include_reporter_config "layouts/shortcodes/generated/metric_reporters_section.html" >}}
 
-All reporters must at least have either the `class` or `factory.class` property. Which property may/should be used depends on the reporter implementation. See the individual reporter configuration sections for more information.
+All reporter configurations must contain the `factory.class` property.
 Some reporters (referred to as `Scheduled`) allow specifying a reporting `interval`.
 
 Example reporter configuration that specifies multiple reporters:
@@ -54,10 +54,9 @@ metrics.reporter.my_other_reporter.host: 192.168.1.1
 metrics.reporter.my_other_reporter.port: 10000
 ```
 
-**Important:** The jar containing the reporter must be accessible when Flink is started. Reporters that support the
- `factory.class` property can be loaded as [plugins]({{< ref "docs/deployment/filesystems/plugins" >}}). Otherwise the jar must be placed
- in the /lib folder. Reporters that are shipped with Flink (i.e., all reporters documented on this page) are available
- by default.
+**Important:** The jar containing the reporter must be accessible when Flink is started.
+ Reporters are loaded as [plugins]({{< ref "docs/deployment/filesystems/plugins" >}}). 
+ All reporters documented on this page are available by default.
 
 You can write your own `Reporter` by implementing the `org.apache.flink.metrics.reporter.MetricReporter` interface.
 If the Reporter should send out reports regularly you have to implement the `Scheduled` interface as well.

docs/content/docs/deployment/metric_reporters.md

@@ -36,7 +36,7 @@ Below is a list of parameters that are generally applicable to all reporters. Al
 
 {{< include_reporter_config "layouts/shortcodes/generated/metric_reporters_section.html" >}}
 
-All reporters must at least have either the `class` or `factory.class` property. Which property may/should be used depends on the reporter implementation. See the individual reporter configuration sections for more information.
+All reporter configurations must contain the `factory.class` property.
 Some reporters (referred to as `Scheduled`) allow specifying a reporting `interval`.
 
 Example reporter configuration that specifies multiple reporters:
@@ -54,10 +54,9 @@ metrics.reporter.my_other_reporter.host: 192.168.1.1
 metrics.reporter.my_other_reporter.port: 10000
 ```
 
-**Important:** The jar containing the reporter must be accessible when Flink is started. Reporters that support the
- `factory.class` property can be loaded as [plugins]({{< ref "docs/deployment/filesystems/plugins" >}}). Otherwise the jar must be placed
- in the /lib folder. Reporters that are shipped with Flink (i.e., all reporters documented on this page) are available
- by default.
+**Important:** The jar containing the reporter must be accessible when Flink is started.
+ Reporters are loaded as [plugins]({{< ref "docs/deployment/filesystems/plugins" >}}).
+ All reporters documented on this page are available by default.
 
 You can write your own `Reporter` by implementing the `org.apache.flink.metrics.reporter.MetricReporter` interface.
 If the Reporter should send out reports regularly you have to implement the `Scheduled` interface as well.

docs/layouts/shortcodes/generated/metric_configuration.html

@@ -56,12 +56,6 @@
             <td>String</td>
             <td>Configures the parameter &lt;parameter&gt; for the reporter named &lt;name&gt;.</td>
         </tr>
-        <tr>
-            <td><h5>metrics.reporter.&lt;name&gt;.class</h5></td>
-            <td style="word-wrap: break-word;">(none)</td>
-            <td>String</td>
-            <td>The reporter class to use for the reporter named &lt;name&gt;.</td>
-        </tr>
         <tr>
             <td><h5>metrics.reporter.&lt;name&gt;.factory.class</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

docs/layouts/shortcodes/generated/metric_reporters_section.html

@@ -8,12 +8,6 @@
         </tr>
     </thead>
     <tbody>
-        <tr>
-            <td><h5>metrics.reporter.&lt;name&gt;.class</h5></td>
-            <td style="word-wrap: break-word;">(none)</td>
-            <td>String</td>
-            <td>The reporter class to use for the reporter named &lt;name&gt;.</td>
-        </tr>
         <tr>
             <td><h5>metrics.reporter.&lt;name&gt;.factory.class</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

flink-core/src/main/java/org/apache/flink/configuration/MetricOptions.java

@@ -67,8 +67,8 @@ public class MetricOptions {
                                     + " any of the names in the list will be started. Otherwise, all reporters that could be found in"
                                     + " the configuration will be started.");
 
-    @Documentation.SuffixOption(NAMED_REPORTER_CONFIG_PREFIX)
-    @Documentation.Section(value = Documentation.Sections.METRIC_REPORTERS, position = 1)
+    /** @deprecated use {@link MetricOptions#REPORTER_FACTORY_CLASS} instead. */
+    @Deprecated
     public static final ConfigOption<String> REPORTER_CLASS =
             key("class")
                     .stringType()

flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/reporter/InstantiateViaFactory.java

@@ -34,10 +34,13 @@
  *
  * <p>Attention: This annotation does not work if the reporter is loaded as a plugin. For these
  * cases, annotate the factory with {@link InterceptInstantiationViaReflection} instead.
+ *
+ * @deprecated Will be removed in a future version. Users should use all reporters as plugins.
  */
 @Retention(RetentionPolicy.RUNTIME)
 @Target(ElementType.TYPE)
 @Public
+@Deprecated
 public @interface InstantiateViaFactory {
     String factoryClassName();
 }

flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/reporter/InterceptInstantiationViaReflection.java

@@ -32,10 +32,12 @@
  * instead.
  *
  * @see InstantiateViaFactory
+ * @deprecated Will be removed in a future version. Users should use all reporters as plugins.
  */
 @Retention(RetentionPolicy.RUNTIME)
 @Target(ElementType.TYPE)
 @Public
+@Deprecated
 public @interface InterceptInstantiationViaReflection {
     String reporterClassName();
 }

flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/reporter/MetricReporter.java

@@ -26,11 +26,7 @@
 /**
  * Reporters are used to export {@link Metric Metrics} to an external backend.
  *
- * <p>Reporters are instantiated either a) via reflection, in which case they must be public,
- * non-abstract, and have a public no-argument constructor. b) via a {@link MetricReporterFactory},
- * in which case no restrictions apply. (recommended)
- *
- * <p>Reporters are neither required nor encouraged to support both instantiation paths.
+ * <p>Reporters are instantiated via a {@link MetricReporterFactory}.
  */
 @Public
 public interface MetricReporter {

flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/reporter/MetricReporterFactory.java

@@ -29,10 +29,6 @@
  * plugin, so long as the reporter jar is self-contained (excluding Flink dependencies) and contains
  * a {@code META-INF/services/org.apache.flink.metrics.reporter.MetricReporterFactory} file
  * containing the qualified class name of the factory.
- *
- * <p>Reporters that previously relied on reflection for instantiation can use the {@link
- * InstantiateViaFactory} annotation to redirect reflection-base instantiation attempts to the
- * factory instead.
  */
 @Public
 public interface MetricReporterFactory {

flink-runtime/src/main/java/org/apache/flink/runtime/metrics/ReporterSetup.java

@@ -71,6 +71,7 @@ public final class ReporterSetup {
 
     // regex pattern to extract the name from reporter configuration keys, e.g. "rep" from
     // "metrics.reporter.rep.class"
+    @SuppressWarnings("deprecation")
     private static final Pattern reporterClassPattern =
             Pattern.compile(
                     Pattern.quote(ConfigConstants.METRICS_REPORTER_PREFIX)
@@ -360,6 +361,7 @@ private static List<ReporterSetup> setupReporters(
         return reporterSetups;
     }
 
+    @SuppressWarnings("deprecation")
     private static Optional<MetricReporter> loadReporter(
             final String reporterName,
             final Configuration reporterConfig,
@@ -375,6 +377,15 @@ private static Optional<MetricReporter> loadReporter(
         }
 
         if (reporterClassName != null) {
+            LOG.warn(
+                    "The reporter configuration of '{}' configures the reporter class, which is a deprecated approach to configure reporters."
+                            + " Please configure a factory class instead: '{}{}.{}: <factoryClass>' to ensure that the configuration"
+                            + " continues to work with future versions.",
+                    reporterName,
+                    ConfigConstants.METRICS_REPORTER_PREFIX,
+                    reporterName,
+                    MetricOptions.REPORTER_FACTORY_CLASS.key());
+
             final Optional<MetricReporterFactory> interceptingFactory =
                     reporterFactories.values().stream()
                             .filter(
@@ -434,6 +445,7 @@ private static Optional<MetricReporter> loadViaFactory(
         return Optional.of(factory.createMetricReporter(metricConfig));
     }
 
+    @SuppressWarnings("deprecation")
     private static Optional<MetricReporter> loadViaReflection(
             final String reporterClassName,
             final String reporterName,
@@ -448,15 +460,6 @@ private static Optional<MetricReporter> loadViaReflection(
         if (alternativeFactoryAnnotation != null) {
             final String alternativeFactoryClassName =
                     alternativeFactoryAnnotation.factoryClassName();
-            LOG.info(
-                    "The reporter configuration of {} is out-dated (but still supported)."
-                            + " Please configure a factory class instead: '{}{}.{}: {}' to ensure that the configuration"
-                            + " continues to work with future versions.",
-                    reporterName,
-                    MetricOptions.REPORTER_CLASS.key(),
-                    reporterName,
-                    MetricOptions.REPORTER_FACTORY_CLASS.key(),
-                    alternativeFactoryClassName);
             return loadViaFactory(
                     alternativeFactoryClassName, reporterName, reporterConfig, reporterFactories);
         }

flink-runtime/src/test/java/org/apache/flink/runtime/metrics/ReporterSetupTest.java

@@ -306,7 +306,8 @@ void testFactoryParsing() throws Exception {
      * are configured.
      */
     @Test
-    void testFactoryPrioritization() throws Exception {
+    @SuppressWarnings("deprecation")
+    public void testFactoryPrioritization() throws Exception {
         final Configuration config = new Configuration();
         config.setString(
                 ConfigConstants.METRICS_REPORTER_PREFIX
@@ -352,7 +353,8 @@ void testFactoryFailureIsolation() throws Exception {
 
     /** Verifies that factory/reflection approaches can be mixed freely. */
     @Test
-    void testMixedSetupsFactoryParsing() throws Exception {
+    @SuppressWarnings("deprecation")
+    public void testMixedSetupsFactoryParsing() throws Exception {
         final Configuration config = new Configuration();
         config.setString(
                 ConfigConstants.METRICS_REPORTER_PREFIX
@@ -401,7 +403,8 @@ void testFactoryArgumentForwarding() throws Exception {
      * InstantiateViaFactory}.
      */
     @Test
-    void testFactoryAnnotation() {
+    @SuppressWarnings("deprecation")
+    public void testFactoryAnnotation() {
         final Configuration config = new Configuration();
         config.setString(
                 ConfigConstants.METRICS_REPORTER_PREFIX
@@ -425,7 +428,8 @@ void testFactoryAnnotation() {
      * org.apache.flink.metrics.reporter.InterceptInstantiationViaReflection}.
      */
     @Test
-    void testReflectionInterception() {
+    @SuppressWarnings("deprecation")
+    public void testReflectionInterception() {
         final Configuration config = new Configuration();
         config.setString(
                 ConfigConstants.METRICS_REPORTER_PREFIX
@@ -534,6 +538,7 @@ public MetricReporter createMetricReporter(Properties config) {
     @InterceptInstantiationViaReflection(
             reporterClassName =
                     "org.apache.flink.runtime.metrics.ReporterSetupTest$InstantiationTypeTrackingTestReporter")
+    @SuppressWarnings("deprecation")
     public static class InterceptingInstantiationTypeTrackingTestReporterFactory
             implements MetricReporterFactory {
 
@@ -565,6 +570,7 @@ public boolean isCreatedByFactory() {
     @InstantiateViaFactory(
             factoryClassName =
                     "org.apache.flink.runtime.metrics.ReporterSetupTest$InstantiationTypeTrackingTestReporterFactory")
+    @SuppressWarnings("deprecation")
     protected static class InstantiationTypeTrackingTestReporter2
             extends InstantiationTypeTrackingTestReporter {}
 }