updating name for evaluator plugin to match naming convention for plugins.

Author: pchoudhury22

docs/content.zh/docs/operations/plugins.md

@@ -188,3 +188,5 @@ The following steps demonstrate how to develop and use a custom mutator.
     ```text
     2023-12-12 06:26:56,667 o.a.f.k.o.u.MutatorUtils [INFO ] Discovered mutator from plugin directory[/opt/flink/plugins]: org.apache.flink.mutator.CustomFlinkMutator.
     ```
+
+## Flink Autoscaler Custom Evaluator

flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobAutoScalerImpl.java

@@ -24,8 +24,8 @@
 import org.apache.flink.autoscaler.event.AutoScalerEventHandler;
 import org.apache.flink.autoscaler.exceptions.NotReadyException;
 import org.apache.flink.autoscaler.metrics.AutoscalerFlinkMetrics;
-import org.apache.flink.autoscaler.metrics.CustomEvaluator;
 import org.apache.flink.autoscaler.metrics.EvaluatedMetrics;
+import org.apache.flink.autoscaler.metrics.FlinkAutoscalerEvaluator;
 import org.apache.flink.autoscaler.realizer.ScalingRealizer;
 import org.apache.flink.autoscaler.state.AutoScalerStateStore;
 import org.apache.flink.autoscaler.tuning.ConfigChanges;
@@ -63,7 +63,7 @@ public class JobAutoScalerImpl<KEY, Context extends JobAutoScalerContext<KEY>>
     private final AutoScalerEventHandler<KEY, Context> eventHandler;
     private final ScalingRealizer<KEY, Context> scalingRealizer;
     private final AutoScalerStateStore<KEY, Context> stateStore;
-    private final Map<String, CustomEvaluator> customEvaluators;
+    private final Map<String, FlinkAutoscalerEvaluator> customEvaluators;
 
     private Clock clock = Clock.systemDefaultZone();
 
