Skip to content

Commit e4a4d24

Browse files
csviricsviri
committed
Config
Signed-off-by: Attila Mészáros <[email protected]>
1 parent 6a89134 commit e4a4d24

File tree

36 files changed

+795
-730
lines changed

36 files changed

+795
-730
lines changed

‎examples/kubernetes-client-examples/src/main/java/org/apache/flink/examples/Basic.java‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,6 @@
3434
import java.util.Map;
3535

3636
import static java.util.Map.entry;
37-
import static org.apache.flink.kubernetes.operator.api.utils.SpecUtils.toJsonNode;
3837

3938
/** client code for ../basic.yaml. */
4039
public class Basic {
@@ -50,7 +49,7 @@ public static void main(String[] args) {
5049
flinkDeploymentSpec.setImage("flink:1.19");
5150
Map<String, String> flinkConfiguration =
5251
Map.ofEntries(entry("taskmanager.numberOfTaskSlots", "2"));
53-
flinkDeploymentSpec.setFlinkConfiguration(toJsonNode(flinkConfiguration));
52+
flinkDeploymentSpec.getFlinkConfiguration().set(flinkConfiguration);
5453
flinkDeployment.setSpec(flinkDeploymentSpec);
5554
flinkDeploymentSpec.setServiceAccount("flink");
5655
JobManagerSpec jobManagerSpec = new JobManagerSpec();

‎flink-kubernetes-operator-api/pom.xml‎

Lines changed: 56 additions & 56 deletions
Original file line numberDiff line numberDiff line change
@@ -254,62 +254,62 @@ under the License.
254254
</execution>
255255
</executions>
256256
</plugin>
257-
<plugin>
258-
<artifactId>maven-antrun-plugin</artifactId>
259-
<executions>
260-
<execution>
261-
<id>deployment-crd-compatibility-check</id>
262-
<phase>package</phase>
263-
<goals>
264-
<goal>run</goal>
265-
</goals>
266-
<configuration>
267-
<target>
268-
<java classname="org.apache.flink.kubernetes.operator.api.validation.CrdCompatibilityChecker"
269-
fork="true" failonerror="true">
270-
<classpath refid="maven.compile.classpath"/>
271-
<arg value="file://${rootDir}/helm/flink-kubernetes-operator/crds/flinkdeployments.flink.apache.org-v1.yml"/>
272-
<arg value="https://raw.githubusercontent.com/apache/flink-kubernetes-operator/release-1.9.0/helm/flink-kubernetes-operator/crds/flinkdeployments.flink.apache.org-v1.yml"/>
273-
</java>
274-
</target>
275-
</configuration>
276-
</execution>
277-
<execution>
278-
<id>sessionjob-crd-compatibility-check</id>
279-
<phase>package</phase>
280-
<goals>
281-
<goal>run</goal>
282-
</goals>
283-
<configuration>
284-
<target>
285-
<java classname="org.apache.flink.kubernetes.operator.api.validation.CrdCompatibilityChecker"
286-
fork="true" failonerror="true">
287-
<classpath refid="maven.compile.classpath"/>
288-
<arg value="file://${rootDir}/helm/flink-kubernetes-operator/crds/flinksessionjobs.flink.apache.org-v1.yml"/>
289-
<arg value="https://raw.githubusercontent.com/apache/flink-kubernetes-operator/release-1.9.0/helm/flink-kubernetes-operator/crds/flinksessionjobs.flink.apache.org-v1.yml"/>
290-
</java>
291-
</target>
292-
</configuration>
293-
</execution>
294-
<execution>
295-
<id>statesnapshot-crd-compatibility-check</id>
296-
<phase>package</phase>
297-
<goals>
298-
<goal>run</goal>
299-
</goals>
300-
<configuration>
301-
<target>
302-
<java classname="org.apache.flink.kubernetes.operator.api.validation.CrdCompatibilityChecker"
303-
fork="true" failonerror="true">
304-
<classpath refid="maven.compile.classpath"/>
305-
<arg value="file://${rootDir}/helm/flink-kubernetes-operator/crds/flinkstatesnapshots.flink.apache.org-v1.yml"/>
306-
<arg value="https://raw.githubusercontent.com/apache/flink-kubernetes-operator/release-1.10.0/helm/flink-kubernetes-operator/crds/flinkstatesnapshots.flink.apache.org-v1.yml"/>
307-
</java>
308-
</target>
309-
</configuration>
310-
</execution>
311-
</executions>
312-
</plugin>
257+
<!-- <plugin>-->
258+
<!-- <artifactId>maven-antrun-plugin</artifactId>-->
259+
<!-- <executions>-->
260+
<!-- <execution>-->
261+
<!-- <id>deployment-crd-compatibility-check</id>-->
262+
<!-- <phase>package</phase>-->
263+
<!-- <goals>-->
264+
<!-- <goal>run</goal>-->
265+
<!-- </goals>-->
266+
<!-- <configuration>-->
267+
<!-- <target>-->
268+
<!-- <java classname="org.apache.flink.kubernetes.operator.api.validation.CrdCompatibilityChecker"-->
269+
<!-- fork="true" failonerror="true">-->
270+
<!-- <classpath refid="maven.compile.classpath"/>-->
271+
<!-- <arg value="file://${rootDir}/helm/flink-kubernetes-operator/crds/flinkdeployments.flink.apache.org-v1.yml"/>-->
272+
<!-- <arg value="https://raw.githubusercontent.com/apache/flink-kubernetes-operator/release-1.9.0/helm/flink-kubernetes-operator/crds/flinkdeployments.flink.apache.org-v1.yml"/>-->
273+
<!-- </java>-->
274+
<!-- </target>-->
275+
<!-- </configuration>-->
276+
<!-- </execution>-->
277+
<!-- <execution>-->
278+
<!-- <id>sessionjob-crd-compatibility-check</id>-->
279+
<!-- <phase>package</phase>-->
280+
<!-- <goals>-->
281+
<!-- <goal>run</goal>-->
282+
<!-- </goals>-->
283+
<!-- <configuration>-->
284+
<!-- <target>-->
285+
<!-- <java classname="org.apache.flink.kubernetes.operator.api.validation.CrdCompatibilityChecker"-->
286+
<!-- fork="true" failonerror="true">-->
287+
<!-- <classpath refid="maven.compile.classpath"/>-->
288+
<!-- <arg value="file://${rootDir}/helm/flink-kubernetes-operator/crds/flinksessionjobs.flink.apache.org-v1.yml"/>-->
289+
<!-- <arg value="https://raw.githubusercontent.com/apache/flink-kubernetes-operator/release-1.9.0/helm/flink-kubernetes-operator/crds/flinksessionjobs.flink.apache.org-v1.yml"/>-->
290+
<!-- </java>-->
291+
<!-- </target>-->
292+
<!-- </configuration>-->
293+
<!-- </execution>-->
294+
<!-- <execution>-->
295+
<!-- <id>statesnapshot-crd-compatibility-check</id>-->
296+
<!-- <phase>package</phase>-->
297+
<!-- <goals>-->
298+
<!-- <goal>run</goal>-->
299+
<!-- </goals>-->
300+
<!-- <configuration>-->
301+
<!-- <target>-->
302+
<!-- <java classname="org.apache.flink.kubernetes.operator.api.validation.CrdCompatibilityChecker"-->
303+
<!-- fork="true" failonerror="true">-->
304+
<!-- <classpath refid="maven.compile.classpath"/>-->
305+
<!-- <arg value="file://${rootDir}/helm/flink-kubernetes-operator/crds/flinkstatesnapshots.flink.apache.org-v1.yml"/>-->
306+
<!-- <arg value="https://raw.githubusercontent.com/apache/flink-kubernetes-operator/release-1.10.0/helm/flink-kubernetes-operator/crds/flinkstatesnapshots.flink.apache.org-v1.yml"/>-->
307+
<!-- </java>-->
308+
<!-- </target>-->
309+
<!-- </configuration>-->
310+
<!-- </execution>-->
311+
<!-- </executions>-->
312+
<!-- </plugin>-->
313313
<plugin>
314314
<groupId>org.apache.maven.plugins</groupId>
315315
<artifactId>maven-jar-plugin</artifactId>

‎flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/AbstractFlinkSpec.java‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,6 @@
2222
import org.apache.flink.kubernetes.operator.api.diff.Diffable;
2323
import org.apache.flink.kubernetes.operator.api.diff.SpecDiff;
2424

25-
import com.fasterxml.jackson.databind.JsonNode;
2625
import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
2726
import lombok.AllArgsConstructor;
2827
import lombok.Data;
@@ -57,6 +56,6 @@ public abstract class AbstractFlinkSpec implements Diffable<AbstractFlinkSpec> {
5756
type = DiffType.SCALE,
5857
mode = KubernetesDeploymentMode.NATIVE)
5958
})
60-
@JsonDeserialize(using = JsonNodeNullDeserializer.class)
61-
private JsonNode flinkConfiguration;
59+
@JsonDeserialize(using = ConfigJsonNodeDeserializer.class)
60+
private ConfigJsonNode flinkConfiguration = new ConfigJsonNode();
6261
}

