AHeise commented on a change in pull request #16150:
URL: https://github.com/apache/flink/pull/16150#discussion_r650213981
##########
File path:
flink-core/src/main/java/org/apache/flink/configuration/Configuration.java
##########
@@ -750,7 +760,29 @@ public boolean contains(ConfigOption<?> configOption) {
//
--------------------------------------------------------------------------------------------
+ private boolean removeConfigInternal(String key, boolean canBePrefixMap) {
+ final boolean keysRemoved;
+ if (canBePrefixMap) {
+ keysRemoved = removePrefixMap(this.confData, key);
+ } else {
+ keysRemoved = false;
+ }
+ return keysRemoved || this.confData.remove(key) != null;
+ }
+
+ private boolean containsInternal(String key, boolean canBePrefixMap) {
+ final boolean containsExactKey = this.confData.containsKey(key);
+ if (!canBePrefixMap || containsExactKey) {
+ return containsExactKey;
+ }
+ return containsPrefixMap(this.confData, key);
+ }
Review comment:
```
if (canBePrefixMap && containsPrefixMap(this.confData, key)) {
return true;
}
return this.confData.containsKey(key);
```
(Could be simplified to oneliner)
##########
File path:
flink-core/src/main/java/org/apache/flink/configuration/Configuration.java
##########
@@ -661,14 +665,16 @@ public boolean containsKey(String key) {
*/
@PublicEvolving
public boolean contains(ConfigOption<?> configOption) {
+ final boolean canBePrefixMap = canBePrefixMap(configOption);
+
+ // first try the current key
synchronized (this.confData) {
- // first try the current key
- if (this.confData.containsKey(configOption.key())) {
+ if (containsInternal(configOption.key(), canBePrefixMap)) {
Review comment:
Not a massive fan of having that logic spread on 3 different places.
Could we extract some generic lookup method and lambda that acts on that?
##########
File path:
flink-core/src/main/java/org/apache/flink/configuration/Configuration.java
##########
@@ -731,13 +738,16 @@ public boolean contains(ConfigOption<?> configOption) {
* @return true is config has been removed, false otherwise
*/
public <T> boolean removeConfig(ConfigOption<T> configOption) {
+ final boolean canBePrefixMap = canBePrefixMap(configOption);
+
synchronized (this.confData) {
// try the current key
- Object oldValue = this.confData.remove(configOption.key());
- if (oldValue == null) {
+ boolean keysRemoved = removeConfigInternal(configOption.key(),
canBePrefixMap);
Review comment:
```
// try the current key
if (removeConfigInternal(configOption.key(), canBePrefixMap)) {
return true;
}
// try the fallback keys
for (FallbackKey fallbackKey : configOption.fallbackKeys()) {
if (removeConfigInternal(fallbackKey.getKey(),
canBePrefixMap)) {
loggingFallback(fallbackKey, configOption);
return true;
}
}
return false;
```
##########
File path:
flink-core/src/main/java/org/apache/flink/configuration/Configuration.java
##########
@@ -750,7 +760,29 @@ public boolean contains(ConfigOption<?> configOption) {
//
--------------------------------------------------------------------------------------------
+ private boolean removeConfigInternal(String key, boolean canBePrefixMap) {
+ final boolean keysRemoved;
+ if (canBePrefixMap) {
+ keysRemoved = removePrefixMap(this.confData, key);
+ } else {
+ keysRemoved = false;
+ }
+ return keysRemoved || this.confData.remove(key) != null;
Review comment:
```
if (canBePrefixMap && removePrefixMap(this.confData, key)) {
return true;
}
return this.confData.remove(key) != null;
```
##########
File path:
flink-core/src/main/java/org/apache/flink/configuration/Configuration.java
##########
@@ -759,31 +791,49 @@ public boolean contains(ConfigOption<?> configOption) {
}
synchronized (this.confData) {
+ if (canBePrefixMap) {
+ removePrefixMap(this.confData, key);
+ }
Review comment:
Is this symmetric?
If I set `map.k1 = 1` and then `map = k2: 2` does it yield the same as vice
versa?
I'm thinking that we should probably keep `k1`.
##########
File path:
flink-core/src/test/java/org/apache/flink/configuration/ConfigurationTest.java
##########
@@ -346,6 +347,70 @@ public void testToMap() {
assertEquals("3 s", configuration.toMap().get(DURATION_OPTION.key()));
}
+ @Test
+ public void testPrefixAndNonPrefixMap() {
+ final Map<String, String> map = new HashMap<>();
+ map.put("prop1", "value1");
+ map.put("prop2", "12");
+
+ // not contained
+ {
Review comment:
Please make separate unit tests.
##########
File path:
flink-core/src/main/java/org/apache/flink/configuration/Configuration.java
##########
@@ -759,31 +791,49 @@ public boolean contains(ConfigOption<?> configOption) {
}
synchronized (this.confData) {
+ if (canBePrefixMap) {
+ removePrefixMap(this.confData, key);
+ }
this.confData.put(key, value);
}
}
private Optional<Object> getRawValue(String key) {
+ return getRawValue(key, false);
+ }
+
+ private Optional<Object> getRawValue(String key, boolean canBePrefixMap) {
if (key == null) {
throw new NullPointerException("Key must not be null.");
}
synchronized (this.confData) {
- return Optional.ofNullable(this.confData.get(key));
+ final Object valueFromExactKey = this.confData.get(key);
+ if (!canBePrefixMap || valueFromExactKey != null) {
+ return Optional.ofNullable(valueFromExactKey);
+ }
+ final Map<String, String> valueFromPrefixMap =
+ convertToPropertiesPrefixed(confData, key);
+ if (valueFromPrefixMap.isEmpty()) {
+ return Optional.empty();
+ }
+ return Optional.of(valueFromPrefixMap);
Review comment:
```
final Object valueFromExactKey = this.confData.get(key);
if (valueFromExactKey == null && canBePrefixMap) {
final Map<String, String> valueFromPrefixMap =
convertToPropertiesPrefixed(confData, key);
if (valueFromPrefixMap.isEmpty()) {
return Optional.empty();
}
return Optional.of(valueFromPrefixMap);
}
return Optional.ofNullable(valueFromExactKey);
```
--
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.
For queries about this service, please contact Infrastructure at:
[email protected]
