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 f5456a4c8 [FLINK-40857] Fix StatusRecorder not breaking out of for loop
f5456a4c8 is described below
commit f5456a4c8d59e28d89039d87ffc6c111c5445951
Author: Mate Czagany <[email protected]>
AuthorDate: Thu Oct 1 13:10:23 2026 +0200
[FLINK-40857] Fix StatusRecorder not breaking out of for loop
---
.../kubernetes/operator/utils/StatusRecorder.java | 45 ++++++++++++++--------
.../operator/utils/StatusRecorderTest.java | 18 ++++++++-
2 files changed, 45 insertions(+), 18 deletions(-)
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/StatusRecorder.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/StatusRecorder.java
index c13280423..428f78f38 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/StatusRecorder.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/StatusRecorder.java
@@ -56,6 +56,8 @@ public class StatusRecorder<CR extends CustomResource<?,
STATUS>, STATUS> {
private static final Logger LOG =
LoggerFactory.getLogger(StatusRecorder.class);
+ private static final int MAX_STATUS_UPDATE_ATTEMPTS = 3;
+
protected final ObjectMapper objectMapper = new ObjectMapper();
protected final ConcurrentHashMap<ResourceID, ObjectNode> statusCache =
@@ -102,23 +104,7 @@ public class StatusRecorder<CR extends CustomResource<?,
STATUS>, STATUS> {
}
var prevStatus = convertPreviousStatus(resource, previousStatusNode);
-
- Exception err = null;
- for (int i = 0; i < 3; i++) {
- // We retry the status update 3 times to avoid some intermittent
connectivity errors
- try {
- replaceStatus(resource, prevStatus, client);
- err = null;
- } catch (KubernetesClientException e) {
- LOG.error("Error while patching status, retrying {}/3...", (i
+ 1), e);
- Thread.sleep(1000);
- err = e;
- }
- }
-
- if (err != null) {
- throw err;
- }
+ replaceStatusWithRetries(resource, prevStatus, client);
statusCache.put(resourceId, newStatusNode);
statusUpdateListener.accept(resource, prevStatus);
@@ -142,6 +128,31 @@ public class StatusRecorder<CR extends CustomResource<?,
STATUS>, STATUS> {
return (STATUS) objectMapper.convertValue(previousStatusNode,
statusClass);
}
+ private void replaceStatusWithRetries(CR resource, STATUS prevStatus,
KubernetesClient client)
+ throws JsonProcessingException, InterruptedException {
+ // We retry the status update to avoid some intermittent connectivity
errors
+ for (int attempt = 1; ; attempt++) {
+ try {
+ replaceStatus(resource, prevStatus, client);
+ return;
+ } catch (KubernetesClientException e) {
+ if (attempt == MAX_STATUS_UPDATE_ATTEMPTS) {
+ LOG.error(
+ "Error while patching status, giving up after {}
attempts",
+ MAX_STATUS_UPDATE_ATTEMPTS,
+ e);
+ throw e;
+ }
+ LOG.error(
+ "Error while patching status, retrying {}/{}...",
+ attempt,
+ MAX_STATUS_UPDATE_ATTEMPTS,
+ e);
+ Thread.sleep(1000);
+ }
+ }
+ }
+
private void replaceStatus(CR resource, STATUS prevStatus,
KubernetesClient client)
throws JsonProcessingException {
int retries = 0;
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/StatusRecorderTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/StatusRecorderTest.java
index 6f9b2e52e..db28c9f98 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/StatusRecorderTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/StatusRecorderTest.java
@@ -45,7 +45,7 @@ public class StatusRecorderTest {
new StatusRecorder<FlinkDeployment, FlinkDeploymentStatus>(
new MetricManager<>(), (e, s) -> {});
var deployment = TestUtils.buildApplicationCluster();
- kubernetesClient.resource(deployment).createOrReplace();
+ kubernetesClient.resource(deployment).create();
var lastRequest = mockServer.getLastRequest();
helper.patchAndCacheStatus(deployment, kubernetesClient);
@@ -63,6 +63,22 @@ public class StatusRecorderTest {
assertThat(lastRequest).isSameAs(mockServer.getLastRequest());
}
+ @Test
+ public void testSuccessfulPatchStopsRetrying() {
+ var statusRecorder =
+ new StatusRecorder<FlinkDeployment, FlinkDeploymentStatus>(
+ new MetricManager<>(), (e, s) -> {});
+ var deployment = TestUtils.buildApplicationCluster();
+ kubernetesClient.resource(deployment).create();
+ statusRecorder.patchAndCacheStatus(deployment, kubernetesClient);
+
deployment.getStatus().getReconciliationStatus().setState(ReconciliationState.ROLLING_BACK);
+ var requestCount = mockServer.getRequestCount();
+
+ statusRecorder.patchAndCacheStatus(deployment, kubernetesClient);
+
+ assertThat(mockServer.getRequestCount()).isEqualTo(requestCount + 1);
+ }
+
@Test
public void testNullLatestResource() {
var statusRecorder =