@@ -80,7 +80,7 @@ public JobAutoScalerImpl(
             AutoScalerEventHandler<KEY, Context> eventHandler,
             ScalingRealizer<KEY, Context> scalingRealizer,
             AutoScalerStateStore<KEY, Context> stateStore,
-            Map<String, CustomEvaluator> customEvaluators) {
+            Map<String, FlinkAutoscalerEvaluator> customEvaluators) {
         this.metricsCollector = metricsCollector;
         this.evaluator = evaluator;
         this.scalingExecutor = scalingExecutor;
@@ -276,12 +276,12 @@ void setClock(Clock clock) {
     }
 
     @VisibleForTesting
-    protected Tuple2<CustomEvaluator, Configuration> getCustomEvaluatorIfRequired(
+    protected Tuple2<FlinkAutoscalerEvaluator, Configuration> getCustomEvaluatorIfRequired(
             Configuration conf) {
         return Optional.ofNullable(conf.get(CUSTOM_EVALUATOR_NAME))
                 .map(
                         name -> {
-                            CustomEvaluator evaluator = customEvaluators.get(name);
+                            FlinkAutoscalerEvaluator evaluator = customEvaluators.get(name);
                             return evaluator != null
                                     ? new Tuple2<>(
                                             evaluator,

flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java

@@ -22,9 +22,9 @@
 import org.apache.flink.autoscaler.config.AutoScalerOptions;
 import org.apache.flink.autoscaler.metrics.CollectedMetricHistory;
 import org.apache.flink.autoscaler.metrics.CollectedMetrics;
-import org.apache.flink.autoscaler.metrics.CustomEvaluator;
 import org.apache.flink.autoscaler.metrics.EvaluatedMetrics;
 import org.apache.flink.autoscaler.metrics.EvaluatedScalingMetric;
+import org.apache.flink.autoscaler.metrics.FlinkAutoscalerEvaluator;
 import org.apache.flink.autoscaler.metrics.MetricAggregator;
 import org.apache.flink.autoscaler.metrics.ScalingMetric;
 import org.apache.flink.autoscaler.topology.JobTopology;
@@ -79,7 +79,9 @@ public EvaluatedMetrics evaluate(
             Configuration conf,
             CollectedMetricHistory collectedMetrics,
             Duration restartTime,
-            @Nullable Tuple2<CustomEvaluator, Configuration> customEvaluatorWithConfig) {
+            @Nullable
+                    Tuple2<FlinkAutoscalerEvaluator, Configuration>
+                            customEvaluatorWithConfig) {
         LOG.debug("Restart time used in metrics evaluation: {}", restartTime);
         var scalingOutput = new HashMap<JobVertexID, Map<ScalingMetric, EvaluatedScalingMetric>>();
         var metricsHistory = collectedMetrics.getMetricHistory();
@@ -93,7 +95,7 @@ public EvaluatedMetrics evaluate(
                                 info ->
                                         Tuple2.of(
                                                 info.f0,
-                                                new CustomEvaluator.Context(
+                                                new FlinkAutoscalerEvaluator.Context(
                                                         new UnmodifiableConfiguration(conf),
                                                         Collections.unmodifiableSortedMap(
                                                                 metricsHistory),
@@ -158,7 +160,9 @@ private Map<ScalingMetric, EvaluatedScalingMetric> evaluateMetrics(
             JobVertexID vertex,
             boolean processingBacklog,
             Duration restartTime,
-            @Nullable Tuple2<CustomEvaluator, CustomEvaluator.Context> customEvaluationSession) {
+            @Nullable
+                    Tuple2<FlinkAutoscalerEvaluator, FlinkAutoscalerEvaluator.Context>
+                            customEvaluationSession) {
 
         var latestVertexMetrics =
                 metricsHistory.get(metricsHistory.lastKey()).getVertexMetrics().get(vertex);
@@ -633,7 +637,7 @@ protected static double computeEdgeDataRate(
 
     /**
      * Executes the provided custom evaluator for the given job vertex. Calls {@link
-     * CustomEvaluator#evaluateVertexMetrics} to evaluate scaling metrics.
+     * FlinkAutoscalerEvaluator#evaluateVertexMetrics} to evaluate scaling metrics.
      *
      * @param vertex The job vertex being evaluated.
      * @param evaluatedMetrics Current evaluated metrics.
@@ -645,7 +649,8 @@ protected static double computeEdgeDataRate(
     protected static Map<ScalingMetric, EvaluatedScalingMetric> runCustomEvaluator(
             JobVertexID vertex,
             Map<ScalingMetric, EvaluatedScalingMetric> evaluatedMetrics,
-            Tuple2<CustomEvaluator, CustomEvaluator.Context> customEvaluationSession) {
+            Tuple2<FlinkAutoscalerEvaluator, FlinkAutoscalerEvaluator.Context>
+                    customEvaluationSession) {
         try {
             return customEvaluationSession.f0.evaluateVertexMetrics(
                     vertex, evaluatedMetrics, customEvaluationSession.f1);

flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/CustomEvaluator.java renamed to flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/FlinkAutoscalerEvaluator.java

@@ -31,11 +31,11 @@
 import java.util.SortedMap;
 
 /**
- * Interface for custom evaluators that allow custom scaling metric evaluations. Implementations of
+ * Interface for custom evaluators that allow tailored scaling metric evaluations. Implementations of
  * this interface can provide custom logic to evaluate vertex metrics and merge them with internally
  * evaluated metrics.
  */
-public interface CustomEvaluator extends Plugin {
+public interface FlinkAutoscalerEvaluator extends Plugin {
 
     /**
      * Returns the name of the custom evaluator.

flink-autoscaler/src/test/java/org/apache/flink/autoscaler/AutoScalerCustomEvaluatorTest.java

@@ -22,7 +22,7 @@
 import org.apache.flink.autoscaler.config.AutoScalerOptions;
 import org.apache.flink.autoscaler.event.TestingEventCollector;
 import org.apache.flink.autoscaler.metrics.AutoscalerFlinkMetrics;
-import org.apache.flink.autoscaler.metrics.CustomEvaluator;
+import org.apache.flink.autoscaler.metrics.FlinkAutoscalerEvaluator;
 import org.apache.flink.autoscaler.metrics.TestCustomEvaluator;
 import org.apache.flink.autoscaler.metrics.TestMetrics;
 import org.apache.flink.autoscaler.realizer.TestingScalingRealizer;
@@ -165,7 +165,7 @@ private void assertFlinkMetricsCount(int scalingCount, int balancedCount) {
         assertEquals(balancedCount, autoscalerFlinkMetrics.getNumBalancedCount());
     }
 
-    private Map<String, CustomEvaluator> createTestCustomEvaluator() {
+    private Map<String, FlinkAutoscalerEvaluator> createTestCustomEvaluator() {
         var testCustomEvaluator = new TestCustomEvaluator();
         testCustomEvaluator.configure(new Configuration());
         return Map.of(testCustomEvaluator.getName(), testCustomEvaluator);

flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobAutoScalerImplTest.java

@@ -23,7 +23,7 @@
 import org.apache.flink.autoscaler.exceptions.NotReadyException;
 import org.apache.flink.autoscaler.metrics.AutoscalerFlinkMetrics;
 import org.apache.flink.autoscaler.metrics.CollectedMetrics;
-import org.apache.flink.autoscaler.metrics.CustomEvaluator;
+import org.apache.flink.autoscaler.metrics.FlinkAutoscalerEvaluator;
 import org.apache.flink.autoscaler.metrics.ScalingMetric;
 import org.apache.flink.autoscaler.metrics.TestCustomEvaluator;
 import org.apache.flink.autoscaler.metrics.TestMetrics;
@@ -466,7 +466,7 @@ private TestingScalingRealizer.Event<JobID, JobAutoScalerContext<JobID>> getEven
 
     @Test
     void testGetCustomEvaluatorIfRequired() {
-        CustomEvaluator testCustomEvaluator = new TestCustomEvaluator();
+        FlinkAutoscalerEvaluator testCustomEvaluator = new TestCustomEvaluator();
         testCustomEvaluator.configure(new Configuration());
         var testCustomEvaluators = Map.of(testCustomEvaluator.getName(), testCustomEvaluator);
 
@@ -491,7 +491,7 @@ void testGetCustomEvaluatorIfRequired() {
                 autoscalerWithCustomEvaluator.getCustomEvaluatorIfRequired(
                         context.getConfiguration());
         assertNotNull(customEvaluatorWithConfig);
-        assertInstanceOf(CustomEvaluator.class, customEvaluatorWithConfig.f0);
+        assertInstanceOf(FlinkAutoscalerEvaluator.class, customEvaluatorWithConfig.f0);
         var customEvaluatorConfig = customEvaluatorWithConfig.f1;
         assertNotNull(customEvaluatorConfig);
         assertEquals(0, customEvaluatorConfig.keySet().size());
@@ -525,7 +525,8 @@ void testGetCustomEvaluatorIfRequired() {
                         context.getConfiguration());
         assertNotNull(customEvaluatorWithConfigContainingAdditionalKeys);
         assertInstanceOf(
-                CustomEvaluator.class, customEvaluatorWithConfigContainingAdditionalKeys.f0);
+                FlinkAutoscalerEvaluator.class,
+                customEvaluatorWithConfigContainingAdditionalKeys.f0);
         var customEvaluatorConfigContainingAdditionalKeys =
                 customEvaluatorWithConfigContainingAdditionalKeys.f1;
         assertNotNull(customEvaluatorConfigContainingAdditionalKeys);

flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricEvaluatorTest.java

@@ -21,8 +21,8 @@
 import org.apache.flink.autoscaler.config.AutoScalerOptions;
 import org.apache.flink.autoscaler.metrics.CollectedMetricHistory;
 import org.apache.flink.autoscaler.metrics.CollectedMetrics;
-import org.apache.flink.autoscaler.metrics.CustomEvaluator;
 import org.apache.flink.autoscaler.metrics.EvaluatedScalingMetric;
+import org.apache.flink.autoscaler.metrics.FlinkAutoscalerEvaluator;
 import org.apache.flink.autoscaler.metrics.MetricAggregator;
 import org.apache.flink.autoscaler.metrics.ScalingMetric;
 import org.apache.flink.autoscaler.metrics.TestCustomEvaluator;
@@ -319,7 +319,7 @@ public void testEvaluateWithCustomEvaluator() {
         var conf = new Configuration();
 
         conf.set(CATCH_UP_DURATION, Duration.ofSeconds(2));
-        CustomEvaluator customEvaluator = new TestCustomEvaluator();
+        FlinkAutoscalerEvaluator customEvaluator = new TestCustomEvaluator();
         var customEvaluatorWithConfig = new Tuple2<>(customEvaluator, new Configuration());
 
         var evaluatedMetrics =
@@ -985,13 +985,13 @@ public void testRunCustomEvaluator() {
                         Map.of()));
 
         var conf = new Configuration();
-        CustomEvaluator customEvaluator = new TestCustomEvaluator();
+        FlinkAutoscalerEvaluator customEvaluator = new TestCustomEvaluator();
         var evaluatedMetrics = new HashMap<ScalingMetric, EvaluatedScalingMetric>();
 
         var testCustomEvaluationSession =
                 Tuple2.of(
                         customEvaluator,
-                        new CustomEvaluator.Context(
+                        new FlinkAutoscalerEvaluator.Context(
                                 new UnmodifiableConfiguration(conf),
                                 Collections.unmodifiableSortedMap(metricHistory),
                                 Collections.unmodifiableMap(new HashMap<>()),

flink-autoscaler/src/test/java/org/apache/flink/autoscaler/metrics/SimpleTrendAdjustor.java

@@ -25,10 +25,10 @@
 import java.util.Map;
 
 /**
- * A test implementation of the {@link CustomEvaluator} interface that provides custom scaling
- * metric evaluations for job vertices in a Flink job.
+ * A test implementation of the {@link FlinkAutoscalerEvaluator} interface that provides
+ * custom scaling metric evaluations for job vertices in a Flink job.
  */
-public class TestCustomEvaluator implements CustomEvaluator {
+public class TestCustomEvaluator implements FlinkAutoscalerEvaluator {
     @Override
     public String getName() {
         return "test-custom-evaluator";

flink-autoscaler/src/test/java/org/apache/flink/autoscaler/metrics/TestCustomEvaluator.java

@@ -18,7 +18,7 @@
 
 package org.apache.flink.kubernetes.operator.autoscaler;
 
-import org.apache.flink.autoscaler.metrics.CustomEvaluator;
+import org.apache.flink.autoscaler.metrics.FlinkAutoscalerEvaluator;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.core.plugin.PluginUtils;
 import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
@@ -38,14 +38,15 @@ public class AutoscalerUtils {
      *
      * @param configManager Flink Config manager
      * @return A map of discovered custom evaluators, where the key is the evaluator name provided
-     *     by {@link CustomEvaluator#getName()}) and the value is the corresponding instance.
+     *     by {@link FlinkAutoscalerEvaluator#getName()}) and the value is the corresponding
+     *     instance.
      */
-    public static Map<String, CustomEvaluator> discoverCustomEvaluators(
+    public static Map<String, FlinkAutoscalerEvaluator> discoverCustomEvaluators(
             FlinkConfigManager configManager) {
         var conf = configManager.getDefaultConfig();
-        Map<String, CustomEvaluator> customEvaluators = new HashMap<>();
+        Map<String, FlinkAutoscalerEvaluator> customEvaluators = new HashMap<>();
         PluginUtils.createPluginManagerFromRootFolder(conf)
-                .load(CustomEvaluator.class)
+                .load(FlinkAutoscalerEvaluator.class)
                 .forEachRemaining(
                         customEvaluator -> {
                             String customEvaluatorName = customEvaluator.getName();