This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new 43719c98b [FLINK-40991][operator] Ignore unknown fields when reading
ClusterHealthInfo (#1221)
43719c98b is described below
commit 43719c98ba05334ff8809965fe9114cdb5c0c7bb
Author: laughingman7743 <[email protected]>
AuthorDate: Sat Oct 31 00:19:59 2026 +0900
[FLINK-40991][operator] Ignore unknown fields when reading
ClusterHealthInfo (#1221)
---
.../operator/health/ClusterHealthInfo.java | 2 ++
.../operator/health/ClusterHealthInfoTest.java | 18 ++++++++++++++
.../observer/ClusterHealthEvaluatorTest.java | 29 ++++++++++++++++++++++
3 files changed, 49 insertions(+)
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/health/ClusterHealthInfo.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/health/ClusterHealthInfo.java
index d215e87c6..b894f7bbf 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/health/ClusterHealthInfo.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/health/ClusterHealthInfo.java
@@ -20,6 +20,7 @@ package org.apache.flink.kubernetes.operator.health;
import org.apache.flink.annotation.Experimental;
import org.apache.flink.kubernetes.operator.observer.ClusterHealthResult;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.AllArgsConstructor;
@@ -31,6 +32,7 @@ import java.time.Clock;
@Experimental
@Data
@AllArgsConstructor
+@JsonIgnoreProperties(ignoreUnknown = true)
public class ClusterHealthInfo {
/** Millisecond timestamp of the last observed health information. */
private long timeStamp;
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/health/ClusterHealthInfoTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/health/ClusterHealthInfoTest.java
index 44cfef2e5..0bb57c910 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/health/ClusterHealthInfoTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/health/ClusterHealthInfoTest.java
@@ -25,6 +25,7 @@ import java.time.Clock;
import java.time.ZoneId;
import static java.time.Instant.ofEpochSecond;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -54,6 +55,23 @@ class ClusterHealthInfoTest {
assertEquals("test-error",
clusterHealthInfoFromJson.getHealthResult().getError());
}
+ @Test
+ public void deserializeShouldIgnoreLegacyHealthyField() {
+ final var clusterHealthInfoJson =
+
"{\"timeStamp\":1,\"numRestarts\":2,\"numRestartsEvaluationTimeStamp\":3,"
+ + "\"numCompletedCheckpoints\":4,"
+ +
"\"numCompletedCheckpointsIncreasedTimeStamp\":5,\"healthy\":false}";
+ final var clusterHealthInfoFromJson =
ClusterHealthInfo.deserialize(clusterHealthInfoJson);
+ assertThat(clusterHealthInfoFromJson.getTimeStamp()).isEqualTo(1);
+ assertThat(clusterHealthInfoFromJson.getNumRestarts()).isEqualTo(2);
+
assertThat(clusterHealthInfoFromJson.getNumRestartsEvaluationTimeStamp()).isEqualTo(3);
+
assertThat(clusterHealthInfoFromJson.getNumCompletedCheckpoints()).isEqualTo(4);
+
assertThat(clusterHealthInfoFromJson.getNumCompletedCheckpointsIncreasedTimeStamp())
+ .isEqualTo(5);
+ assertThat(clusterHealthInfoFromJson.getHealthResult())
+ .isEqualTo(ClusterHealthResult.healthy());
+ }
+
@Test
public void serializationRoundTrip() {
var clock = Clock.fixed(ofEpochSecond(1), ZoneId.systemDefault());
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/ClusterHealthEvaluatorTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/ClusterHealthEvaluatorTest.java
index ee7c184a2..6c19800cc 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/ClusterHealthEvaluatorTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/ClusterHealthEvaluatorTest.java
@@ -263,6 +263,35 @@ class ClusterHealthEvaluatorTest {
assertClusterHealthIs(false);
}
+ @ParameterizedTest
+ @MethodSource("provideParametersEvaluateLegacyClusterHealthInfo")
+ public void evaluateShouldReevaluateLegacyClusterHealthInfo(
+ boolean legacyHealthy, int observedNumRestarts, boolean
expectedHealthy) {
+ configuration.set(OPERATOR_CLUSTER_HEALTH_CHECK_RESTARTS_WINDOW,
Duration.ofMinutes(5));
+ configuration.set(OPERATOR_CLUSTER_HEALTH_CHECK_RESTARTS_THRESHOLD,
100);
+ final var timeStamp = validInstant1.toEpochMilli();
+ clusterInfo.put(
+ ClusterHealthInfo.class.getSimpleName(),
+ String.format(
+
"{\"timeStamp\":%d,\"numRestarts\":0,\"numRestartsEvaluationTimeStamp\":%d,"
+ + "\"numCompletedCheckpoints\":1,"
+ +
"\"numCompletedCheckpointsIncreasedTimeStamp\":%d,"
+ + "\"healthy\":%b}",
+ timeStamp, timeStamp, timeStamp, legacyHealthy));
+ final var observedClusterHealthInfo =
+ createClusterHealthInfo(validInstant2, observedNumRestarts, 1);
+
+ clusterHealthEvaluator.evaluate(configuration, clusterInfo,
observedClusterHealthInfo);
+ assertClusterHealthIs(expectedHealthy);
+ }
+
+ private static Stream<Arguments>
provideParametersEvaluateLegacyClusterHealthInfo() {
+ return Stream.of(
+ Arguments.of(false, 0, true),
+ Arguments.of(true, 101, false),
+ Arguments.of(false, 101, false));
+ }
+
@Test
public void
evaluateShouldMarkClusterHealthyWhenNoCompletedCheckpointsInsideWindow() {
configuration.set(OPERATOR_CLUSTER_HEALTH_CHECK_CHECKPOINT_PROGRESS_ENABLED,
true);