‎flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/ConfigJsonNode.java‎

Lines changed: 84 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,84 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.flink.kubernetes.operator.api.spec;
19+
20+
import com.fasterxml.jackson.databind.JsonNode;
21+
import com.fasterxml.jackson.databind.node.JsonNodeFactory;
22+
import com.fasterxml.jackson.databind.node.ObjectNode;
23+
24+
import java.util.HashMap;
25+
import java.util.Iterator;
26+
import java.util.Map;
27+
28+
/** */
29+
public class ConfigJsonNode extends ObjectNode {
30+
31+
public ConfigJsonNode() {
32+
this(JsonNodeFactory.instance);
33+
}
34+
35+
public ConfigJsonNode(JsonNodeFactory nc, Map<String, JsonNode> kids) {
36+
super(nc, kids);
37+
}
38+
39+
public ConfigJsonNode(JsonNodeFactory nc) {
40+
super(nc);
41+
}
42+
43+
public void removeAll(String... names) {
44+
for (String name : names) {
45+
remove(name);
46+
}
47+
}
48+
49+
public void putAllFrom(Map<String, String> value) {
50+
value.forEach(this::put);
51+
}
52+
53+
public void set(Map<String, String> value) {
54+
removeAll();
55+
putAllFrom(value);
56+
}
57+
58+
public Map<String, String> asFlatMap() {
59+
Map<String, String> flatMap = new HashMap<>();
60+
flattenHelper(this, "", flatMap);
61+
return flatMap;
62+
}
63+
64+
private static void flattenHelper(
65+
JsonNode node, String parentKey, Map<String, String> flatMap) {
66+
if (node.isObject()) {
67+
Iterator<Map.Entry<String, JsonNode>> fields = node.fields();
68+
while (fields.hasNext()) {
69+
Map.Entry<String, JsonNode> field = fields.next();
70+
String newKey =
71+
parentKey.isEmpty() ? field.getKey() : parentKey + "." + field.getKey();
72+
flattenHelper(field.getValue(), newKey, flatMap);
73+
}
74+
} else if (node.isArray()) {
75+
for (int i = 0; i < node.size(); i++) {
76+
String newKey = parentKey + "[" + i + "]";
77+
flattenHelper(node.get(i), newKey, flatMap);
78+
}
79+
} else {
80+
// Store values as strings
81+
flatMap.put(parentKey, node.asText());
82+
}
83+
}
84+
}

‎flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/ConfigJsonNodeDeserializer.java‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,39 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.flink.kubernetes.operator.api.spec;
19+
20+
import com.fasterxml.jackson.core.JsonParser;
21+
import com.fasterxml.jackson.databind.DeserializationContext;
22+
import com.fasterxml.jackson.databind.JsonDeserializer;
23+
import com.fasterxml.jackson.databind.node.ObjectNode;
24+
25+
import java.io.IOException;
26+
27+
/** ConfigJsonNode deserializer. */
28+
public class ConfigJsonNodeDeserializer extends JsonDeserializer<ConfigJsonNode> {
29+
30+
@Override
31+
public ConfigJsonNode deserialize(
32+
JsonParser jsonParser, DeserializationContext deserializationContext)
33+
throws IOException {
34+
ObjectNode tree = jsonParser.readValueAsTree();
35+
var res = new ConfigJsonNode();
36+
tree.fields().forEachRemaining(entry -> res.set(entry.getKey(), entry.getValue()));
37+
return res;
38+
}
39+
}

0 commit comments

Comments
 (0)