Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[FLINK-33942][configuration] Fix the bug that DelegatingConfiguration misses the prefix in some get methods #24004

Merged
merged 3 commits into from
Dec 29, 2023
Merged

[FLINK-33942][configuration] Fix the bug that DelegatingConfiguration misses the prefix in some get methods #24004

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
bafc661
[FLINK-33942][configuration][junit5-migration] Migrate DelegatingConf…
1996fanrui Dec 26, 2023
24bf955
[FLINK-33942][configuration] Fix the bug that DelegatingConfiguration…
1996fanrui Dec 26, 2023
f6d8713
[FLINK-33942][configuration][refactor] Using ConfigOption instead of …
1996fanrui Dec 26, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 11 additions & 11 deletions flink-core/src/main/java/org/apache/flink/configuration/DelegatingConfiguration.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,7 +90,7 @@ public void setString(String key, String value) {

@Override
public void setString(ConfigOption<String> key, String value) {
this.backingConfig.setString(prefix + key.key(), value);
this.backingConfig.setString(prefixOption(key, prefix), value);
}

@Override
Expand All @@ -117,7 +117,7 @@ public int getInteger(ConfigOption<Integer> configOption) {

@Override
public int getInteger(ConfigOption<Integer> configOption, int overrideDefault) {
return this.backingConfig.getInteger(configOption, overrideDefault);
return this.backingConfig.getInteger(prefixOption(configOption, prefix), overrideDefault);
}

@Override
Expand All @@ -127,7 +127,7 @@ public void setInteger(String key, int value) {

@Override
public void setInteger(ConfigOption<Integer> key, int value) {
this.backingConfig.setInteger(prefix + key.key(), value);
this.backingConfig.setInteger(prefixOption(key, prefix), value);
}

@Override
Expand All @@ -142,7 +142,7 @@ public long getLong(ConfigOption<Long> configOption) {

@Override
public long getLong(ConfigOption<Long> configOption, long overrideDefault) {
return this.backingConfig.getLong(configOption, overrideDefault);
return this.backingConfig.getLong(prefixOption(configOption, prefix), overrideDefault);
}

@Override
Expand All @@ -152,7 +152,7 @@ public void setLong(String key, long value) {

@Override
public void setLong(ConfigOption<Long> key, long value) {
this.backingConfig.setLong(prefix + key.key(), value);
this.backingConfig.setLong(prefixOption(key, prefix), value);
}

@Override
Expand All @@ -172,12 +172,12 @@ public void setBoolean(String key, boolean value) {

@Override
public void setBoolean(ConfigOption<Boolean> key, boolean value) {
this.backingConfig.setBoolean(prefix + key.key(), value);
this.backingConfig.setBoolean(prefixOption(key, prefix), value);
}

@Override
public boolean getBoolean(ConfigOption<Boolean> configOption, boolean overrideDefault) {
return this.backingConfig.getBoolean(configOption, overrideDefault);
return this.backingConfig.getBoolean(prefixOption(configOption, prefix), overrideDefault);
}

@Override
Expand All @@ -192,7 +192,7 @@ public float getFloat(ConfigOption<Float> configOption) {

@Override
public float getFloat(ConfigOption<Float> configOption, float overrideDefault) {
return this.backingConfig.getFloat(configOption, overrideDefault);
return this.backingConfig.getFloat(prefixOption(configOption, prefix), overrideDefault);
}

@Override
Expand All @@ -202,7 +202,7 @@ public void setFloat(String key, float value) {

@Override
public void setFloat(ConfigOption<Float> key, float value) {
this.backingConfig.setFloat(prefix + key.key(), value);
this.backingConfig.setFloat(prefixOption(key, prefix), value);
}

@Override
Expand All @@ -217,7 +217,7 @@ public double getDouble(ConfigOption<Double> configOption) {

@Override
public double getDouble(ConfigOption<Double> configOption, double overrideDefault) {
return this.backingConfig.getDouble(configOption, overrideDefault);
return this.backingConfig.getDouble(prefixOption(configOption, prefix), overrideDefault);
}

@Override
Expand All @@ -227,7 +227,7 @@ public void setDouble(String key, double value) {

@Override
public void setDouble(ConfigOption<Double> key, double value) {
this.backingConfig.setDouble(prefix + key.key(), value);
this.backingConfig.setDouble(prefixOption(key, prefix), value);
}

@Override
Expand Down
94 changes: 69 additions & 25 deletions flink-core/src/test/java/org/apache/flink/configuration/DelegatingConfigurationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,23 +18,21 @@

package org.apache.flink.configuration;

import org.assertj.core.api.Assertions;
import org.junit.Test;
import org.junit.jupiter.api.Test;

import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.util.Map;
import java.util.Properties;
import java.util.Set;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.assertj.core.api.Assertions.assertThat;

/** Tests for the {@link DelegatingConfiguration}. */
public class DelegatingConfigurationTest {
class DelegatingConfigurationTest {

@Test
public void testIfDelegatesImplementAllMethods() throws IllegalArgumentException {
void testIfDelegatesImplementAllMethods() throws IllegalArgumentException {

// For each method in the Configuration class...
Method[] confMethods = Configuration.class.getDeclaredMethods();
Expand Down Expand Up @@ -70,27 +68,28 @@ public void testIfDelegatesImplementAllMethods() throws IllegalArgumentException
}
}

assertTrue(
"Configuration method '"
+ configurationMethod.getName()
+ "' has not been wrapped correctly in DelegatingConfiguration wrapper",
hasMethod);
assertThat(hasMethod)
.as(
"Configuration method '"
+ configurationMethod.getName()
+ "' has not been wrapped correctly in DelegatingConfiguration wrapper")
.isTrue();
}
}

@Test
public void testDelegationConfigurationWithNullPrefix() {
void testDelegationConfigurationWithNullPrefix() {
Configuration backingConf = new Configuration();
backingConf.setValueInternal("test-key", "value", false);

DelegatingConfiguration configuration = new DelegatingConfiguration(backingConf, null);
Set<String> keySet = configuration.keySet();

assertEquals(keySet, backingConf.keySet());
assertThat(backingConf.keySet()).isEqualTo(keySet);
}

@Test
public void testDelegationConfigurationWithPrefix() {
void testDelegationConfigurationWithPrefix() {
String prefix = "pref-";
String expectedKey = "key";

Expand All @@ -102,9 +101,7 @@ public void testDelegationConfigurationWithPrefix() {

DelegatingConfiguration configuration = new DelegatingConfiguration(backingConf, prefix);
Set<String> keySet = configuration.keySet();

assertEquals(keySet.size(), 1);
assertEquals(keySet.iterator().next(), expectedKey);
assertThat(keySet).hasSize(1).containsExactly(expectedKey);

/*
* Key does not match the prefix
Expand All @@ -113,13 +110,11 @@ public void testDelegationConfigurationWithPrefix() {
backingConf.setValueInternal("test-key", "value", false);

configuration = new DelegatingConfiguration(backingConf, prefix);
keySet = configuration.keySet();

assertTrue(keySet.isEmpty());
assertThat(configuration.keySet()).isEmpty();
}

@Test
public void testDelegationConfigurationToMapConsistentWithAddAllToProperties() {
void testDelegationConfigurationToMapConsistentWithAddAllToProperties() {
Configuration conf = new Configuration();
conf.setString("k0", "v0");
conf.setString("prefix.k1", "v1");
Expand All @@ -136,15 +131,64 @@ public void testDelegationConfigurationToMapConsistentWithAddAllToProperties() {
mapProperties.put(entry.getKey(), entry.getValue());
}
// Verification
assertEquals(properties, mapProperties);
assertThat(mapProperties).isEqualTo(properties);
}

@Test
public void testSetReturnsDelegatingConfiguration() {
void testSetReturnsDelegatingConfiguration() {
final Configuration conf = new Configuration();
final DelegatingConfiguration delegatingConf = new DelegatingConfiguration(conf, "prefix.");

Assertions.assertThat(delegatingConf.set(CoreOptions.DEFAULT_PARALLELISM, 1))
.isSameAs(delegatingConf);
assertThat(delegatingConf.set(CoreOptions.DEFAULT_PARALLELISM, 1)).isSameAs(delegatingConf);
}

@Test
void testGetWithOverrideDefault() {
Configuration original = new Configuration();
final DelegatingConfiguration delegatingConf =
new DelegatingConfiguration(original, "prefix.");

// Test for integer
ConfigOption<Integer> integerOption =
ConfigOptions.key("integer.key").intType().noDefaultValue();

// integerOption doesn't exist in delegatingConf, and it should be overrideDefault.
original.setInteger(integerOption, 1);
assertThat(delegatingConf.getInteger(integerOption, 2)).isEqualTo(2);

// integerOption exists in delegatingConf, and it should be value that set before.
delegatingConf.setInteger(integerOption, 3);
assertThat(delegatingConf.getInteger(integerOption, 2)).isEqualTo(3);

// Test for float
ConfigOption<Float> floatOption =
ConfigOptions.key("float.key").floatType().noDefaultValue();
original.setFloat(floatOption, 4f);
assertThat(delegatingConf.getFloat(floatOption, 5f)).isEqualTo(5f);
delegatingConf.setFloat(floatOption, 6f);
assertThat(delegatingConf.getFloat(floatOption, 5f)).isEqualTo(6f);

// Test for double
ConfigOption<Double> doubleOption =
ConfigOptions.key("double.key").doubleType().noDefaultValue();
original.setDouble(doubleOption, 7d);
assertThat(delegatingConf.getDouble(doubleOption, 8d)).isEqualTo(8d);
delegatingConf.setDouble(doubleOption, 9f);
assertThat(delegatingConf.getDouble(doubleOption, 8d)).isEqualTo(9f);

// Test for long
ConfigOption<Long> longOption = ConfigOptions.key("long.key").longType().noDefaultValue();
original.setLong(longOption, 10L);
assertThat(delegatingConf.getLong(longOption, 11L)).isEqualTo(11L);
delegatingConf.setLong(longOption, 12L);
assertThat(delegatingConf.getLong(longOption, 11L)).isEqualTo(12L);

// Test for boolean
ConfigOption<Boolean> booleanOption =
ConfigOptions.key("boolean.key").booleanType().noDefaultValue();
original.setBoolean(booleanOption, false);
assertThat(delegatingConf.getBoolean(booleanOption, true)).isEqualTo(true);
delegatingConf.setBoolean(booleanOption, false);
assertThat(delegatingConf.getBoolean(booleanOption, true)).isEqualTo(false);
}
}