diff --git a/docs/layouts/shortcodes/generated/auto_scaler_configuration.html b/docs/layouts/shortcodes/generated/auto_scaler_configuration.html
index 0221b56741..0b5f33a8e2 100644
--- a/docs/layouts/shortcodes/generated/auto_scaler_configuration.html
+++ b/docs/layouts/shortcodes/generated/auto_scaler_configuration.html
@@ -117,7 +117,7 @@
| Enable vertex scaling execution by the autoscaler. If disabled, the autoscaler will only collect metrics and evaluate the suggested parallelism for each vertex but will not upgrade the jobs. |
- job.autoscaler.scaling.report.interval |
+ job.autoscaler.scaling.event.interval |
30 min |
Duration |
Time interval to resend the identical event |
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobVertexScaler.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobVertexScaler.java
index 9bc46b7e57..aba6ac6749 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobVertexScaler.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobVertexScaler.java
@@ -39,6 +39,7 @@
import static org.apache.flink.autoscaler.config.AutoScalerOptions.MAX_SCALE_DOWN_FACTOR;
import static org.apache.flink.autoscaler.config.AutoScalerOptions.MAX_SCALE_UP_FACTOR;
import static org.apache.flink.autoscaler.config.AutoScalerOptions.SCALE_UP_GRACE_PERIOD;
+import static org.apache.flink.autoscaler.config.AutoScalerOptions.SCALING_EVENT_INTERVAL;
import static org.apache.flink.autoscaler.config.AutoScalerOptions.TARGET_UTILIZATION;
import static org.apache.flink.autoscaler.config.AutoScalerOptions.VERTEX_MAX_PARALLELISM;
import static org.apache.flink.autoscaler.config.AutoScalerOptions.VERTEX_MIN_PARALLELISM;
@@ -219,7 +220,7 @@ private boolean detectIneffectiveScaleUp(
INEFFECTIVE_SCALING,
message,
null,
- null);
+ conf.get(SCALING_EVENT_INTERVAL));
if (conf.get(AutoScalerOptions.SCALING_EFFECTIVENESS_DETECTION_ENABLED)) {
LOG.warn(
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingExecutor.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingExecutor.java
index 406b17d734..0e1a47b8e7 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingExecutor.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingExecutor.java
@@ -38,23 +38,15 @@
import java.util.SortedMap;
import static org.apache.flink.autoscaler.config.AutoScalerOptions.SCALING_ENABLED;
+import static org.apache.flink.autoscaler.config.AutoScalerOptions.SCALING_EVENT_INTERVAL;
import static org.apache.flink.autoscaler.metrics.ScalingHistoryUtils.addToScalingHistoryAndStore;
import static org.apache.flink.autoscaler.metrics.ScalingHistoryUtils.getTrimmedScalingHistory;
-import static org.apache.flink.autoscaler.metrics.ScalingMetric.EXPECTED_PROCESSING_RATE;
import static org.apache.flink.autoscaler.metrics.ScalingMetric.SCALE_DOWN_RATE_THRESHOLD;
import static org.apache.flink.autoscaler.metrics.ScalingMetric.SCALE_UP_RATE_THRESHOLD;
-import static org.apache.flink.autoscaler.metrics.ScalingMetric.TARGET_DATA_RATE;
import static org.apache.flink.autoscaler.metrics.ScalingMetric.TRUE_PROCESSING_RATE;
/** Class responsible for executing scaling decisions. */
public class ScalingExecutor> {
- public static final String SCALING_SUMMARY_ENTRY =
- " Vertex ID %s | Parallelism %d -> %d | Processing capacity %.2f -> %.2f | Target data rate %.2f";
- public static final String SCALING_SUMMARY_HEADER_SCALING_DISABLED =
- "Recommended parallelism change:";
- public static final String SCALING_SUMMARY_HEADER_SCALING_ENABLED = "Scaling vertices:";
- @VisibleForTesting static final String SCALING_REPORT_REASON = "ScalingReport";
-
private static final Logger LOG = LoggerFactory.getLogger(ScalingExecutor.class);
private final JobVertexScaler jobVertexScaler;
@@ -100,18 +92,11 @@ public boolean scaleResource(
updateRecommendedParallelism(evaluatedMetrics, scalingSummaries);
- var scalingEnabled = conf.get(SCALING_ENABLED);
-
- var scalingReport = scalingReport(scalingSummaries, scalingEnabled);
- autoScalerEventHandler.handleEvent(
- context,
- AutoScalerEventHandler.Type.Normal,
- SCALING_REPORT_REASON,
- scalingReport,
- "ScalingExecutor",
- scalingEnabled ? null : conf.get(AutoScalerOptions.SCALING_REPORT_INTERVAL));
+ var scaleEnabled = conf.get(SCALING_ENABLED);
+ autoScalerEventHandler.handleScalingEvent(
+ context, scalingSummaries, scaleEnabled, conf.get(SCALING_EVENT_INTERVAL));
- if (!scalingEnabled) {
+ if (!scaleEnabled) {
return false;
}
@@ -136,27 +121,6 @@ private void updateRecommendedParallelism(
scalingSummary.getNewParallelism())));
}
- private static String scalingReport(
- Map scalingSummaries, boolean scalingEnabled) {
- StringBuilder sb =
- new StringBuilder(
- scalingEnabled
- ? SCALING_SUMMARY_HEADER_SCALING_ENABLED
- : SCALING_SUMMARY_HEADER_SCALING_DISABLED);
- scalingSummaries.forEach(
- (v, s) ->
- sb.append(
- String.format(
- SCALING_SUMMARY_ENTRY,
- v,
- s.getCurrentParallelism(),
- s.getNewParallelism(),
- s.getMetrics().get(TRUE_PROCESSING_RATE).getAverage(),
- s.getMetrics().get(EXPECTED_PROCESSING_RATE).getCurrent(),
- s.getMetrics().get(TARGET_DATA_RATE).getAverage())));
- return sb.toString();
- }
-
protected static boolean allVerticesWithinUtilizationTarget(
Map> evaluatedMetrics,
Map scalingSummaries) {
@@ -190,7 +154,8 @@ protected static boolean allVerticesWithinUtilizationTarget(
return true;
}
- private Map computeScalingSummary(
+ @VisibleForTesting
+ Map computeScalingSummary(
Context context,
Map> evaluatedMetrics,
Map> scalingHistory) {
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/config/AutoScalerOptions.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/config/AutoScalerOptions.java
index 626921bcb6..e8872a35c5 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/config/AutoScalerOptions.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/config/AutoScalerOptions.java
@@ -231,10 +231,11 @@ private static ConfigOptions.OptionBuilder autoScalerConfig(String key) {
.withDescription(
"A (semicolon-separated) list of vertex ids in hexstring for which to disable scaling. Caution: For non-sink vertices this will still scale their downstream operators until https://issues.apache.org/jira/browse/FLINK-31215 is implemented.");
- public static final ConfigOption SCALING_REPORT_INTERVAL =
- autoScalerConfig("scaling.report.interval")
+ public static final ConfigOption SCALING_EVENT_INTERVAL =
+ autoScalerConfig("scaling.event.interval")
.durationType()
.defaultValue(Duration.ofSeconds(1800))
+ .withDeprecatedKeys(deprecatedOperatorConfigKey("scaling.event.interval"))
.withDescription("Time interval to resend the identical event");
public static final ConfigOption FLINK_CLIENT_TIMEOUT =
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/event/AutoScalerEventHandler.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/event/AutoScalerEventHandler.java
index a5a0edfefe..5c49f9c550 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/event/AutoScalerEventHandler.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/event/AutoScalerEventHandler.java
@@ -19,10 +19,17 @@
import org.apache.flink.annotation.Experimental;
import org.apache.flink.autoscaler.JobAutoScalerContext;
+import org.apache.flink.autoscaler.ScalingSummary;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
import javax.annotation.Nullable;
import java.time.Duration;
+import java.util.Map;
+
+import static org.apache.flink.autoscaler.metrics.ScalingMetric.EXPECTED_PROCESSING_RATE;
+import static org.apache.flink.autoscaler.metrics.ScalingMetric.TARGET_DATA_RATE;
+import static org.apache.flink.autoscaler.metrics.ScalingMetric.TRUE_PROCESSING_RATE;
/**
* Handler for autoscaler events.
@@ -32,12 +39,17 @@
*/
@Experimental
public interface AutoScalerEventHandler> {
+ String SCALING_SUMMARY_ENTRY =
+ " Vertex ID %s | Parallelism %d -> %d | Processing capacity %.2f -> %.2f | Target data rate %.2f";
+ String SCALING_SUMMARY_HEADER_SCALING_DISABLED = "Recommended parallelism change:";
+ String SCALING_SUMMARY_HEADER_SCALING_ENABLED = "Scaling vertices:";
+ String SCALING_REPORT_REASON = "ScalingReport";
+ String SCALING_REPORT_KEY = "ScalingExecutor";
/**
* Handle the event.
*
- * @param interval When interval is great than 0, events that repeat within the interval will be
- * ignored.
+ * @param interval Define the interval to suppress duplicate events. No dedupe if null.
*/
void handleEvent(
Context context,
@@ -47,6 +59,50 @@ void handleEvent(
@Nullable String messageKey,
@Nullable Duration interval);
+ /**
+ * Handle scaling reports.
+ *
+ * @param interval Define the interval to suppress duplicate events.
+ * @param scaled Whether AutoScaler actually scaled the Flink job or just generate advice for
+ * scaling.
+ */
+ default void handleScalingEvent(
+ Context context,
+ Map scalingSummaries,
+ boolean scaled,
+ Duration interval) {
+ // Provide default implementation without proper deduplication
+ var scalingReport = scalingReport(scalingSummaries, scaled);
+ handleEvent(
+ context,
+ Type.Normal,
+ SCALING_REPORT_REASON,
+ scalingReport,
+ SCALING_REPORT_KEY,
+ interval);
+ }
+
+ static String scalingReport(
+ Map scalingSummaries, boolean scalingEnabled) {
+ StringBuilder sb =
+ new StringBuilder(
+ scalingEnabled
+ ? SCALING_SUMMARY_HEADER_SCALING_ENABLED
+ : SCALING_SUMMARY_HEADER_SCALING_DISABLED);
+ scalingSummaries.forEach(
+ (v, s) ->
+ sb.append(
+ String.format(
+ SCALING_SUMMARY_ENTRY,
+ v,
+ s.getCurrentParallelism(),
+ s.getNewParallelism(),
+ s.getMetrics().get(TRUE_PROCESSING_RATE).getAverage(),
+ s.getMetrics().get(EXPECTED_PROCESSING_RATE).getCurrent(),
+ s.getMetrics().get(TARGET_DATA_RATE).getAverage())));
+ return sb.toString();
+ }
+
/** The type of the events. */
enum Type {
Normal,
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobVertexScalerTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobVertexScalerTest.java
index d70db971dd..1b01087c8a 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobVertexScalerTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobVertexScalerTest.java
@@ -358,6 +358,39 @@ public void testSendingIneffectiveScalingEvents() {
assertThat(event.getMessage())
.isEqualTo(String.format(INEFFECTIVE_MESSAGE_FORMAT, jobVertexID));
assertThat(event.getReason()).isEqualTo(INEFFECTIVE_SCALING);
+ assertEquals(1, event.getCount());
+
+ // Repeat ineffective scale with default interval, no event is triggered
+ assertEquals(
+ 20,
+ vertexScaler.computeScaleTargetParallelism(
+ context, jobVertexID, evaluated, history));
+ assertFalse(evaluated.containsKey(ScalingMetric.EXPECTED_PROCESSING_RATE));
+ assertEquals(0, eventCollector.events.size());
+
+ // Repeat ineffective scale with postive interval, no event is triggered
+ conf.set(AutoScalerOptions.SCALING_EVENT_INTERVAL, Duration.ofSeconds(1800));
+ assertEquals(
+ 20,
+ vertexScaler.computeScaleTargetParallelism(
+ context, jobVertexID, evaluated, history));
+ assertFalse(evaluated.containsKey(ScalingMetric.EXPECTED_PROCESSING_RATE));
+ assertEquals(0, eventCollector.events.size());
+
+ // Ineffective scale with interval set to 0, an event is triggered
+ conf.set(AutoScalerOptions.SCALING_EVENT_INTERVAL, Duration.ZERO);
+ assertEquals(
+ 20,
+ vertexScaler.computeScaleTargetParallelism(
+ context, jobVertexID, evaluated, history));
+ assertFalse(evaluated.containsKey(ScalingMetric.EXPECTED_PROCESSING_RATE));
+ assertEquals(1, eventCollector.events.size());
+ event = eventCollector.events.poll();
+ assertThat(event).isNotNull();
+ assertThat(event.getMessage())
+ .isEqualTo(String.format(INEFFECTIVE_MESSAGE_FORMAT, jobVertexID));
+ assertThat(event.getReason()).isEqualTo(INEFFECTIVE_SCALING);
+ assertEquals(2, event.getCount());
}
private Map evaluated(
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingExecutorTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingExecutorTest.java
index 36fb6e31db..bae6f36b2a 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingExecutorTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingExecutorTest.java
@@ -37,8 +37,11 @@
import java.util.Map;
import java.util.stream.Collectors;
-import static org.apache.flink.autoscaler.ScalingExecutor.SCALING_SUMMARY_ENTRY;
import static org.apache.flink.autoscaler.TestingAutoscalerUtils.createDefaultJobAutoScalerContext;
+import static org.apache.flink.autoscaler.event.AutoScalerEventHandler.SCALING_REPORT_REASON;
+import static org.apache.flink.autoscaler.event.AutoScalerEventHandler.SCALING_SUMMARY_ENTRY;
+import static org.apache.flink.autoscaler.event.AutoScalerEventHandler.SCALING_SUMMARY_HEADER_SCALING_DISABLED;
+import static org.apache.flink.autoscaler.event.AutoScalerEventHandler.SCALING_SUMMARY_HEADER_SCALING_ENABLED;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -151,19 +154,20 @@ public void testVertexesExclusionForScaling() throws Exception {
@ParameterizedTest
@ValueSource(booleans = {true, false})
- public void testScalingEventsWith0Interval(boolean scalingEnabled) throws Exception {
+ public void testScalingEventsWith0IntervalConfig(boolean scalingEnabled) throws Exception {
testScalingEvents(scalingEnabled, Duration.ofSeconds(0));
}
@ParameterizedTest
@ValueSource(booleans = {true, false})
- public void testScalingEventsWithInterval(boolean scalingEnabled) throws Exception {
+ public void testScalingEventsWithIntervalConfig(boolean scalingEnabled) throws Exception {
testScalingEvents(scalingEnabled, Duration.ofSeconds(1800));
}
@ParameterizedTest
@ValueSource(booleans = {true, false})
- public void testScalingEventsWithDefaultInterval(boolean scalingEnabled) throws Exception {
+ public void testScalingEventsWithDefaultIntervalConfig(boolean scalingEnabled)
+ throws Exception {
testScalingEvents(scalingEnabled, null);
}
@@ -175,17 +179,13 @@ private void testScalingEvents(boolean scalingEnabled, Duration interval) throws
var metrics = Map.of(jobVertexID, evaluated(1, 110, 100));
if (interval != null) {
- conf.set(AutoScalerOptions.SCALING_REPORT_INTERVAL, interval);
+ conf.set(AutoScalerOptions.SCALING_EVENT_INTERVAL, interval);
}
assertEquals(scalingEnabled, scalingDecisionExecutor.scaleResource(context, metrics));
assertEquals(scalingEnabled, scalingDecisionExecutor.scaleResource(context, metrics));
- int expectedSize =
- (interval == null || (!interval.isNegative() && !interval.isZero()))
- && !scalingEnabled
- ? 1
- : 2;
+ int expectedSize = (interval == null || interval.toMillis() > 0) && !scalingEnabled ? 1 : 2;
assertEquals(expectedSize, eventCollector.events.size());
TestingEventCollector.Event> event;
@@ -208,9 +208,9 @@ private void testScalingEvents(boolean scalingEnabled, Duration interval) throws
event.getMessage()
.contains(
scalingEnabled
- ? ScalingExecutor.SCALING_SUMMARY_HEADER_SCALING_ENABLED
- : ScalingExecutor.SCALING_SUMMARY_HEADER_SCALING_DISABLED));
- assertEquals(ScalingExecutor.SCALING_REPORT_REASON, event.getReason());
+ ? SCALING_SUMMARY_HEADER_SCALING_ENABLED
+ : SCALING_SUMMARY_HEADER_SCALING_DISABLED));
+ assertEquals(SCALING_REPORT_REASON, event.getReason());
metrics = Map.of(jobVertexID, evaluated(1, 110, 101));
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/event/TestingEventCollector.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/event/TestingEventCollector.java
index bd742cde5e..485f9c7e22 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/event/TestingEventCollector.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/event/TestingEventCollector.java
@@ -20,6 +20,7 @@
import org.apache.flink.autoscaler.JobAutoScalerContext;
import lombok.Getter;
+import lombok.Setter;
import javax.annotation.Nullable;
@@ -30,6 +31,8 @@
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
+import static org.apache.flink.autoscaler.config.AutoScalerOptions.SCALING_ENABLED;
+
/** Testing {@link AutoScalerEventHandler} implementation. */
public class TestingEventCollector>
implements AutoScalerEventHandler {
@@ -45,17 +48,18 @@ public void handleEvent(
String reason,
String message,
@Nullable String messageKey,
- @Nullable Duration interval) {
+ Duration interval) {
String eventKey =
generateEventKey(context, type, reason, messageKey != null ? messageKey : message);
Event event = eventMap.get(eventKey);
+ var scaled = context.getConfiguration().get(SCALING_ENABLED);
if (event == null) {
- Event newEvent = new Event<>(context, type, reason, message, messageKey);
+ Event newEvent = new Event<>(context, reason, message, messageKey);
events.add(newEvent);
eventMap.put(eventKey, newEvent);
return;
- }
- if (Objects.equals(event.getMessage(), message)
+ } else if (((!scaled && Objects.equals(event.getMessage(), message))
+ || !Objects.equals(reason, SCALING_REPORT_REASON))
&& interval != null
&& Instant.now()
.isBefore(event.getLastUpdateTimestamp().plusMillis(interval.toMillis()))) {
@@ -63,6 +67,8 @@ public void handleEvent(
return;
}
event.incrementCount();
+ event.setMessage(message);
+ event.setLastUpdateTimestamp(Instant.now());
events.add(event);
}
@@ -73,29 +79,21 @@ private String generateEventKey(Context context, Type type, String reason, Strin
/** The collected event. */
public static class Event> {
- @Getter private Instant lastUpdateTimestamp;
+ @Getter @Setter private Instant lastUpdateTimestamp;
@Getter private final Context context;
- @Getter private final Type type;
-
@Getter private final String reason;
- @Getter private final String message;
+ @Getter @Setter private String message;
@Getter @Nullable private final String messageKey;
@Getter private int count;
- public Event(
- Context context,
- Type type,
- String reason,
- String message,
- @Nullable String messageKey) {
+ public Event(Context context, String reason, String message, @Nullable String messageKey) {
this.lastUpdateTimestamp = Instant.now();
this.context = context;
- this.type = type;
this.reason = reason;
this.message = message;
this.messageKey = messageKey;
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/KubernetesAutoScalerEventHandler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/KubernetesAutoScalerEventHandler.java
index 54fe84ae25..d83e2db4d0 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/KubernetesAutoScalerEventHandler.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/KubernetesAutoScalerEventHandler.java
@@ -17,19 +17,26 @@
package org.apache.flink.kubernetes.operator.autoscaler;
+import org.apache.flink.autoscaler.ScalingSummary;
import org.apache.flink.autoscaler.event.AutoScalerEventHandler;
import org.apache.flink.kubernetes.operator.utils.EventRecorder;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
import io.javaoperatorsdk.operator.processing.event.ResourceID;
import javax.annotation.Nullable;
import java.time.Duration;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
/** An event handler which posts events to the Kubernetes events API. */
public class KubernetesAutoScalerEventHandler
implements AutoScalerEventHandler {
+ public static final String PARALLELISM_MAP_KEY = "parallelismMap";
private final EventRecorder eventRecorder;
public KubernetesAutoScalerEventHandler(EventRecorder eventRecorder) {
@@ -44,25 +51,71 @@ public void handleEvent(
String message,
@Nullable String messageKey,
@Nullable Duration interval) {
- if (interval == null) {
- eventRecorder.triggerEvent(
- context.getResource(),
- EventRecorder.Type.valueOf(type.name()),
- reason,
- message,
- EventRecorder.Component.Operator,
- messageKey,
- context.getKubernetesClient());
+ eventRecorder.triggerEventWithInterval(
+ context.getResource(),
+ EventRecorder.Type.valueOf(type.name()),
+ reason,
+ message,
+ EventRecorder.Component.Operator,
+ messageKey,
+ context.getKubernetesClient(),
+ interval);
+ }
+
+ @Override
+ public void handleScalingEvent(
+ KubernetesJobAutoScalerContext context,
+ Map scalingSummaries,
+ boolean scaled,
+ Duration interval) {
+ if (scaled) {
+ AutoScalerEventHandler.super.handleScalingEvent(
+ context, scalingSummaries, scaled, null);
} else {
- eventRecorder.triggerEventByInterval(
+ var conf = context.getConfiguration();
+ var scalingReport = AutoScalerEventHandler.scalingReport(scalingSummaries, scaled);
+ var labels = Map.of(PARALLELISM_MAP_KEY, getParallelismHashCode(scalingSummaries));
+
+ @Nullable
+ Predicate> dedupePredicate =
+ new Predicate>() {
+ @Override
+ public boolean test(Map stringStringMap) {
+ return stringStringMap != null
+ && Objects.equals(
+ stringStringMap.get(PARALLELISM_MAP_KEY),
+ getParallelismHashCode(scalingSummaries));
+ }
+ };
+
+ eventRecorder.triggerEventWithLabels(
context.getResource(),
- EventRecorder.Type.valueOf(type.name()),
- reason,
+ EventRecorder.Type.Normal,
+ AutoScalerEventHandler.SCALING_REPORT_REASON,
+ scalingReport,
EventRecorder.Component.Operator,
- message,
- messageKey,
+ AutoScalerEventHandler.SCALING_REPORT_KEY,
context.getKubernetesClient(),
- interval);
+ interval,
+ dedupePredicate,
+ labels);
}
}
+
+ private static String getParallelismHashCode(
+ Map scalingSummaryHashMap) {
+ return Integer.toString(
+ scalingSummaryHashMap.entrySet().stream()
+ .collect(
+ Collectors.toMap(
+ e -> e.getKey().toString(),
+ e ->
+ String.format(
+ "Parallelism %d -> %d",
+ e.getValue()
+ .getCurrentParallelism(),
+ e.getValue().getNewParallelism())))
+ .hashCode()
+ & 0x7FFFFFFF);
+ }
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java
index 6f39582a4d..dfe98a8968 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java
@@ -203,8 +203,8 @@ private void triggerSpecChangeEvent(CR cr, DiffResult specDiff, Kubernetes
cr,
EventRecorder.Type.Normal,
EventRecorder.Reason.SpecChanged,
- EventRecorder.Component.JobManagerDeployment,
String.format(MSG_SPEC_CHANGED, specDiff.getType(), specDiff),
+ EventRecorder.Component.JobManagerDeployment,
"SpecChange: " + cr.getMetadata().getGeneration(),
client);
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventRecorder.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventRecorder.java
index f29b3a0051..e99c8ef9d2 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventRecorder.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventRecorder.java
@@ -30,7 +30,9 @@
import java.time.Duration;
import java.util.Collection;
+import java.util.Map;
import java.util.function.BiConsumer;
+import java.util.function.Predicate;
/** Helper class for creating Kubernetes events for Flink resources. */
public class EventRecorder {
@@ -48,15 +50,15 @@ public boolean triggerEvent(
Component component,
String message,
KubernetesClient client) {
- return triggerEvent(resource, type, reason, component, message, null, client);
+ return triggerEvent(resource, type, reason, message, component, null, client);
}
public boolean triggerEventOnce(
AbstractFlinkResource resource,
Type type,
Reason reason,
- Component component,
String message,
+ Component component,
String messageKey,
KubernetesClient client) {
return triggerEventOnce(
@@ -67,8 +69,8 @@ public boolean triggerEvent(
AbstractFlinkResource resource,
Type type,
Reason reason,
- Component component,
String message,
+ Component component,
@Nullable String messageKey,
KubernetesClient client) {
return triggerEvent(
@@ -83,7 +85,7 @@ public boolean triggerEvent(
Component component,
String messageKey,
KubernetesClient client) {
- return EventUtils.createOrUpdateEvent(
+ return EventUtils.createOrUpdateEventWithInterval(
client,
resource,
type,
@@ -91,7 +93,33 @@ public boolean triggerEvent(
message,
component,
e -> eventListener.accept(resource, e),
- messageKey);
+ messageKey,
+ null);
+ }
+
+ /**
+ * @param interval Interval for dedupe. Null mean no dedupe.
+ * @return
+ */
+ public boolean triggerEventWithInterval(
+ AbstractFlinkResource resource,
+ Type type,
+ String reason,
+ String message,
+ Component component,
+ String messageKey,
+ KubernetesClient client,
+ @Nullable Duration interval) {
+ return EventUtils.createOrUpdateEventWithInterval(
+ client,
+ resource,
+ type,
+ reason,
+ message,
+ component,
+ e -> eventListener.accept(resource, e),
+ messageKey,
+ interval);
}
public boolean triggerEventOnce(
@@ -113,16 +141,24 @@ public boolean triggerEventOnce(
messageKey);
}
- public boolean triggerEventByInterval(
+ /**
+ * @param interval Interval for dedupe. Null mean no dedupe.
+ * @param dedupePredicate Predicate for dedupe algorithm..
+ * @param labels Labels to store in meta data for dedupe. Do nothing if null.
+ * @return
+ */
+ public boolean triggerEventWithLabels(
AbstractFlinkResource resource,
Type type,
String reason,
- Component component,
String message,
+ Component component,
@Nullable String messageKey,
KubernetesClient client,
- Duration interval) {
- return EventUtils.createByInterval(
+ @Nullable Duration interval,
+ @Nullable Predicate> dedupePredicate,
+ @Nullable Map labels) {
+ return EventUtils.createOrUpdateEventWithLabels(
client,
resource,
type,
@@ -131,7 +167,9 @@ public boolean triggerEventByInterval(
component,
e -> eventListener.accept(resource, e),
messageKey,
- interval);
+ interval,
+ dedupePredicate,
+ labels);
}
public boolean triggerEvent(
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventUtils.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventUtils.java
index dfdbdd9e79..ef32cd5915 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventUtils.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventUtils.java
@@ -20,6 +20,7 @@
import io.fabric8.kubernetes.api.model.Event;
import io.fabric8.kubernetes.api.model.EventBuilder;
import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.fabric8.kubernetes.api.model.ObjectMeta;
import io.fabric8.kubernetes.api.model.ObjectReferenceBuilder;
import io.fabric8.kubernetes.client.KubernetesClient;
@@ -27,8 +28,9 @@
import java.time.Duration;
import java.time.Instant;
-import java.util.Objects;
+import java.util.Map;
import java.util.function.Consumer;
+import java.util.function.Predicate;
/**
* The util to generate an event for the target resource. It is copied from
@@ -54,7 +56,7 @@ public static String generateEventName(
& 0x7FFFFFFF);
}
- public static boolean createOrUpdateEvent(
+ public static boolean createOrUpdateEventWithInterval(
KubernetesClient client,
HasMetadata target,
EventRecorder.Type type,
@@ -62,8 +64,9 @@ public static boolean createOrUpdateEvent(
String message,
EventRecorder.Component component,
Consumer eventListener,
- @Nullable String messageKey) {
- return createByInterval(
+ @Nullable String messageKey,
+ @Nullable Duration interval) {
+ return createOrUpdateEventWithLabels(
client,
target,
type,
@@ -72,10 +75,12 @@ public static boolean createOrUpdateEvent(
component,
eventListener,
messageKey,
- Duration.ofSeconds(0));
+ interval,
+ null,
+ Map.of());
}
- private static Event findExistingEvent(
+ public static Event findExistingEvent(
KubernetesClient client, HasMetadata target, String eventName) {
return client.v1()
.events()
@@ -102,13 +107,13 @@ public static boolean createIfNotExists(
if (existing != null) {
return false;
} else {
- createNewEvent(
- client, target, type, reason, message, component, eventListener, eventName);
+ Event event = buildEvent(target, type, reason, message, component, eventName);
+ eventListener.accept(client.resource(event).createOrReplace());
return true;
}
}
- public static boolean createByInterval(
+ public static boolean createOrUpdateEventWithLabels(
KubernetesClient client,
HasMetadata target,
EventRecorder.Type type,
@@ -117,53 +122,49 @@ public static boolean createByInterval(
EventRecorder.Component component,
Consumer eventListener,
@Nullable String messageKey,
- Duration interval) {
-
+ @Nullable Duration interval,
+ @Nullable Predicate> dedupePredicate,
+ @Nullable Map labels) {
String eventName =
generateEventName(
target, type, reason, messageKey != null ? messageKey : message, component);
Event existing = findExistingEvent(client, target, eventName);
if (existing != null) {
- if (Objects.equals(existing.getMessage(), message)
- && Instant.now()
- .isBefore(
- Instant.parse(existing.getLastTimestamp())
- .plusMillis(interval.toMillis()))) {
- return false;
- } else {
- createUpdatedEvent(existing, client, message, eventListener);
+ if (labelCheck(existing, dedupePredicate) && intervalCheck(existing, interval)) {
return false;
}
+ updatedEventWithLabels(existing, client, message, eventListener, labels);
+ return false;
} else {
- createNewEvent(
- client, target, type, reason, message, component, eventListener, eventName);
+ Event event = buildEvent(target, type, reason, message, component, eventName);
+ setLabels(event, labels);
+ eventListener.accept(client.resource(event).createOrReplace());
return true;
}
}
- private static void createUpdatedEvent(
+ private static void updatedEventWithLabels(
Event existing,
KubernetesClient client,
String message,
- Consumer eventListener) {
+ Consumer eventListener,
+ @Nullable Map labels) {
existing.setLastTimestamp(Instant.now().toString());
existing.setCount(existing.getCount() + 1);
existing.setMessage(message);
+ setLabels(existing, labels);
eventListener.accept(client.resource(existing).createOrReplace());
}
- private static void createNewEvent(
- KubernetesClient client,
- HasMetadata target,
- EventRecorder.Type type,
- String reason,
- String message,
- EventRecorder.Component component,
- Consumer eventListener,
- String eventName) {
- Event event = buildEvent(target, type, reason, message, component, eventName);
- eventListener.accept(client.resource(event).createOrReplace());
+ private static void setLabels(Event existing, @Nullable Map labels) {
+ if (labels == null) {
+ return;
+ }
+ if (existing.getMetadata() == null) {
+ existing.setMetadata(new ObjectMeta());
+ }
+ existing.getMetadata().setLabels(labels);
}
private static Event buildEvent(
@@ -197,4 +198,19 @@ private static Event buildEvent(
.endMetadata()
.build();
}
+
+ private static boolean intervalCheck(Event existing, @Nullable Duration interval) {
+ return interval != null
+ && Instant.now()
+ .isBefore(
+ Instant.parse(existing.getLastTimestamp())
+ .plusMillis(interval.toMillis()));
+ }
+
+ private static boolean labelCheck(
+ Event existing, Predicate> dedupePredicate) {
+ return dedupePredicate == null
+ || (existing.getMetadata() != null
+ && dedupePredicate.test(existing.getMetadata().getLabels()));
+ }
}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/KubernetesAutoScalerEventHandlerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/KubernetesAutoScalerEventHandlerTest.java
new file mode 100644
index 0000000000..bed6a6ab2c
--- /dev/null
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/KubernetesAutoScalerEventHandlerTest.java
@@ -0,0 +1,332 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.kubernetes.operator.autoscaler;
+
+import org.apache.flink.autoscaler.ScalingSummary;
+import org.apache.flink.autoscaler.event.AutoScalerEventHandler;
+import org.apache.flink.autoscaler.metrics.EvaluatedScalingMetric;
+import org.apache.flink.autoscaler.metrics.ScalingMetric;
+import org.apache.flink.kubernetes.operator.utils.EventCollector;
+import org.apache.flink.kubernetes.operator.utils.EventRecorder;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+
+import io.fabric8.kubernetes.api.model.Event;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.time.Duration;
+import java.util.Map;
+
+import static org.apache.flink.autoscaler.event.AutoScalerEventHandler.SCALING_SUMMARY_ENTRY;
+import static org.apache.flink.kubernetes.operator.autoscaler.KubernetesAutoScalerEventHandler.PARALLELISM_MAP_KEY;
+import static org.apache.flink.kubernetes.operator.autoscaler.TestingKubernetesAutoscalerUtils.createContext;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/** Test for {@link KubernetesAutoScalerStateStore}. */
+@EnableKubernetesMockClient(crud = true)
+public class KubernetesAutoScalerEventHandlerTest {
+
+ private KubernetesClient kubernetesClient;
+
+ private KubernetesAutoScalerEventHandler eventHandler;
+
+ private KubernetesJobAutoScalerContext ctx;
+
+ ConfigMapStore configMapStore;
+
+ KubernetesAutoScalerStateStore stateStore;
+
+ private EventCollector eventCollector;
+
+ @BeforeEach
+ void setup() {
+ eventCollector = new EventCollector();
+ var eventRecorder = new EventRecorder(eventCollector);
+ ctx = createContext("cr1", kubernetesClient);
+ eventHandler = new KubernetesAutoScalerEventHandler(eventRecorder);
+ stateStore = new KubernetesAutoScalerStateStore(configMapStore);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"", "0", "1800"})
+ void testHandEventsWithNoMessageKey(String intervalString) {
+ testHandEvents(intervalString, null);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"", "0", "1800"})
+ void testHandEventsWithMessageKey(String intervalString) {
+ testHandEvents(intervalString, "key");
+ }
+
+ private void testHandEvents(String intervalString, String messageKey) {
+ Duration interval =
+ intervalString.isBlank() ? null : Duration.ofSeconds(Long.valueOf(intervalString));
+ var jobVertexID = new JobVertexID();
+
+ eventHandler.handleEvent(
+ ctx,
+ AutoScalerEventHandler.Type.Normal,
+ EventRecorder.Reason.IneffectiveScaling.name(),
+ "message",
+ messageKey,
+ interval);
+ var event = eventCollector.events.poll();
+ assertEquals(EventRecorder.Reason.IneffectiveScaling.name(), event.getReason());
+ assertEquals(1, event.getCount());
+
+ // Resend
+ eventHandler.handleEvent(
+ ctx,
+ AutoScalerEventHandler.Type.Normal,
+ EventRecorder.Reason.IneffectiveScaling.name(),
+ "message",
+ messageKey,
+ interval);
+ if (interval != null && interval.toMillis() > 0) {
+ assertEquals(0, eventCollector.events.size());
+ } else {
+ assertEquals(1, eventCollector.events.size());
+ event = eventCollector.events.poll();
+ assertEquals("message", event.getMessage());
+ assertEquals(2, event.getCount());
+ }
+
+ // Message changed
+ eventHandler.handleEvent(
+ ctx,
+ AutoScalerEventHandler.Type.Normal,
+ EventRecorder.Reason.IneffectiveScaling.name(),
+ "message1",
+ messageKey,
+ interval);
+ if (messageKey != null && interval != null && interval.toMillis() > 0) {
+ assertEquals(0, eventCollector.events.size());
+ } else {
+ assertEquals(1, eventCollector.events.size());
+ event = eventCollector.events.poll();
+ assertEquals("message1", event.getMessage());
+ assertEquals(messageKey == null ? 1 : 3, event.getCount());
+ }
+
+ // Message key changed
+ eventHandler.handleEvent(
+ ctx,
+ AutoScalerEventHandler.Type.Normal,
+ EventRecorder.Reason.IneffectiveScaling.name(),
+ "message1",
+ "newKey",
+ interval);
+ assertEquals(1, eventCollector.events.size());
+ event = eventCollector.events.poll();
+ assertEquals("message1", event.getMessage());
+ assertEquals(1, event.getCount());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testHandleScalingEventsWith0Interval(boolean scalingEnabled) {
+ testHandleScalingEvents(scalingEnabled, Duration.ofSeconds(0));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testHandleScalingEventsWithInterval(boolean scalingEnabled) {
+ testHandleScalingEvents(scalingEnabled, Duration.ofSeconds(1800));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testHandleScalingEventsWithNullInterval(boolean scalingEnabled) {
+ testHandleScalingEvents(scalingEnabled, null);
+ }
+
+ private void testHandleScalingEvents(boolean scaled, Duration interval) {
+ var jobVertexID = JobVertexID.fromHexString("1b51e99e55e89e404d9a0443fd98d9e2");
+
+ var evaluatedScalingMetric = new EvaluatedScalingMetric();
+ evaluatedScalingMetric.setAverage(1);
+ evaluatedScalingMetric.setCurrent(2);
+ Map scalingSummaries1 =
+ Map.of(
+ jobVertexID,
+ new ScalingSummary(
+ 1,
+ 2,
+ Map.of(
+ ScalingMetric.TRUE_PROCESSING_RATE,
+ evaluatedScalingMetric,
+ ScalingMetric.EXPECTED_PROCESSING_RATE,
+ evaluatedScalingMetric,
+ ScalingMetric.TARGET_DATA_RATE,
+ evaluatedScalingMetric)));
+
+ eventHandler.handleScalingEvent(ctx, scalingSummaries1, scaled, interval);
+ var event = eventCollector.events.poll();
+ assertTrue(
+ event.getMessage()
+ .contains(
+ String.format(
+ SCALING_SUMMARY_ENTRY,
+ jobVertexID,
+ 1,
+ 2,
+ 1.00,
+ 2.00,
+ 1.00)));
+
+ assertEquals(EventRecorder.Reason.ScalingReport.name(), event.getReason());
+ assertEquals(
+ scaled ? null : "1286380436",
+ event.getMetadata().getLabels().get(PARALLELISM_MAP_KEY));
+ assertEquals(1, event.getCount());
+
+ // Parallelism map doesn't change.
+ eventHandler.handleScalingEvent(ctx, scalingSummaries1, scaled, interval);
+ Event newEvent;
+ if (interval != null && interval.toMillis() > 0 && !scaled) {
+ assertEquals(0, eventCollector.events.size());
+ } else {
+ assertEquals(1, eventCollector.events.size());
+ newEvent = eventCollector.events.poll();
+ assertEquals(event.getMetadata().getUid(), newEvent.getMetadata().getUid());
+ assertEquals(2, newEvent.getCount());
+ }
+
+ // Parallelism map changes. New recommendation
+ Map scalingSummaries2 =
+ Map.of(
+ jobVertexID,
+ new ScalingSummary(
+ 1,
+ 3,
+ Map.of(
+ ScalingMetric.TRUE_PROCESSING_RATE,
+ evaluatedScalingMetric,
+ ScalingMetric.EXPECTED_PROCESSING_RATE,
+ evaluatedScalingMetric,
+ ScalingMetric.TARGET_DATA_RATE,
+ evaluatedScalingMetric)));
+ eventHandler.handleScalingEvent(ctx, scalingSummaries2, scaled, interval);
+
+ assertEquals(1, eventCollector.events.size());
+
+ newEvent = eventCollector.events.poll();
+ assertEquals(event.getMetadata().getUid(), newEvent.getMetadata().getUid());
+ assertEquals(
+ interval != null && interval.toMillis() > 0 && !scaled ? 2 : 3,
+ newEvent.getCount());
+
+ // Parallelism map doesn't change but metrics changed.
+ evaluatedScalingMetric.setCurrent(3);
+ Map scalingSummaries3 =
+ Map.of(
+ jobVertexID,
+ new ScalingSummary(
+ 1,
+ 3,
+ Map.of(
+ ScalingMetric.TRUE_PROCESSING_RATE,
+ evaluatedScalingMetric,
+ ScalingMetric.EXPECTED_PROCESSING_RATE,
+ evaluatedScalingMetric,
+ ScalingMetric.TARGET_DATA_RATE,
+ evaluatedScalingMetric)));
+ eventHandler.handleScalingEvent(ctx, scalingSummaries2, scaled, interval);
+
+ if (interval != null && interval.toMillis() > 0 && !scaled) {
+ assertEquals(0, eventCollector.events.size());
+ } else {
+ assertEquals(1, eventCollector.events.size());
+ newEvent = eventCollector.events.poll();
+ assertEquals(4, newEvent.getCount());
+ }
+ }
+
+ @Test
+ public void testSwitchingScalingEnabled() {
+ var jobVertexID = JobVertexID.fromHexString("1b51e99e55e89e404d9a0443fd98d9e2");
+ var evaluatedScalingMetric = new EvaluatedScalingMetric();
+ var interval = Duration.ofSeconds(1800);
+ evaluatedScalingMetric.setAverage(1);
+ evaluatedScalingMetric.setCurrent(2);
+ Map scalingSummaries1 =
+ Map.of(
+ jobVertexID,
+ new ScalingSummary(
+ 1,
+ 2,
+ Map.of(
+ ScalingMetric.TRUE_PROCESSING_RATE,
+ evaluatedScalingMetric,
+ ScalingMetric.EXPECTED_PROCESSING_RATE,
+ evaluatedScalingMetric,
+ ScalingMetric.TARGET_DATA_RATE,
+ evaluatedScalingMetric)));
+
+ eventHandler.handleScalingEvent(ctx, scalingSummaries1, true, interval);
+ var event = eventCollector.events.poll();
+ assertEquals(null, event.getMetadata().getLabels().get(PARALLELISM_MAP_KEY));
+ assertEquals(1, event.getCount());
+
+ // Get recommendation event even parallelism map doesn't change and within supression
+ // interval
+ eventHandler.handleScalingEvent(ctx, scalingSummaries1, false, interval);
+ assertEquals(1, eventCollector.events.size());
+ event = eventCollector.events.poll();
+ assertTrue(
+ event.getMessage()
+ .contains(
+ String.format(
+ SCALING_SUMMARY_ENTRY,
+ jobVertexID,
+ 1,
+ 2,
+ 1.00,
+ 2.00,
+ 1.00)));
+
+ assertEquals("1286380436", event.getMetadata().getLabels().get(PARALLELISM_MAP_KEY));
+ assertEquals(2, event.getCount());
+
+ // Get recommendation event even parallelism map doesn't change and within supression
+ // interval
+ eventHandler.handleScalingEvent(ctx, scalingSummaries1, true, interval);
+ assertEquals(1, eventCollector.events.size());
+ event = eventCollector.events.poll();
+ assertTrue(
+ event.getMessage()
+ .contains(
+ String.format(
+ SCALING_SUMMARY_ENTRY,
+ jobVertexID,
+ 1,
+ 2,
+ 1.00,
+ 2.00,
+ 1.00)));
+
+ assertEquals(null, event.getMetadata().getLabels().get(PARALLELISM_MAP_KEY));
+ assertEquals(3, event.getCount());
+ }
+}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/EventUtilsTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/EventUtilsTest.java
index c58b020b35..24a4c99939 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/EventUtilsTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/EventUtilsTest.java
@@ -25,9 +25,15 @@
import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import javax.annotation.Nullable;
import java.time.Duration;
+import java.util.Map;
import java.util.function.Consumer;
+import java.util.function.Predicate;
/** Test for {@link EventUtils}. */
@EnableKubernetesMockClient(crud = true)
@@ -57,7 +63,7 @@ public void accept(Event event) {
message,
EventRecorder.Component.Operator);
Assertions.assertTrue(
- EventUtils.createOrUpdateEvent(
+ EventUtils.createOrUpdateEventWithInterval(
kubernetesClient,
flinkApp,
EventRecorder.Type.Warning,
@@ -65,6 +71,7 @@ public void accept(Event event) {
message,
EventRecorder.Component.Operator,
consumer,
+ null,
null));
var event =
kubernetesClient
@@ -80,7 +87,7 @@ public void accept(Event event) {
eventConsumed = null;
Assertions.assertFalse(
- EventUtils.createOrUpdateEvent(
+ EventUtils.createOrUpdateEventWithInterval(
kubernetesClient,
flinkApp,
EventRecorder.Type.Warning,
@@ -88,7 +95,9 @@ public void accept(Event event) {
message,
EventRecorder.Component.Operator,
consumer,
+ null,
null));
+
event =
kubernetesClient
.v1()
@@ -101,7 +110,7 @@ public void accept(Event event) {
Assertions.assertEquals(2, event.getCount());
Assertions.assertTrue(
- EventUtils.createOrUpdateEvent(
+ EventUtils.createOrUpdateEventWithInterval(
kubernetesClient,
flinkApp,
EventRecorder.Type.Warning,
@@ -109,11 +118,15 @@ public void accept(Event event) {
null,
EventRecorder.Component.Operator,
consumer,
+ null,
null));
}
- @Test
- public void testCreateUpdatedEvent() {
+ @ParameterizedTest
+ @ValueSource(strings = {"", "0", "1800"})
+ public void testCreateWithInterval(String intervalString) {
+ Duration interval =
+ intervalString.isBlank() ? null : Duration.ofSeconds(Long.valueOf(intervalString));
var consumer =
new Consumer() {
@Override
@@ -123,25 +136,25 @@ public void accept(Event event) {
};
var flinkApp = TestUtils.buildApplicationCluster();
var reason = "Cleanup";
- var message = "message";
var eventName =
EventUtils.generateEventName(
flinkApp,
EventRecorder.Type.Warning,
reason,
- message,
+ "mk",
EventRecorder.Component.Operator);
+
Assertions.assertTrue(
- EventUtils.createByInterval(
+ EventUtils.createOrUpdateEventWithInterval(
kubernetesClient,
flinkApp,
EventRecorder.Type.Warning,
reason,
- message,
+ "message1",
EventRecorder.Component.Operator,
consumer,
- null,
- Duration.ofSeconds(1800)));
+ "mk",
+ interval));
var event =
kubernetesClient
.v1()
@@ -150,22 +163,21 @@ public void accept(Event event) {
.withName(eventName)
.get();
Assertions.assertNotNull(event);
- Assertions.assertEquals(eventConsumed, event);
+ Assertions.assertEquals("message1", event.getMessage());
Assertions.assertEquals(1, event.getCount());
- Assertions.assertEquals(reason, event.getReason());
- eventConsumed = null;
Assertions.assertFalse(
- EventUtils.createByInterval(
+ EventUtils.createOrUpdateEventWithInterval(
kubernetesClient,
flinkApp,
EventRecorder.Type.Warning,
reason,
- message,
+ "message2",
EventRecorder.Component.Operator,
consumer,
- null,
- Duration.ofSeconds(1800)));
+ "mk",
+ null));
+
event =
kubernetesClient
.v1()
@@ -174,24 +186,48 @@ public void accept(Event event) {
.withName(eventName)
.get();
Assertions.assertNotNull(event);
- Assertions.assertNull(eventConsumed);
- Assertions.assertEquals(1, event.getCount());
+ Assertions.assertEquals("message2", event.getMessage());
+ Assertions.assertEquals(2, event.getCount());
+ }
- Assertions.assertTrue(
- EventUtils.createByInterval(
- kubernetesClient,
- flinkApp,
- EventRecorder.Type.Warning,
- reason,
- null,
- EventRecorder.Component.Operator,
- consumer,
- null,
- Duration.ofSeconds(1800)));
+ @ParameterizedTest
+ @ValueSource(strings = {"", "0", "1800"})
+ public void testCreateWithLabelsAndAllTruePredicate(String intervalString) {
+ @Nullable
+ Predicate> dedupePredicate =
+ new Predicate>() {
+ @Override
+ public boolean test(Map stringStringMap) {
+ return true;
+ }
+ };
+ testCreateWithIntervalLabelsAndPredicate(intervalString, dedupePredicate);
}
- @Test
- public void testCreateWithMessageKey() {
+ @ParameterizedTest
+ @ValueSource(strings = {"", "0", "1800"})
+ public void testCreateWithLabelsAndAllFalsePredicate(String intervalString) {
+ @Nullable
+ Predicate> dedupePredicate =
+ new Predicate>() {
+ @Override
+ public boolean test(Map stringStringMap) {
+ return false;
+ }
+ };
+ testCreateWithIntervalLabelsAndPredicate(intervalString, dedupePredicate);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"", "0", "1800"})
+ public void testCreateWithLabelsAndNullPredicate(String intervalString) {
+ testCreateWithIntervalLabelsAndPredicate(intervalString, null);
+ }
+
+ private void testCreateWithIntervalLabelsAndPredicate(
+ String intervalString, @Nullable Predicate> dedupePredicate) {
+ Duration interval =
+ intervalString.isBlank() ? null : Duration.ofSeconds(Long.valueOf(intervalString));
var consumer =
new Consumer() {
@Override
@@ -209,39 +245,34 @@ public void accept(Event event) {
"mk",
EventRecorder.Component.Operator);
+ // Set up an event with empty labels
Assertions.assertTrue(
- EventUtils.createOrUpdateEvent(
+ EventUtils.createIfNotExists(
kubernetesClient,
flinkApp,
EventRecorder.Type.Warning,
reason,
- "message1",
+ "message",
EventRecorder.Component.Operator,
consumer,
"mk"));
- var event =
- kubernetesClient
- .v1()
- .events()
- .inNamespace(flinkApp.getMetadata().getNamespace())
- .withName(eventName)
- .get();
- Assertions.assertNotNull(event);
- Assertions.assertEquals("message1", event.getMessage());
- Assertions.assertEquals(1, event.getCount());
+ // Update the event with label.
+ var labels = Map.of("a", "b");
Assertions.assertFalse(
- EventUtils.createOrUpdateEvent(
+ EventUtils.createOrUpdateEventWithLabels(
kubernetesClient,
flinkApp,
EventRecorder.Type.Warning,
reason,
- "message2",
+ "message1",
EventRecorder.Component.Operator,
consumer,
- "mk"));
-
- event =
+ "mk",
+ interval,
+ dedupePredicate,
+ labels));
+ var event =
kubernetesClient
.v1()
.events()
@@ -249,42 +280,34 @@ public void accept(Event event) {
.withName(eventName)
.get();
Assertions.assertNotNull(event);
- Assertions.assertEquals("message2", event.getMessage());
- Assertions.assertEquals(2, event.getCount());
- }
-
- @Test
- public void testCreateByIntervalWithMessageKey() {
- var consumer =
- new Consumer() {
- @Override
- public void accept(Event event) {
- eventConsumed = event;
- }
- };
- var flinkApp = TestUtils.buildApplicationCluster();
- var reason = "Cleanup";
- var eventName =
- EventUtils.generateEventName(
- flinkApp,
- EventRecorder.Type.Warning,
- reason,
- "mk",
- EventRecorder.Component.Operator);
+ if ((dedupePredicate == null || (dedupePredicate.test(labels)))
+ && interval != null
+ && interval.toMillis() > 0) {
+ Assertions.assertEquals("message", event.getMessage());
+ Assertions.assertEquals(1, event.getCount());
+ Assertions.assertEquals(null, event.getMetadata().getLabels().get("a"));
+ } else {
+ Assertions.assertEquals("message1", event.getMessage());
+ Assertions.assertEquals(2, event.getCount());
+ Assertions.assertEquals("b", event.getMetadata().getLabels().get("a"));
+ }
- eventConsumed = null;
- Assertions.assertTrue(
- EventUtils.createByInterval(
+ // Update with duplicate labels.
+ Assertions.assertFalse(
+ EventUtils.createOrUpdateEventWithLabels(
kubernetesClient,
flinkApp,
EventRecorder.Type.Warning,
reason,
- "message1",
+ "message2",
EventRecorder.Component.Operator,
consumer,
"mk",
- Duration.ofSeconds(1800)));
- var event =
+ interval,
+ dedupePredicate,
+ labels));
+
+ event =
kubernetesClient
.v1()
.events()
@@ -292,23 +315,33 @@ public void accept(Event event) {
.withName(eventName)
.get();
Assertions.assertNotNull(event);
- Assertions.assertEquals(eventConsumed, event);
- Assertions.assertEquals("message1", event.getMessage());
- Assertions.assertEquals(1, event.getCount());
+ if ((dedupePredicate == null || (dedupePredicate.test(labels)))
+ && interval != null
+ && interval.toMillis() > 0) {
+ Assertions.assertEquals("message", event.getMessage());
+ Assertions.assertEquals(1, event.getCount());
+ Assertions.assertEquals(null, event.getMetadata().getLabels().get("a"));
+ } else {
+ Assertions.assertEquals("message2", event.getMessage());
+ Assertions.assertEquals(3, event.getCount());
+ Assertions.assertEquals("b", event.getMetadata().getLabels().get("a"));
+ }
- eventConsumed = null;
+ // Update with empty label.
+ labels = Map.of();
Assertions.assertFalse(
- EventUtils.createByInterval(
+ EventUtils.createOrUpdateEventWithLabels(
kubernetesClient,
flinkApp,
EventRecorder.Type.Warning,
reason,
- "message2",
+ "message3",
EventRecorder.Component.Operator,
consumer,
"mk",
- Duration.ofSeconds(1800)));
-
+ interval,
+ dedupePredicate,
+ labels));
event =
kubernetesClient
.v1()
@@ -317,22 +350,32 @@ public void accept(Event event) {
.withName(eventName)
.get();
Assertions.assertNotNull(event);
- Assertions.assertEquals(eventConsumed, event);
- Assertions.assertEquals("message2", event.getMessage());
- Assertions.assertEquals(2, event.getCount());
+ if ((dedupePredicate == null || (dedupePredicate.test(labels)))
+ && interval != null
+ && interval.toMillis() > 0) {
+ Assertions.assertEquals("message", event.getMessage());
+ Assertions.assertEquals(1, event.getCount());
+ Assertions.assertEquals(null, event.getMetadata().getLabels().get("a"));
+ } else {
+ Assertions.assertEquals("message3", event.getMessage());
+ Assertions.assertEquals(4, event.getCount());
+ Assertions.assertEquals(null, event.getMetadata().getLabels().get("a"));
+ }
- eventConsumed = null;
+ // Update the event with null label
Assertions.assertFalse(
- EventUtils.createByInterval(
+ EventUtils.createOrUpdateEventWithLabels(
kubernetesClient,
flinkApp,
EventRecorder.Type.Warning,
reason,
- "message2",
+ "message4",
EventRecorder.Component.Operator,
consumer,
"mk",
- Duration.ofSeconds(1800)));
+ interval,
+ dedupePredicate,
+ null));
event =
kubernetesClient
@@ -342,20 +385,38 @@ public void accept(Event event) {
.withName(eventName)
.get();
Assertions.assertNotNull(event);
- Assertions.assertNull(eventConsumed);
-
- eventConsumed = null;
+ if ((dedupePredicate == null || (dedupePredicate.test(labels)))
+ && interval != null
+ && interval.toMillis() > 0) {
+ Assertions.assertEquals("message", event.getMessage());
+ Assertions.assertEquals(1, event.getCount());
+ Assertions.assertEquals(null, event.getMetadata().getLabels().get("a"));
+ } else {
+ Assertions.assertEquals("message4", event.getMessage());
+ Assertions.assertEquals(
+ dedupePredicate != null
+ && dedupePredicate.test(labels)
+ && interval != null
+ && interval.toMillis() > 0
+ ? 4
+ : 5,
+ event.getCount());
+ Assertions.assertEquals(null, event.getMetadata().getLabels().get("a"));
+ }
+ // Create a new event
Assertions.assertTrue(
- EventUtils.createByInterval(
+ EventUtils.createOrUpdateEventWithLabels(
kubernetesClient,
flinkApp,
EventRecorder.Type.Warning,
reason,
- "message2",
+ "message1",
EventRecorder.Component.Operator,
consumer,
"mk2",
- Duration.ofSeconds(1800)));
+ interval,
+ dedupePredicate,
+ Map.of("a", "b")));
eventName =
EventUtils.generateEventName(
flinkApp,
@@ -371,9 +432,9 @@ public void accept(Event event) {
.withName(eventName)
.get();
Assertions.assertNotNull(event);
- Assertions.assertEquals(eventConsumed, event);
- Assertions.assertEquals("message2", event.getMessage());
+ Assertions.assertEquals("message1", event.getMessage());
Assertions.assertEquals(1, event.getCount());
+ Assertions.assertEquals(event.getMetadata().getLabels().get("a"), "b");
}
@Test