You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "mxm (via GitHub)" <gi...@apache.org> on 2023/11/13 13:34:09 UTC
[PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
mxm opened a new pull request, #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710
The state store implementation should take care of optimizing flush
operations. That is already the case for the Kubernetes-based
implementation. The in-memory state store does not require flushing.
The current logic tries to save on API calls but there is no actual savings
because the ConfigMap needs to be retrieved at least once to check whether its
empty. So instead of checking for emptiness, we can simply clear it.
Further, I've removed the optimization to keep some of the data when disabling
the autoscaler. The metric collection contract requires that the entire metric
window contains data. By removing some of the data, once turned on again, we
could add metric gaps to the metric history if there are still metrics left from
the last metric window which overlap with the current. The reason is that the
metric window is counted from the oldest timestamp found in the metric history.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1392789442
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobAutoScalerImpl.java:
##########
@@ -121,37 +120,10 @@ public void cleanup(KEY jobKey) {
}
private void clearStatesAfterAutoscalerDisabled(Context ctx) throws Exception {
- var needFlush = false;
- var parallelismOverrides = stateStore.getParallelismOverrides(ctx);
- if (!parallelismOverrides.isEmpty()) {
- needFlush = true;
- stateStore.removeParallelismOverrides(ctx);
- }
-
- var collectedMetrics = stateStore.getCollectedMetrics(ctx);
- if (!collectedMetrics.isEmpty()) {
- needFlush = true;
- stateStore.removeCollectedMetrics(ctx);
- }
-
- var scalingHistory = stateStore.getScalingHistory(ctx);
- if (!scalingHistory.isEmpty()) {
- var trimmedScalingHistory =
- trimScalingHistory(clock.instant(), ctx.getConfiguration(), scalingHistory);
- if (trimmedScalingHistory.isEmpty()) {
- // All scaling histories are trimmed.
- needFlush = true;
- stateStore.removeScalingHistory(ctx);
- } else if (!scalingHistory.equals(trimmedScalingHistory)) {
- // Some scaling histories are trimmed.
- needFlush = true;
- stateStore.storeScalingHistory(ctx, trimmedScalingHistory);
- }
- }
-
- if (needFlush) {
- stateStore.flush(ctx);
- }
+ stateStore.removeParallelismOverrides(ctx);
+ stateStore.removeCollectedMetrics(ctx);
+ stateStore.removeScalingHistory(ctx);
+ stateStore.flush(ctx);
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1391538017
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ConfigMapStore.java:
##########
@@ -67,19 +67,14 @@ protected Optional<String> getSerializedState(
}
protected void removeSerializedState(KubernetesJobAutoScalerContext jobContext, String key) {
- getConfigMap(jobContext)
- .ifPresentOrElse(
- configMap -> configMap.getData().remove(key),
- () -> {
- throw new IllegalStateException(
- "The configMap isn't created, so the remove is unavailable.");
- });
+ getConfigMap(jobContext).ifPresent(configMap -> configMap.getData().remove(key));
}
public void flush(KubernetesJobAutoScalerContext jobContext) {
Optional<ConfigMap> configMapOpt = cache.get(jobContext.getJobKey());
if (configMapOpt == null || configMapOpt.isEmpty()) {
LOG.debug("The configMap isn't updated, so skip the flush.");
+ // Do not flush if there are no updates.
Review Comment:
Thanks for adding the tests. You are right that the retrieval will always go to Kubernetes. The current implementation has some bugs with the bookkeeping to minimize the number of API calls. It does make more requests than necessary.
I've rewrote the implementation to minimize the number of times we go to Kubernetes. We only retrieve once now on lookup. If the ConfigMap does not exist, we will only create the ConfigMap once we flush back any changes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#issuecomment-1812440379
🚢
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1393033026
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapView.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * 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.state;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.util.Preconditions;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.client.dsl.Resource;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+class ConfigMapView {
+
+ enum State {
+ NEEDS_CREATE,
+ NEEDS_UPDATE,
+ UP_TO_DATE
+ }
+
+ private State state;
+
+ private ConfigMap configMap;
+
+ private final Function<ConfigMap, Resource<ConfigMap>> resourceRetriever;
+
+ public ConfigMapView(
+ Supplier<ConfigMap> configMapRetriever,
+ Supplier<ConfigMap> configMapBuilder,
+ Function<ConfigMap, Resource<ConfigMap>> resourceRetriever) {
+ var existingConfigMap = configMapRetriever.get();
+ if (existingConfigMap != null) {
+ refreshConfigMap(existingConfigMap);
+ } else {
+ this.configMap = configMapBuilder.get();
+ this.state = State.NEEDS_CREATE;
+ }
+ this.resourceRetriever = resourceRetriever;
Review Comment:
The current state is the following btw:
```
Function<ConfigMap, Resource<ConfigMap>> resource:
ConfigMap configMap;
State state;
```
That occupies less space than what you are suggesting. Resource is only a pointer to an existing method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1391256838
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ConfigMapStore.java:
##########
@@ -67,19 +67,14 @@ protected Optional<String> getSerializedState(
}
protected void removeSerializedState(KubernetesJobAutoScalerContext jobContext, String key) {
- getConfigMap(jobContext)
- .ifPresentOrElse(
- configMap -> configMap.getData().remove(key),
- () -> {
- throw new IllegalStateException(
- "The configMap isn't created, so the remove is unavailable.");
- });
+ getConfigMap(jobContext).ifPresent(configMap -> configMap.getData().remove(key));
}
public void flush(KubernetesJobAutoScalerContext jobContext) {
Optional<ConfigMap> configMapOpt = cache.get(jobContext.getJobKey());
if (configMapOpt == null || configMapOpt.isEmpty()) {
LOG.debug("The configMap isn't updated, so skip the flush.");
+ // Do not flush if there are no updates.
Review Comment:
We may need to added an extra cache to store the state that last updated. When the flush is called, then compare them. And call kubernetes if they are not same.
And it's better to add a test to check the flush doesn't call kubernetes if state isn't updated.
Also, I'm not sure whether this PR is reasonable. After this refactor, the clean up logic is indeed clear. However, these complex operations are moved into state store. When we implememt other persistent state stores in the future, we still need to re-implememt this similar logic again.
If this logic can be retained in autoscaler, only one code is fine.
Please correct me if I miss some information, thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1393008888
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapView.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * 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.state;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.util.Preconditions;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.client.dsl.Resource;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+class ConfigMapView {
+
+ enum State {
+ NEEDS_CREATE,
+ NEEDS_UPDATE,
+ UP_TO_DATE
+ }
+
+ private State state;
+
+ private ConfigMap configMap;
+
+ private final Function<ConfigMap, Resource<ConfigMap>> resourceRetriever;
+
+ public ConfigMapView(
+ Supplier<ConfigMap> configMapRetriever,
+ Supplier<ConfigMap> configMapBuilder,
+ Function<ConfigMap, Resource<ConfigMap>> resourceRetriever) {
+ var existingConfigMap = configMapRetriever.get();
+ if (existingConfigMap != null) {
+ refreshConfigMap(existingConfigMap);
+ } else {
+ this.configMap = configMapBuilder.get();
+ this.state = State.NEEDS_CREATE;
+ }
+ this.resourceRetriever = resourceRetriever;
Review Comment:
I think we could further simplify this by keeping only:
```
Resource<ConfigMap> resource:
ConfigMap configMap;
State state;
```
And pass those directly in the constructor / through a static create method instead of the Suppliers/Functions which feel a bit unnecessary
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapView.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * 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.state;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.util.Preconditions;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.client.dsl.Resource;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+class ConfigMapView {
+
+ enum State {
+ NEEDS_CREATE,
+ NEEDS_UPDATE,
+ UP_TO_DATE
+ }
+
+ private State state;
+
+ private ConfigMap configMap;
+
+ private final Function<ConfigMap, Resource<ConfigMap>> resourceRetriever;
+
+ public ConfigMapView(
+ Supplier<ConfigMap> configMapRetriever,
+ Supplier<ConfigMap> configMapBuilder,
+ Function<ConfigMap, Resource<ConfigMap>> resourceRetriever) {
+ var existingConfigMap = configMapRetriever.get();
+ if (existingConfigMap != null) {
+ refreshConfigMap(existingConfigMap);
+ } else {
+ this.configMap = configMapBuilder.get();
+ this.state = State.NEEDS_CREATE;
Review Comment:
I am not super familiar with the `Resource<T>` class but couldn't we use it also to get the `ConfigMap`? Then we could simply use that and don't need the `configMapRetriever`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1393023182
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapView.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * 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.state;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.util.Preconditions;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.client.dsl.Resource;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+class ConfigMapView {
+
+ enum State {
+ NEEDS_CREATE,
+ NEEDS_UPDATE,
+ UP_TO_DATE
+ }
+
+ private State state;
+
+ private ConfigMap configMap;
+
+ private final Function<ConfigMap, Resource<ConfigMap>> resourceRetriever;
+
+ public ConfigMapView(
+ Supplier<ConfigMap> configMapRetriever,
+ Supplier<ConfigMap> configMapBuilder,
+ Function<ConfigMap, Resource<ConfigMap>> resourceRetriever) {
+ var existingConfigMap = configMapRetriever.get();
+ if (existingConfigMap != null) {
+ refreshConfigMap(existingConfigMap);
+ } else {
+ this.configMap = configMapBuilder.get();
+ this.state = State.NEEDS_CREATE;
Review Comment:
Lookup is performed by name and namespace. Update / Create uses the ConfigMap object which potentially involves version checks. That's how the code currently works. I didn't want to mess with the APIs to avoid introducing regressions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1394045676
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapStore.java:
##########
@@ -99,49 +93,30 @@ public void removeInfoFromCache(ResourceID resourceID) {
cache.remove(resourceID);
}
- private Optional<ConfigMap> getConfigMap(KubernetesJobAutoScalerContext jobContext) {
+ private ConfigMapView getConfigMap(KubernetesJobAutoScalerContext jobContext) {
return cache.computeIfAbsent(
jobContext.getJobKey(), (id) -> getConfigMapFromKubernetes(jobContext));
}
- private Map<String, String> getOrCreateState(KubernetesJobAutoScalerContext jobContext) {
+ private ConfigMapView getOrCreateState(KubernetesJobAutoScalerContext jobContext) {
return cache.compute(
- jobContext.getJobKey(),
- (id, configMapOpt) -> {
- // If in the cache and valid simply return
- if (configMapOpt != null && configMapOpt.isPresent()) {
- return configMapOpt;
- }
- // Otherwise get or create
- return Optional.of(getOrCreateConfigMapFromKubernetes(jobContext));
- })
- .get()
- .getData();
+ jobContext.getJobKey(),
+ (id, configMapView) -> {
+ // If in the cache and valid simply return
+ if (configMapView != null) {
+ return configMapView;
+ }
+ // Otherwise retrieve if it exists
+ return getConfigMapFromKubernetes(jobContext);
+ });
}
@VisibleForTesting
- protected Optional<ConfigMap> getConfigMapFromKubernetes(
- KubernetesJobAutoScalerContext jobContext) {
+ ConfigMapView getConfigMapFromKubernetes(KubernetesJobAutoScalerContext jobContext) {
Review Comment:
```suggestion
ConfigMapView createConfigMapView(KubernetesJobAutoScalerContext jobContext) {
```
nit
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapView.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * 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.state;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.util.Preconditions;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.client.dsl.Resource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.function.Function;
+
+class ConfigMapView {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ConfigMapView.class);
+
+ enum State {
+ /** ConfigMap is only stored locally, not created in Kubernetes yet. */
+ NEEDS_CREATE,
+ /** ConfigMap exists in Kubernetes but there are newer local changes. */
+ NEEDS_UPDATE,
+ /** ConfigMap view reflects the actual contents of Kubernetes ConfigMap. */
+ UP_TO_DATE
+ }
+
+ private State state;
+
+ private ConfigMap configMap;
+
+ private final Function<ConfigMap, Resource<ConfigMap>> resourceRetriever;
+
+ public ConfigMapView(
+ ConfigMap configMap, Function<ConfigMap, Resource<ConfigMap>> resourceRetriever) {
Review Comment:
```suggestion
ConfigMap newConfigMap, Function<ConfigMap, Resource<ConfigMap>> resourceRetriever) {
```
How about rename the configMap to `newConfigMap` or `emptyConfigMap`? It's clear to let us know it's only used when `resourceRetriever.get()` is null.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapStore.java:
##########
@@ -99,49 +93,30 @@ public void removeInfoFromCache(ResourceID resourceID) {
cache.remove(resourceID);
}
- private Optional<ConfigMap> getConfigMap(KubernetesJobAutoScalerContext jobContext) {
+ private ConfigMapView getConfigMap(KubernetesJobAutoScalerContext jobContext) {
return cache.computeIfAbsent(
jobContext.getJobKey(), (id) -> getConfigMapFromKubernetes(jobContext));
}
- private Map<String, String> getOrCreateState(KubernetesJobAutoScalerContext jobContext) {
+ private ConfigMapView getOrCreateState(KubernetesJobAutoScalerContext jobContext) {
return cache.compute(
- jobContext.getJobKey(),
- (id, configMapOpt) -> {
- // If in the cache and valid simply return
- if (configMapOpt != null && configMapOpt.isPresent()) {
- return configMapOpt;
- }
- // Otherwise get or create
- return Optional.of(getOrCreateConfigMapFromKubernetes(jobContext));
- })
- .get()
- .getData();
+ jobContext.getJobKey(),
+ (id, configMapView) -> {
+ // If in the cache and valid simply return
+ if (configMapView != null) {
+ return configMapView;
+ }
Review Comment:
nit: How about using the `computeIfAbsent` instead of `compute`? If so, we don't need to care about `configMapView != null`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1393549254
##########
flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/state/KubernetesAutoScalerStateStoreTest.java:
##########
@@ -330,4 +331,42 @@ public void testDiscardInvalidHistory() throws Exception {
ctx, KubernetesAutoScalerStateStore.SCALING_HISTORY_KEY))
.isEmpty();
}
+
+ @Test
+ public void testDiscardAllState() {
Review Comment:
Hi, this test is accessing `configMapStore` instead of `KubernetesAutoScalerStateStore`.
Should we test `KubernetesAutoScalerStateStore`? If we test `configMapStore`, it's better to move this test to `ConfigMapStoreTest`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1393995132
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapStore.java:
##########
@@ -44,13 +43,13 @@ public class ConfigMapStore {
private final KubernetesClient kubernetesClient;
- // The cache for each resourceId may be in three states:
- // 1. The resourceId doesn't exist : ConfigMap isn't loaded from kubernetes, or it's deleted
- // 2 Exists, Optional.empty() : The ConfigMap doesn't exist in Kubernetes
- // 3. Exists, Not Empty : We have loaded the ConfigMap from kubernetes, it may not be the same
- // if not flushed already
- private final ConcurrentHashMap<ResourceID, Optional<ConfigMap>> cache =
- new ConcurrentHashMap<>();
+ // The cache for each resourceId may be in four states:
+ // 1. No cache entry: ConfigMap isn't loaded from kubernetes, or it's deleted.
+ // 2. Cache entry, not created : The ConfigMap doesn't exist in Kubernetes.
+ // 3. Cache entry, not flushed : The ConfigMap exists in Kubernetes, but it is not updated yet.
+ // 4. Cache entry, flushed and created : We have loaded the ConfigMap from kubernetes, and it's
+ // up-to-date.
Review Comment:
Sure. Take a look at https://github.com/apache/flink-kubernetes-operator/pull/710/commits/e6568d958a141e77be54af6c456212faa69869b8. Does that reflect what you had in mind?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1392785211
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ConfigMapStore.java:
##########
@@ -44,13 +43,44 @@ public class ConfigMapStore {
private final KubernetesClient kubernetesClient;
- // The cache for each resourceId may be in three states:
- // 1. The resourceId doesn't exist : ConfigMap isn't loaded from kubernetes, or it's deleted
- // 2 Exists, Optional.empty() : The ConfigMap doesn't exist in Kubernetes
- // 3. Exists, Not Empty : We have loaded the ConfigMap from kubernetes, it may not be the same
- // if not flushed already
- private final ConcurrentHashMap<ResourceID, Optional<ConfigMap>> cache =
- new ConcurrentHashMap<>();
+ static class ConfigMapState {
Review Comment:
Actually now a top-level class.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1393033026
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapView.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * 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.state;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.util.Preconditions;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.client.dsl.Resource;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+class ConfigMapView {
+
+ enum State {
+ NEEDS_CREATE,
+ NEEDS_UPDATE,
+ UP_TO_DATE
+ }
+
+ private State state;
+
+ private ConfigMap configMap;
+
+ private final Function<ConfigMap, Resource<ConfigMap>> resourceRetriever;
+
+ public ConfigMapView(
+ Supplier<ConfigMap> configMapRetriever,
+ Supplier<ConfigMap> configMapBuilder,
+ Function<ConfigMap, Resource<ConfigMap>> resourceRetriever) {
+ var existingConfigMap = configMapRetriever.get();
+ if (existingConfigMap != null) {
+ refreshConfigMap(existingConfigMap);
+ } else {
+ this.configMap = configMapBuilder.get();
+ this.state = State.NEEDS_CREATE;
+ }
+ this.resourceRetriever = resourceRetriever;
Review Comment:
The current state is the following btw:
```
Function<ConfigMap, Resource<ConfigMap>> resource:
ConfigMap configMap;
State state;
```
That occupies less space than what you are suggesting. `resource` is only a pointer to an existing method. In your case it is a reference to a heap object.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1392788100
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ConfigMapStore.java:
##########
@@ -63,29 +93,29 @@ protected void putSerializedState(
protected Optional<String> getSerializedState(
KubernetesJobAutoScalerContext jobContext, String key) {
- return getConfigMap(jobContext).map(configMap -> configMap.getData().get(key));
+ return Optional.ofNullable(getConfigMap(jobContext).configMap.getData().get(key));
}
protected void removeSerializedState(KubernetesJobAutoScalerContext jobContext, String key) {
- getConfigMap(jobContext)
- .ifPresentOrElse(
- configMap -> configMap.getData().remove(key),
- () -> {
- throw new IllegalStateException(
- "The configMap isn't created, so the remove is unavailable.");
- });
+ getConfigMap(jobContext).removeKey(key);
}
public void flush(KubernetesJobAutoScalerContext jobContext) {
- Optional<ConfigMap> configMapOpt = cache.get(jobContext.getJobKey());
- if (configMapOpt == null || configMapOpt.isEmpty()) {
+ ConfigMapState configMapState = cache.get(jobContext.getJobKey());
+ if (configMapState == null || (configMapState.flushed && configMapState.exists)) {
Review Comment:
This should be nicer now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1393029595
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapView.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * 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.state;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.util.Preconditions;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.client.dsl.Resource;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+class ConfigMapView {
+
+ enum State {
+ NEEDS_CREATE,
+ NEEDS_UPDATE,
+ UP_TO_DATE
+ }
+
+ private State state;
+
+ private ConfigMap configMap;
+
+ private final Function<ConfigMap, Resource<ConfigMap>> resourceRetriever;
+
+ public ConfigMapView(
+ Supplier<ConfigMap> configMapRetriever,
+ Supplier<ConfigMap> configMapBuilder,
+ Function<ConfigMap, Resource<ConfigMap>> resourceRetriever) {
+ var existingConfigMap = configMapRetriever.get();
+ if (existingConfigMap != null) {
+ refreshConfigMap(existingConfigMap);
+ } else {
+ this.configMap = configMapBuilder.get();
+ this.state = State.NEEDS_CREATE;
+ }
+ this.resourceRetriever = resourceRetriever;
Review Comment:
That may be true. Let me try. I didn't want to change the retrieval / create / update logic.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1393545577
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ConfigMapStore.java:
##########
@@ -44,13 +43,44 @@ public class ConfigMapStore {
private final KubernetesClient kubernetesClient;
- // The cache for each resourceId may be in three states:
- // 1. The resourceId doesn't exist : ConfigMap isn't loaded from kubernetes, or it's deleted
- // 2 Exists, Optional.empty() : The ConfigMap doesn't exist in Kubernetes
- // 3. Exists, Not Empty : We have loaded the ConfigMap from kubernetes, it may not be the same
- // if not flushed already
- private final ConcurrentHashMap<ResourceID, Optional<ConfigMap>> cache =
- new ConcurrentHashMap<>();
+ static class ConfigMapState {
+ private boolean flushed = true;
+ private boolean exists = true;
+
+ @VisibleForTesting ConfigMap configMap;
Review Comment:
Great work, Max! 👍
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ConfigMapStore.java:
##########
@@ -44,13 +43,44 @@ public class ConfigMapStore {
private final KubernetesClient kubernetesClient;
- // The cache for each resourceId may be in three states:
- // 1. The resourceId doesn't exist : ConfigMap isn't loaded from kubernetes, or it's deleted
- // 2 Exists, Optional.empty() : The ConfigMap doesn't exist in Kubernetes
- // 3. Exists, Not Empty : We have loaded the ConfigMap from kubernetes, it may not be the same
- // if not flushed already
- private final ConcurrentHashMap<ResourceID, Optional<ConfigMap>> cache =
- new ConcurrentHashMap<>();
+ static class ConfigMapState {
+ private boolean flushed = true;
+ private boolean exists = true;
+
+ @VisibleForTesting ConfigMap configMap;
Review Comment:
Great work, Max! 👍
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapStore.java:
##########
@@ -44,13 +43,13 @@ public class ConfigMapStore {
private final KubernetesClient kubernetesClient;
- // The cache for each resourceId may be in three states:
- // 1. The resourceId doesn't exist : ConfigMap isn't loaded from kubernetes, or it's deleted
- // 2 Exists, Optional.empty() : The ConfigMap doesn't exist in Kubernetes
- // 3. Exists, Not Empty : We have loaded the ConfigMap from kubernetes, it may not be the same
- // if not flushed already
- private final ConcurrentHashMap<ResourceID, Optional<ConfigMap>> cache =
- new ConcurrentHashMap<>();
+ // The cache for each resourceId may be in four states:
+ // 1. No cache entry: ConfigMap isn't loaded from kubernetes, or it's deleted.
+ // 2. Cache entry, not created : The ConfigMap doesn't exist in Kubernetes.
+ // 3. Cache entry, not flushed : The ConfigMap exists in Kubernetes, but it is not updated yet.
+ // 4. Cache entry, flushed and created : We have loaded the ConfigMap from kubernetes, and it's
+ // up-to-date.
Review Comment:
Would you mind updating these comments based on the latest `ConfigMapView.State`? For example, using `NEEDS_CREATE` instead of `not created`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1391256838
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ConfigMapStore.java:
##########
@@ -67,19 +67,14 @@ protected Optional<String> getSerializedState(
}
protected void removeSerializedState(KubernetesJobAutoScalerContext jobContext, String key) {
- getConfigMap(jobContext)
- .ifPresentOrElse(
- configMap -> configMap.getData().remove(key),
- () -> {
- throw new IllegalStateException(
- "The configMap isn't created, so the remove is unavailable.");
- });
+ getConfigMap(jobContext).ifPresent(configMap -> configMap.getData().remove(key));
}
public void flush(KubernetesJobAutoScalerContext jobContext) {
Optional<ConfigMap> configMapOpt = cache.get(jobContext.getJobKey());
if (configMapOpt == null || configMapOpt.isEmpty()) {
LOG.debug("The configMap isn't updated, so skip the flush.");
+ // Do not flush if there are no updates.
Review Comment:
We may need to added a extra cache to store the state that last updated. When the flush is called, then compare them. And call kubernetes if they are not same.
And it's better to add a test to check the flush doesn't call kubernetes if state isn't updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1391193239
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ConfigMapStore.java:
##########
@@ -67,19 +67,14 @@ protected Optional<String> getSerializedState(
}
protected void removeSerializedState(KubernetesJobAutoScalerContext jobContext, String key) {
- getConfigMap(jobContext)
- .ifPresentOrElse(
- configMap -> configMap.getData().remove(key),
- () -> {
- throw new IllegalStateException(
- "The configMap isn't created, so the remove is unavailable.");
- });
+ getConfigMap(jobContext).ifPresent(configMap -> configMap.getData().remove(key));
}
public void flush(KubernetesJobAutoScalerContext jobContext) {
Optional<ConfigMap> configMapOpt = cache.get(jobContext.getJobKey());
if (configMapOpt == null || configMapOpt.isEmpty()) {
LOG.debug("The configMap isn't updated, so skip the flush.");
+ // Do not flush if there are no updates.
Review Comment:
I'm thinking whether any updating or removeing can be determined, please give me a few times to test it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1391497915
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ConfigMapStore.java:
##########
@@ -67,19 +67,14 @@ protected Optional<String> getSerializedState(
}
protected void removeSerializedState(KubernetesJobAutoScalerContext jobContext, String key) {
- getConfigMap(jobContext)
- .ifPresentOrElse(
- configMap -> configMap.getData().remove(key),
- () -> {
- throw new IllegalStateException(
- "The configMap isn't created, so the remove is unavailable.");
- });
+ getConfigMap(jobContext).ifPresent(configMap -> configMap.getData().remove(key));
}
public void flush(KubernetesJobAutoScalerContext jobContext) {
Optional<ConfigMap> configMapOpt = cache.get(jobContext.getJobKey());
if (configMapOpt == null || configMapOpt.isEmpty()) {
LOG.debug("The configMap isn't updated, so skip the flush.");
+ // Do not flush if there are no updates.
Review Comment:
In this case the flush only calls Kubernetes because there is still data in the state store that wasn't removed. Maybe we should add a `clearAll` method to the state store interface and simply call that with the contract that subsequent `clearAll()` and `flush()` actions should be idempotent and ideally should not call Kubernetes again and again
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1392784565
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ConfigMapStore.java:
##########
@@ -44,13 +43,44 @@ public class ConfigMapStore {
private final KubernetesClient kubernetesClient;
- // The cache for each resourceId may be in three states:
- // 1. The resourceId doesn't exist : ConfigMap isn't loaded from kubernetes, or it's deleted
- // 2 Exists, Optional.empty() : The ConfigMap doesn't exist in Kubernetes
- // 3. Exists, Not Empty : We have loaded the ConfigMap from kubernetes, it may not be the same
- // if not flushed already
- private final ConcurrentHashMap<ResourceID, Optional<ConfigMap>> cache =
- new ConcurrentHashMap<>();
+ static class ConfigMapState {
Review Comment:
Sure, moved.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1394011445
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapStore.java:
##########
@@ -44,13 +43,13 @@ public class ConfigMapStore {
private final KubernetesClient kubernetesClient;
- // The cache for each resourceId may be in three states:
- // 1. The resourceId doesn't exist : ConfigMap isn't loaded from kubernetes, or it's deleted
- // 2 Exists, Optional.empty() : The ConfigMap doesn't exist in Kubernetes
- // 3. Exists, Not Empty : We have loaded the ConfigMap from kubernetes, it may not be the same
- // if not flushed already
- private final ConcurrentHashMap<ResourceID, Optional<ConfigMap>> cache =
- new ConcurrentHashMap<>();
+ // The cache for each resourceId may be in four states:
+ // 1. No cache entry: ConfigMap isn't loaded from kubernetes, or it's deleted.
+ // 2. Cache entry, not created : The ConfigMap doesn't exist in Kubernetes.
+ // 3. Cache entry, not flushed : The ConfigMap exists in Kubernetes, but it is not updated yet.
+ // 4. Cache entry, flushed and created : We have loaded the ConfigMap from kubernetes, and it's
+ // up-to-date.
Review Comment:
Thanks, it's fine for me.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1391694357
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ConfigMapStore.java:
##########
@@ -44,13 +43,44 @@ public class ConfigMapStore {
private final KubernetesClient kubernetesClient;
- // The cache for each resourceId may be in three states:
- // 1. The resourceId doesn't exist : ConfigMap isn't loaded from kubernetes, or it's deleted
- // 2 Exists, Optional.empty() : The ConfigMap doesn't exist in Kubernetes
- // 3. Exists, Not Empty : We have loaded the ConfigMap from kubernetes, it may not be the same
- // if not flushed already
- private final ConcurrentHashMap<ResourceID, Optional<ConfigMap>> cache =
- new ConcurrentHashMap<>();
+ static class ConfigMapState {
Review Comment:
Could we please move this class declaration to the end of the enclosing class?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#issuecomment-1808809666
Please have a look at the new implementation. Also see: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1391538017
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1393033026
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapView.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * 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.state;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.util.Preconditions;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.client.dsl.Resource;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+class ConfigMapView {
+
+ enum State {
+ NEEDS_CREATE,
+ NEEDS_UPDATE,
+ UP_TO_DATE
+ }
+
+ private State state;
+
+ private ConfigMap configMap;
+
+ private final Function<ConfigMap, Resource<ConfigMap>> resourceRetriever;
+
+ public ConfigMapView(
+ Supplier<ConfigMap> configMapRetriever,
+ Supplier<ConfigMap> configMapBuilder,
+ Function<ConfigMap, Resource<ConfigMap>> resourceRetriever) {
+ var existingConfigMap = configMapRetriever.get();
+ if (existingConfigMap != null) {
+ refreshConfigMap(existingConfigMap);
+ } else {
+ this.configMap = configMapBuilder.get();
+ this.state = State.NEEDS_CREATE;
+ }
+ this.resourceRetriever = resourceRetriever;
Review Comment:
The current state is the following btw:
```
Function<ConfigMap, Resource<ConfigMap>> resource:
ConfigMap configMap;
State state;
```
That occupies less space than what you are suggesting. `resource` is only a pointer to an existing method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1394117281
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapView.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * 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.state;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.util.Preconditions;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.client.dsl.Resource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.function.Function;
+
+class ConfigMapView {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ConfigMapView.class);
+
+ enum State {
+ /** ConfigMap is only stored locally, not created in Kubernetes yet. */
+ NEEDS_CREATE,
+ /** ConfigMap exists in Kubernetes but there are newer local changes. */
+ NEEDS_UPDATE,
+ /** ConfigMap view reflects the actual contents of Kubernetes ConfigMap. */
+ UP_TO_DATE
+ }
+
+ private State state;
+
+ private ConfigMap configMap;
+
+ private final Function<ConfigMap, Resource<ConfigMap>> resourceRetriever;
+
+ public ConfigMapView(
+ ConfigMap configMap, Function<ConfigMap, Resource<ConfigMap>> resourceRetriever) {
Review Comment:
Renamed to `configMapSkeleton`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#issuecomment-1812434659
Thanks @mxm for the cleaning up and simplifying, great work!
LGTM for now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1391225899
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ConfigMapStore.java:
##########
@@ -67,19 +67,14 @@ protected Optional<String> getSerializedState(
}
protected void removeSerializedState(KubernetesJobAutoScalerContext jobContext, String key) {
- getConfigMap(jobContext)
- .ifPresentOrElse(
- configMap -> configMap.getData().remove(key),
- () -> {
- throw new IllegalStateException(
- "The configMap isn't created, so the remove is unavailable.");
- });
+ getConfigMap(jobContext).ifPresent(configMap -> configMap.getData().remove(key));
}
public void flush(KubernetesJobAutoScalerContext jobContext) {
Optional<ConfigMap> configMapOpt = cache.get(jobContext.getJobKey());
if (configMapOpt == null || configMapOpt.isEmpty()) {
LOG.debug("The configMap isn't updated, so skip the flush.");
+ // Do not flush if there are no updates.
Review Comment:
I write 2 tests:
1. one is `testRemoveAndFlushEveryTime`, it's similar to current PR code
2. one is `testQueryAndFlushIfNeeded`, it's similar to master code
- `testRemoveAndFlushEveryTime` will remove and flush every time, and it will call kubernetes every time.
- `testQueryAndFlushIfNeeded` will query from cache, and value is null, so remove and flush isn't necessary. So it doens't call kubernetes.
My question is: the extra call is expected? And I think this PR is fine if it doesn't introuduce any extra opration. Please correct my if my understanding is wrong, thanks a lot.
These 2 tests can be added to ConfigMapStoreTest, and they can run directly.
```
@Test
void testRemoveAndFlushEveryTime() {
KubernetesJobAutoScalerContext ctx = createContext("cr1", kubernetesClient);
var key = "key";
var value = "value";
var configMapStore = new ConfigMapStore(kubernetesClient);
configMapStore.putSerializedState(ctx, key, value);
assertEquals(2, mockWebServer.getRequestCount());
configMapStore.flush(ctx);
assertEquals(3, mockWebServer.getRequestCount());
// Get from cache
assertThat(configMapStore.getSerializedState(ctx, key)).hasValue(value);
assertEquals(3, mockWebServer.getRequestCount());
// Get from kubernetes
configMapStore = new ConfigMapStore(kubernetesClient);
assertThat(configMapStore.getSerializedState(ctx, key)).hasValue(value);
assertEquals(4, mockWebServer.getRequestCount());
// Remove the key
configMapStore.removeSerializedState(ctx, key);
assertEquals(4, mockWebServer.getRequestCount());
configMapStore.flush(ctx);
assertEquals(5, mockWebServer.getRequestCount());
// Every flush will call kubernetes.
configMapStore.flush(ctx);
assertEquals(6, mockWebServer.getRequestCount());
configMapStore.flush(ctx);
assertEquals(7, mockWebServer.getRequestCount());
configMapStore.flush(ctx);
assertEquals(8, mockWebServer.getRequestCount());
}
@Test
void testQueryAndFlushIfNeeded() {
KubernetesJobAutoScalerContext ctx = createContext("cr1", kubernetesClient);
var key = "key";
var value = "value";
var configMapStore = new ConfigMapStore(kubernetesClient);
configMapStore.putSerializedState(ctx, key, value);
assertEquals(2, mockWebServer.getRequestCount());
configMapStore.flush(ctx);
assertEquals(3, mockWebServer.getRequestCount());
// Get from cache
assertThat(configMapStore.getSerializedState(ctx, key)).hasValue(value);
assertEquals(3, mockWebServer.getRequestCount());
// Get from kubernetes
configMapStore = new ConfigMapStore(kubernetesClient);
assertThat(configMapStore.getSerializedState(ctx, key)).hasValue(value);
assertEquals(4, mockWebServer.getRequestCount());
// Remove the key
configMapStore.removeSerializedState(ctx, key);
assertEquals(4, mockWebServer.getRequestCount());
configMapStore.flush(ctx);
assertEquals(5, mockWebServer.getRequestCount());
// get is null, so don't flush
configMapStore.getSerializedState(ctx, key);
assertEquals(5, mockWebServer.getRequestCount());
configMapStore.getSerializedState(ctx, key);
assertEquals(5, mockWebServer.getRequestCount());
configMapStore.getSerializedState(ctx, key);
assertEquals(5, mockWebServer.getRequestCount());
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1391225899
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ConfigMapStore.java:
##########
@@ -67,19 +67,14 @@ protected Optional<String> getSerializedState(
}
protected void removeSerializedState(KubernetesJobAutoScalerContext jobContext, String key) {
- getConfigMap(jobContext)
- .ifPresentOrElse(
- configMap -> configMap.getData().remove(key),
- () -> {
- throw new IllegalStateException(
- "The configMap isn't created, so the remove is unavailable.");
- });
+ getConfigMap(jobContext).ifPresent(configMap -> configMap.getData().remove(key));
}
public void flush(KubernetesJobAutoScalerContext jobContext) {
Optional<ConfigMap> configMapOpt = cache.get(jobContext.getJobKey());
if (configMapOpt == null || configMapOpt.isEmpty()) {
LOG.debug("The configMap isn't updated, so skip the flush.");
+ // Do not flush if there are no updates.
Review Comment:
I write 2 tests:
1. one is `testRemoveAndFlushEveryTime`, it's similar to current PR code
2. one is `testQueryAndFlushIfNeeded`, it's similar to master code
- `testRemoveAndFlushEveryTime` will remove and flush every time, and it will call kubernetes every time.
- `testQueryAndFlushIfNeeded` will query from cache, and value is null, so remove and flush isn't necessary. So it doens't call kubernetes.
My question is: the extra call is expected? And I think this PR is fine if it doesn't introuduce any extra opration.
These 2 tests can be added to ConfigMapStoreTest, and they can run directly.
```
@Test
void testRemoveAndFlushEveryTime() {
KubernetesJobAutoScalerContext ctx = createContext("cr1", kubernetesClient);
var key = "key";
var value = "value";
var configMapStore = new ConfigMapStore(kubernetesClient);
configMapStore.putSerializedState(ctx, key, value);
assertEquals(2, mockWebServer.getRequestCount());
configMapStore.flush(ctx);
assertEquals(3, mockWebServer.getRequestCount());
// Get from cache
assertThat(configMapStore.getSerializedState(ctx, key)).hasValue(value);
assertEquals(3, mockWebServer.getRequestCount());
// Get from kubernetes
configMapStore = new ConfigMapStore(kubernetesClient);
assertThat(configMapStore.getSerializedState(ctx, key)).hasValue(value);
assertEquals(4, mockWebServer.getRequestCount());
// Remove the key
configMapStore.removeSerializedState(ctx, key);
assertEquals(4, mockWebServer.getRequestCount());
configMapStore.flush(ctx);
assertEquals(5, mockWebServer.getRequestCount());
// Every flush will call kubernetes.
configMapStore.flush(ctx);
assertEquals(6, mockWebServer.getRequestCount());
configMapStore.flush(ctx);
assertEquals(7, mockWebServer.getRequestCount());
configMapStore.flush(ctx);
assertEquals(8, mockWebServer.getRequestCount());
}
@Test
void testQueryAndFlushIfNeeded() {
KubernetesJobAutoScalerContext ctx = createContext("cr1", kubernetesClient);
var key = "key";
var value = "value";
var configMapStore = new ConfigMapStore(kubernetesClient);
configMapStore.putSerializedState(ctx, key, value);
assertEquals(2, mockWebServer.getRequestCount());
configMapStore.flush(ctx);
assertEquals(3, mockWebServer.getRequestCount());
// Get from cache
assertThat(configMapStore.getSerializedState(ctx, key)).hasValue(value);
assertEquals(3, mockWebServer.getRequestCount());
// Get from kubernetes
configMapStore = new ConfigMapStore(kubernetesClient);
assertThat(configMapStore.getSerializedState(ctx, key)).hasValue(value);
assertEquals(4, mockWebServer.getRequestCount());
// Remove the key
configMapStore.removeSerializedState(ctx, key);
assertEquals(4, mockWebServer.getRequestCount());
configMapStore.flush(ctx);
assertEquals(5, mockWebServer.getRequestCount());
// get is null, so don't flush
configMapStore.getSerializedState(ctx, key);
assertEquals(5, mockWebServer.getRequestCount());
configMapStore.getSerializedState(ctx, key);
assertEquals(5, mockWebServer.getRequestCount());
configMapStore.getSerializedState(ctx, key);
assertEquals(5, mockWebServer.getRequestCount());
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1391499500
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobAutoScalerImpl.java:
##########
@@ -121,37 +120,10 @@ public void cleanup(KEY jobKey) {
}
private void clearStatesAfterAutoscalerDisabled(Context ctx) throws Exception {
- var needFlush = false;
- var parallelismOverrides = stateStore.getParallelismOverrides(ctx);
- if (!parallelismOverrides.isEmpty()) {
- needFlush = true;
- stateStore.removeParallelismOverrides(ctx);
- }
-
- var collectedMetrics = stateStore.getCollectedMetrics(ctx);
- if (!collectedMetrics.isEmpty()) {
- needFlush = true;
- stateStore.removeCollectedMetrics(ctx);
- }
-
- var scalingHistory = stateStore.getScalingHistory(ctx);
- if (!scalingHistory.isEmpty()) {
- var trimmedScalingHistory =
- trimScalingHistory(clock.instant(), ctx.getConfiguration(), scalingHistory);
- if (trimmedScalingHistory.isEmpty()) {
- // All scaling histories are trimmed.
- needFlush = true;
- stateStore.removeScalingHistory(ctx);
- } else if (!scalingHistory.equals(trimmedScalingHistory)) {
- // Some scaling histories are trimmed.
- needFlush = true;
- stateStore.storeScalingHistory(ctx, trimmedScalingHistory);
- }
- }
-
- if (needFlush) {
- stateStore.flush(ctx);
- }
+ stateStore.removeParallelismOverrides(ctx);
+ stateStore.removeCollectedMetrics(ctx);
+ stateStore.removeScalingHistory(ctx);
+ stateStore.flush(ctx);
Review Comment:
can we add a new `clearAll` or `removeAll` method that would guarantee that all state is cleaned and then we can test for subsequent flushing not hitting k8s api
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1392315670
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ConfigMapStore.java:
##########
@@ -44,13 +43,44 @@ public class ConfigMapStore {
private final KubernetesClient kubernetesClient;
- // The cache for each resourceId may be in three states:
- // 1. The resourceId doesn't exist : ConfigMap isn't loaded from kubernetes, or it's deleted
- // 2 Exists, Optional.empty() : The ConfigMap doesn't exist in Kubernetes
- // 3. Exists, Not Empty : We have loaded the ConfigMap from kubernetes, it may not be the same
- // if not flushed already
- private final ConcurrentHashMap<ResourceID, Optional<ConfigMap>> cache =
- new ConcurrentHashMap<>();
+ static class ConfigMapState {
+ private boolean flushed = true;
+ private boolean exists = true;
+
+ @VisibleForTesting ConfigMap configMap;
+
+ public Map<String, String> getData() {
+ return Collections.unmodifiableMap(configMap.getData());
+ }
+
+ public void clear() {
Review Comment:
Maybe we should actually use this instead of calling the separate clear methods. Makes it much "cleaner"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1391538017
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ConfigMapStore.java:
##########
@@ -67,19 +67,14 @@ protected Optional<String> getSerializedState(
}
protected void removeSerializedState(KubernetesJobAutoScalerContext jobContext, String key) {
- getConfigMap(jobContext)
- .ifPresentOrElse(
- configMap -> configMap.getData().remove(key),
- () -> {
- throw new IllegalStateException(
- "The configMap isn't created, so the remove is unavailable.");
- });
+ getConfigMap(jobContext).ifPresent(configMap -> configMap.getData().remove(key));
}
public void flush(KubernetesJobAutoScalerContext jobContext) {
Optional<ConfigMap> configMapOpt = cache.get(jobContext.getJobKey());
if (configMapOpt == null || configMapOpt.isEmpty()) {
LOG.debug("The configMap isn't updated, so skip the flush.");
+ // Do not flush if there are no updates.
Review Comment:
Thanks for adding the tests. You are right that the retrieval will always go to Kubernetes. The current implementation has some bugs with the bookkeeping to minimize the number of API calls. It does make more requests than necessary.
I've rewrote the ConfigMapStore to minimize the number of times we go to Kubernetes. We only retrieve once now on lookup. If the ConfigMap does not exist, we will only create the ConfigMap once we flush back any changes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1393995934
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapView.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * 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.state;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.util.Preconditions;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.client.dsl.Resource;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+class ConfigMapView {
+
+ enum State {
+ NEEDS_CREATE,
+ NEEDS_UPDATE,
+ UP_TO_DATE
+ }
+
+ private State state;
+
+ private ConfigMap configMap;
+
+ private final Function<ConfigMap, Resource<ConfigMap>> resourceRetriever;
+
+ public ConfigMapView(
+ Supplier<ConfigMap> configMapRetriever,
+ Supplier<ConfigMap> configMapBuilder,
+ Function<ConfigMap, Resource<ConfigMap>> resourceRetriever) {
+ var existingConfigMap = configMapRetriever.get();
+ if (existingConfigMap != null) {
+ refreshConfigMap(existingConfigMap);
+ } else {
+ this.configMap = configMapBuilder.get();
+ this.state = State.NEEDS_CREATE;
Review Comment:
Updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1394033152
##########
flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/state/KubernetesAutoScalerStateStoreTest.java:
##########
@@ -330,4 +331,42 @@ public void testDiscardInvalidHistory() throws Exception {
ctx, KubernetesAutoScalerStateStore.SCALING_HISTORY_KEY))
.isEmpty();
}
+
+ @Test
+ public void testDiscardAllState() {
Review Comment:
Good catch! I've added tests for both classes to test the actual class: https://github.com/apache/flink-kubernetes-operator/pull/710/commits/d36180265b753c991ca86dc5e8d7e6347aa45ec2
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#issuecomment-1812689805
Small follow-up which was meant to be included in this PR: https://github.com/apache/flink-kubernetes-operator/pull/714
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1392788916
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ConfigMapStore.java:
##########
@@ -44,13 +43,44 @@ public class ConfigMapStore {
private final KubernetesClient kubernetesClient;
- // The cache for each resourceId may be in three states:
- // 1. The resourceId doesn't exist : ConfigMap isn't loaded from kubernetes, or it's deleted
- // 2 Exists, Optional.empty() : The ConfigMap doesn't exist in Kubernetes
- // 3. Exists, Not Empty : We have loaded the ConfigMap from kubernetes, it may not be the same
- // if not flushed already
- private final ConcurrentHashMap<ResourceID, Optional<ConfigMap>> cache =
- new ConcurrentHashMap<>();
+ static class ConfigMapState {
+ private boolean flushed = true;
+ private boolean exists = true;
+
+ @VisibleForTesting ConfigMap configMap;
+
+ public Map<String, String> getData() {
+ return Collections.unmodifiableMap(configMap.getData());
+ }
+
+ public void clear() {
Review Comment:
Added a method to clear all state.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1392786446
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ConfigMapStore.java:
##########
@@ -44,13 +43,44 @@ public class ConfigMapStore {
private final KubernetesClient kubernetesClient;
- // The cache for each resourceId may be in three states:
- // 1. The resourceId doesn't exist : ConfigMap isn't loaded from kubernetes, or it's deleted
- // 2 Exists, Optional.empty() : The ConfigMap doesn't exist in Kubernetes
- // 3. Exists, Not Empty : We have loaded the ConfigMap from kubernetes, it may not be the same
- // if not flushed already
- private final ConcurrentHashMap<ResourceID, Optional<ConfigMap>> cache =
- new ConcurrentHashMap<>();
+ static class ConfigMapState {
+ private boolean flushed = true;
+ private boolean exists = true;
+
+ @VisibleForTesting ConfigMap configMap;
Review Comment:
The updated code should reflect your request.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1391193239
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ConfigMapStore.java:
##########
@@ -67,19 +67,14 @@ protected Optional<String> getSerializedState(
}
protected void removeSerializedState(KubernetesJobAutoScalerContext jobContext, String key) {
- getConfigMap(jobContext)
- .ifPresentOrElse(
- configMap -> configMap.getData().remove(key),
- () -> {
- throw new IllegalStateException(
- "The configMap isn't created, so the remove is unavailable.");
- });
+ getConfigMap(jobContext).ifPresent(configMap -> configMap.getData().remove(key));
}
public void flush(KubernetesJobAutoScalerContext jobContext) {
Optional<ConfigMap> configMapOpt = cache.get(jobContext.getJobKey());
if (configMapOpt == null || configMapOpt.isEmpty()) {
LOG.debug("The configMap isn't updated, so skip the flush.");
+ // Do not flush if there are no updates.
Review Comment:
I'm thinking whether any updating or removeing can be detected, please give me a few times to test it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1391256838
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ConfigMapStore.java:
##########
@@ -67,19 +67,14 @@ protected Optional<String> getSerializedState(
}
protected void removeSerializedState(KubernetesJobAutoScalerContext jobContext, String key) {
- getConfigMap(jobContext)
- .ifPresentOrElse(
- configMap -> configMap.getData().remove(key),
- () -> {
- throw new IllegalStateException(
- "The configMap isn't created, so the remove is unavailable.");
- });
+ getConfigMap(jobContext).ifPresent(configMap -> configMap.getData().remove(key));
}
public void flush(KubernetesJobAutoScalerContext jobContext) {
Optional<ConfigMap> configMapOpt = cache.get(jobContext.getJobKey());
if (configMapOpt == null || configMapOpt.isEmpty()) {
LOG.debug("The configMap isn't updated, so skip the flush.");
+ // Do not flush if there are no updates.
Review Comment:
We may need to added an extra cache to store the state that last updated. When the flush is called, then compare them. And call kubernetes if they are not same.
And it's better to add a test to check the flush doesn't call kubernetes if state isn't updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1391225899
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ConfigMapStore.java:
##########
@@ -67,19 +67,14 @@ protected Optional<String> getSerializedState(
}
protected void removeSerializedState(KubernetesJobAutoScalerContext jobContext, String key) {
- getConfigMap(jobContext)
- .ifPresentOrElse(
- configMap -> configMap.getData().remove(key),
- () -> {
- throw new IllegalStateException(
- "The configMap isn't created, so the remove is unavailable.");
- });
+ getConfigMap(jobContext).ifPresent(configMap -> configMap.getData().remove(key));
}
public void flush(KubernetesJobAutoScalerContext jobContext) {
Optional<ConfigMap> configMapOpt = cache.get(jobContext.getJobKey());
if (configMapOpt == null || configMapOpt.isEmpty()) {
LOG.debug("The configMap isn't updated, so skip the flush.");
+ // Do not flush if there are no updates.
Review Comment:
I write 2 tests:
1. one is `testRemoveAndFlushEveryTime`, it's similar to current PR code
2. one is `testQueryAndFlushIfNeeded`, it's similar to master code
- `testRemoveAndFlushEveryTime` will remove and flush every time, and it will call kubernetes every time.
- `testQueryAndFlushIfNeeded` will query from cache, and value is null, so remove and flush isn't necessary. So it doens't call kubernetes.
My question is: the extra call is expected? And I think this PR is fine if it doesn't introuduce any extra opration. Please correct my if my understanding is wrong, thanks a lot.
These 2 tests can be added to ConfigMapStoreTest, and they can run directly.
```
@Test
void testRemoveAndFlushEveryTime() {
KubernetesJobAutoScalerContext ctx = createContext("cr1", kubernetesClient);
var key = "key";
var value = "value";
var configMapStore = new ConfigMapStore(kubernetesClient);
configMapStore.putSerializedState(ctx, key, value);
assertEquals(2, mockWebServer.getRequestCount());
configMapStore.flush(ctx);
assertEquals(3, mockWebServer.getRequestCount());
// Get from cache
assertThat(configMapStore.getSerializedState(ctx, key)).hasValue(value);
assertEquals(3, mockWebServer.getRequestCount());
// Get from kubernetes
configMapStore = new ConfigMapStore(kubernetesClient);
assertThat(configMapStore.getSerializedState(ctx, key)).hasValue(value);
assertEquals(4, mockWebServer.getRequestCount());
// Remove the key
configMapStore.removeSerializedState(ctx, key);
assertEquals(4, mockWebServer.getRequestCount());
configMapStore.flush(ctx);
assertEquals(5, mockWebServer.getRequestCount());
// Every flush will call kubernetes.
configMapStore.flush(ctx);
assertEquals(6, mockWebServer.getRequestCount());
configMapStore.flush(ctx);
assertEquals(7, mockWebServer.getRequestCount());
configMapStore.flush(ctx);
assertEquals(8, mockWebServer.getRequestCount());
}
@Test
void testQueryAndFlushIfNeeded() {
KubernetesJobAutoScalerContext ctx = createContext("cr1", kubernetesClient);
var key = "key";
var value = "value";
var configMapStore = new ConfigMapStore(kubernetesClient);
configMapStore.putSerializedState(ctx, key, value);
assertEquals(2, mockWebServer.getRequestCount());
configMapStore.flush(ctx);
assertEquals(3, mockWebServer.getRequestCount());
// Get from cache
assertThat(configMapStore.getSerializedState(ctx, key)).hasValue(value);
assertEquals(3, mockWebServer.getRequestCount());
// Get from kubernetes
configMapStore = new ConfigMapStore(kubernetesClient);
assertThat(configMapStore.getSerializedState(ctx, key)).hasValue(value);
assertEquals(4, mockWebServer.getRequestCount());
// Remove the key
configMapStore.removeSerializedState(ctx, key);
assertEquals(4, mockWebServer.getRequestCount());
configMapStore.flush(ctx);
assertEquals(5, mockWebServer.getRequestCount());
// get is Optional.empty(), so don't flush
assertThat(configMapStore.getSerializedState(ctx, key)).isEmpty();
assertEquals(5, mockWebServer.getRequestCount());
assertThat(configMapStore.getSerializedState(ctx, key)).isEmpty();
assertEquals(5, mockWebServer.getRequestCount());
assertThat(configMapStore.getSerializedState(ctx, key)).isEmpty();
assertEquals(5, mockWebServer.getRequestCount());
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1394116830
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapStore.java:
##########
@@ -99,49 +93,30 @@ public void removeInfoFromCache(ResourceID resourceID) {
cache.remove(resourceID);
}
- private Optional<ConfigMap> getConfigMap(KubernetesJobAutoScalerContext jobContext) {
+ private ConfigMapView getConfigMap(KubernetesJobAutoScalerContext jobContext) {
return cache.computeIfAbsent(
jobContext.getJobKey(), (id) -> getConfigMapFromKubernetes(jobContext));
}
- private Map<String, String> getOrCreateState(KubernetesJobAutoScalerContext jobContext) {
+ private ConfigMapView getOrCreateState(KubernetesJobAutoScalerContext jobContext) {
return cache.compute(
- jobContext.getJobKey(),
- (id, configMapOpt) -> {
- // If in the cache and valid simply return
- if (configMapOpt != null && configMapOpt.isPresent()) {
- return configMapOpt;
- }
- // Otherwise get or create
- return Optional.of(getOrCreateConfigMapFromKubernetes(jobContext));
- })
- .get()
- .getData();
+ jobContext.getJobKey(),
+ (id, configMapView) -> {
+ // If in the cache and valid simply return
+ if (configMapView != null) {
+ return configMapView;
+ }
Review Comment:
That method was duplicated anyhow, removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710#discussion_r1392008277
##########
flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobAutoScalerImplTest.java:
##########
@@ -299,60 +296,15 @@ void testAutoscalerDisabled() throws Exception {
scalingHistory.put(Instant.ofEpochMilli(100), new ScalingSummary());
scalingHistory.put(Instant.ofEpochMilli(200), new ScalingSummary());
- // Test all scaling aren't expired
- getInstantScalingSummaryTreeMap(
- scalingHistory, Clock.fixed(Instant.ofEpochMilli(250), ZoneId.systemDefault()), 2);
+ stateStore.storeScalingHistory(context, Map.of(new JobVertexID(), scalingHistory));
+ assertFalse(stateStore.getScalingHistory(context).isEmpty());
- // Test one scaling aren't expired
- getInstantScalingSummaryTreeMap(
- scalingHistory, Clock.fixed(Instant.ofEpochMilli(350), ZoneId.systemDefault()), 1);
-
- // Test all scaling are expired
- getInstantScalingSummaryTreeMap(
- scalingHistory, Clock.fixed(Instant.ofEpochMilli(450), ZoneId.systemDefault()), 0);
- }
-
- private void getInstantScalingSummaryTreeMap(
- SortedMap<Instant, ScalingSummary> scalingHistoryData,
- Clock clock,
- int expectedScalingHistorySize)
- throws Exception {
- stateStore = new TestingAutoscalerStateStore<>();
var autoscaler =
new JobAutoScalerImpl<>(
null, null, null, eventCollector, scalingRealizer, stateStore);
-
- enrichStateStore(scalingHistoryData);
- stateStore.flush(context);
- assertThat(stateStore.getFlushCount()).isEqualTo(1);
-
- autoscaler.setClock(clock);
autoscaler.scale(context);
- assertThat(stateStore.getParallelismOverrides(context)).isEmpty();
- assertThat(stateStore.getCollectedMetrics(context)).isEmpty();
-
- if (expectedScalingHistorySize > 0) {
- Map<JobVertexID, SortedMap<Instant, ScalingSummary>> scalingHistory =
- stateStore.getScalingHistory(context);
- assertThat(scalingHistory).isNotEmpty();
- assertThat(scalingHistory.values())
- .allMatch(aa -> aa.size() == expectedScalingHistorySize);
- } else {
- assertThat(stateStore.getScalingHistory(context)).isEmpty();
- }
- assertThat(stateStore.getFlushCount()).isEqualTo(2);
- }
-
- private void enrichStateStore(SortedMap<Instant, ScalingSummary> scalingHistory) {
- var v1 = new JobVertexID();
- var v2 = new JobVertexID();
- stateStore.storeParallelismOverrides(
- context, Map.of(v1.toString(), "1", v2.toString(), "2"));
-
- var metricHistory = new TreeMap<Instant, CollectedMetrics>();
- stateStore.storeCollectedMetrics(context, metricHistory);
- stateStore.storeScalingHistory(context, Map.of(v1, scalingHistory, v2, scalingHistory));
+ assertTrue(stateStore.getScalingHistory(context).isEmpty());
Review Comment:
Could we test `ParallelismOverrides` and `CollectedMetrics` as well?
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ConfigMapStore.java:
##########
@@ -44,13 +43,44 @@ public class ConfigMapStore {
private final KubernetesClient kubernetesClient;
- // The cache for each resourceId may be in three states:
- // 1. The resourceId doesn't exist : ConfigMap isn't loaded from kubernetes, or it's deleted
- // 2 Exists, Optional.empty() : The ConfigMap doesn't exist in Kubernetes
- // 3. Exists, Not Empty : We have loaded the ConfigMap from kubernetes, it may not be the same
- // if not flushed already
- private final ConcurrentHashMap<ResourceID, Optional<ConfigMap>> cache =
- new ConcurrentHashMap<>();
+ static class ConfigMapState {
+ private boolean flushed = true;
+ private boolean exists = true;
+
+ @VisibleForTesting ConfigMap configMap;
+
+ public Map<String, String> getData() {
Review Comment:
```suggestion
@VisibleForTesting
public Map<String, String> getData() {
```
`getData` is only used for test
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ConfigMapStore.java:
##########
@@ -63,29 +93,29 @@ protected void putSerializedState(
protected Optional<String> getSerializedState(
KubernetesJobAutoScalerContext jobContext, String key) {
- return getConfigMap(jobContext).map(configMap -> configMap.getData().get(key));
+ return Optional.ofNullable(getConfigMap(jobContext).configMap.getData().get(key));
}
protected void removeSerializedState(KubernetesJobAutoScalerContext jobContext, String key) {
- getConfigMap(jobContext)
- .ifPresentOrElse(
- configMap -> configMap.getData().remove(key),
- () -> {
- throw new IllegalStateException(
- "The configMap isn't created, so the remove is unavailable.");
- });
+ getConfigMap(jobContext).removeKey(key);
}
public void flush(KubernetesJobAutoScalerContext jobContext) {
- Optional<ConfigMap> configMapOpt = cache.get(jobContext.getJobKey());
- if (configMapOpt == null || configMapOpt.isEmpty()) {
+ ConfigMapState configMapState = cache.get(jobContext.getJobKey());
+ if (configMapState == null || (configMapState.flushed && configMapState.exists)) {
Review Comment:
It's better to extract a method `isFlushNeeded` for `ConfigMapState` instead of call `configMapState.flushed && configMapState.exists` directly.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ConfigMapStore.java:
##########
@@ -44,13 +43,44 @@ public class ConfigMapStore {
private final KubernetesClient kubernetesClient;
- // The cache for each resourceId may be in three states:
- // 1. The resourceId doesn't exist : ConfigMap isn't loaded from kubernetes, or it's deleted
- // 2 Exists, Optional.empty() : The ConfigMap doesn't exist in Kubernetes
- // 3. Exists, Not Empty : We have loaded the ConfigMap from kubernetes, it may not be the same
- // if not flushed already
- private final ConcurrentHashMap<ResourceID, Optional<ConfigMap>> cache =
- new ConcurrentHashMap<>();
+ static class ConfigMapState {
+ private boolean flushed = true;
+ private boolean exists = true;
+
+ @VisibleForTesting ConfigMap configMap;
+
+ public Map<String, String> getData() {
+ return Collections.unmodifiableMap(configMap.getData());
+ }
+
+ public void clear() {
Review Comment:
It's not used, right? Could it be removed?
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ConfigMapStore.java:
##########
@@ -44,13 +43,44 @@ public class ConfigMapStore {
private final KubernetesClient kubernetesClient;
- // The cache for each resourceId may be in three states:
- // 1. The resourceId doesn't exist : ConfigMap isn't loaded from kubernetes, or it's deleted
- // 2 Exists, Optional.empty() : The ConfigMap doesn't exist in Kubernetes
- // 3. Exists, Not Empty : We have loaded the ConfigMap from kubernetes, it may not be the same
- // if not flushed already
- private final ConcurrentHashMap<ResourceID, Optional<ConfigMap>> cache =
- new ConcurrentHashMap<>();
+ static class ConfigMapState {
+ private boolean flushed = true;
+ private boolean exists = true;
+
+ @VisibleForTesting ConfigMap configMap;
Review Comment:
Could we add a setter for `ConfigMap configMap`, and update `exists` inside of this setter. It's safer and clearer for ConfigMapState, and easy to read and understand.
And could `ConfigMapState` as a separate class instead of `inner class of ConfigMapStroe` and mark all fields as private? I see `ConfigMapStroe` update the configMap directly, and it's dangerous in the future if some developers update it without update `flushed` or `exists`. And it's easy to introduce bug in the future.
In short: I prefer `ConfigMapState` to expose the interface rather than the details.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ConfigMapStore.java:
##########
@@ -44,13 +43,44 @@ public class ConfigMapStore {
private final KubernetesClient kubernetesClient;
- // The cache for each resourceId may be in three states:
- // 1. The resourceId doesn't exist : ConfigMap isn't loaded from kubernetes, or it's deleted
- // 2 Exists, Optional.empty() : The ConfigMap doesn't exist in Kubernetes
- // 3. Exists, Not Empty : We have loaded the ConfigMap from kubernetes, it may not be the same
- // if not flushed already
- private final ConcurrentHashMap<ResourceID, Optional<ConfigMap>> cache =
- new ConcurrentHashMap<>();
+ static class ConfigMapState {
+ private boolean flushed = true;
+ private boolean exists = true;
+
+ @VisibleForTesting ConfigMap configMap;
+
+ public Map<String, String> getData() {
+ return Collections.unmodifiableMap(configMap.getData());
+ }
+
+ public void clear() {
+ if (configMap.getData().isEmpty()) {
+ return;
+ }
+ configMap.getData().clear();
+ flushed = false;
+ }
+
+ public void removeKey(String key) {
+ var oldKey = configMap.getData().remove(key);
+ if (oldKey != null) {
+ flushed = false;
+ }
+ }
+
+ public void put(String key, String value) {
+ configMap.getData().put(key, value);
+ flushed = false;
+ }
+ }
+
+ // The cache for each resourceId may be in four states:
+ // 1. No cache entry: ConfigMap isn't loaded from kubernetes, or it's deleted.
+ // 2 Cache entry, not created : The ConfigMap doesn't exist in Kubernetes.
+ // 2 Cache entry, not flushed : The ConfigMap exists in Kubernetes, but it is not updated yet.
+ // 3. Cache entry, flushed and created : We have loaded the ConfigMap from kubernetes, and it's
+ // up-to-date.
Review Comment:
```suggestion
// 2. Cache entry, not created : The ConfigMap doesn't exist in Kubernetes.
// 3. Cache entry, not flushed : The ConfigMap exists in Kubernetes, but it is not updated yet.
// 4. Cache entry, flushed and created : We have loaded the ConfigMap from kubernetes, and it's
// up-to-date.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33527] Simplify state store cleanup logic [flink-kubernetes-operator]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora merged PR #710:
URL: https://github.com/apache/flink-kubernetes-operator/pull/710
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org