[FLINK-17789][core] DelegatingConfiguration should remove prefix instead of add prefix in toMap

This closes #12905

flink-core/src/main/java/org/apache/flink/configuration/DelegatingConfiguration.java

@@ -310,11 +310,13 @@ public final class DelegatingConfiguration extends Configuration {
 	@Override
 	public Map<String, String> toMap() {
 		Map<String, String> map = backingConfig.toMap();
-		Map<String, String> prefixed = new HashMap<>(map.size());
+		Map<String, String> prefixed = new HashMap<>();
 		for (Map.Entry<String, String> entry : map.entrySet()) {
-			prefixed.put(prefix + entry.getKey(), entry.getValue());
+			if (entry.getKey().startsWith(prefix)) {
+				String keyWithoutPrefix = entry.getKey().substring(prefix.length());
+				prefixed.put(keyWithoutPrefix, entry.getValue());
+			}
 		}
-
 		return prefixed;
 	}
 

flink-core/src/test/java/org/apache/flink/configuration/DelegatingConfigurationTest.java

@@ -23,6 +23,8 @@ import org.junit.Test;
 import java.lang.reflect.InvocationTargetException;
 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;
@@ -114,4 +116,25 @@ public class DelegatingConfigurationTest {
 
 		assertTrue(keySet.isEmpty());
 	}
+
+	@Test
+	public void testDelegationConfigurationToMapConsistentWithAddAllToProperties()  {
+		Configuration conf = new Configuration();
+		conf.setString("k0", "v0");
+		conf.setString("prefix.k1", "v1");
+		conf.setString("prefix.prefix.k2", "v2");
+		conf.setString("k3.prefix.prefix.k3", "v3");
+		DelegatingConfiguration dc = new DelegatingConfiguration(conf, "prefix.");
+		// Collect all properties
+		Properties properties = new Properties();
+		dc.addAllToProperties(properties);
+		// Convert the Map<String, String> object into a Properties object
+		Map<String, String> map = dc.toMap();
+		Properties mapProperties = new Properties();
+		for (Map.Entry<String, String> entry : map.entrySet()) {
+			mapProperties.put(entry.getKey(), entry.getValue());
+		}
+		// Verification
+		assertEquals(properties, mapProperties);
+	}
 }