From 120ab1e167e52eaf49686b005be24fe8d719516a Mon Sep 17 00:00:00 2001
From: Terence Yim <terencey@google.com>
Date: Fri, 6 Sep 2019 12:21:07 -0700
Subject: [PATCH] Support mutating CConfiguration via Map

---
 .../cdap/common/conf/CConfigurationUtil.java  | 40 ++++++++++++++++---
 .../common/conf/CConfigurationMapTest.java    | 36 +++++++++++------
 2 files changed, 58 insertions(+), 18 deletions(-)

diff --git a/cdap-common/src/main/java/io/cdap/cdap/common/conf/CConfigurationUtil.java b/cdap-common/src/main/java/io/cdap/cdap/common/conf/CConfigurationUtil.java
index 9b2ac4e39dc2..9724625a4ae5 100644
--- a/cdap-common/src/main/java/io/cdap/cdap/common/conf/CConfigurationUtil.java
+++ b/cdap-common/src/main/java/io/cdap/cdap/common/conf/CConfigurationUtil.java
@@ -21,14 +21,14 @@
 import java.io.File;
 import java.net.URI;
 import java.util.AbstractMap;
+import java.util.AbstractSet;
 import java.util.ArrayList;
-import java.util.Collections;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Properties;
 import java.util.Set;
-import java.util.stream.Collectors;
-import java.util.stream.StreamSupport;
 
 /**
  * Utilities for {@link CConfiguration}.
@@ -38,14 +38,42 @@ public class CConfigurationUtil extends Configuration {
   private CConfigurationUtil() { }
 
   /**
-   * Returns an immutable {@link Map} that is backed by the given {@link CConfiguration}.
+   * Returns a {@link Map} that is backed by the given {@link CConfiguration}. Updates to the returned Map
+   * will be reflected in the {@link CConfiguration}.
    */
   public static Map<String, String> asMap(CConfiguration cConf) {
     return new AbstractMap<String, String>() {
       @Override
       public Set<Entry<String, String>> entrySet() {
-        return Collections.unmodifiableSet(
-          StreamSupport.stream(cConf.spliterator(), false).collect(Collectors.toSet()));
+        return new AbstractSet<Entry<String, String>>() {
+
+          @Override
+          public boolean add(Entry<String, String> entry) {
+            String oldValue = cConf.get(entry.getKey());
+            if (Objects.equals(oldValue, entry.getValue())) {
+              return false;
+            }
+            cConf.set(entry.getKey(), entry.getValue());
+            return true;
+          }
+
+          @Override
+          public Iterator<Entry<String, String>> iterator() {
+            return cConf.iterator();
+          }
+
+          @Override
+          public int size() {
+            return cConf.size();
+          }
+        };
+      }
+
+      @Override
+      public String put(String key, String value) {
+        String oldValue = cConf.get(key);
+        cConf.set(key, value);
+        return oldValue;
       }
 
       @Override
diff --git a/cdap-common/src/test/java/io/cdap/cdap/common/conf/CConfigurationMapTest.java b/cdap-common/src/test/java/io/cdap/cdap/common/conf/CConfigurationMapTest.java
index 0bcf7754a9b5..6dded4c95d58 100644
--- a/cdap-common/src/test/java/io/cdap/cdap/common/conf/CConfigurationMapTest.java
+++ b/cdap-common/src/test/java/io/cdap/cdap/common/conf/CConfigurationMapTest.java
@@ -61,23 +61,35 @@ public void testIsEmpty() {
     Assert.assertTrue(CConfigurationUtil.asMap(cConf).isEmpty());
   }
 
-  @Test (expected = UnsupportedOperationException.class)
-  public void testDisallowPut() {
-    CConfigurationUtil.asMap(CConfiguration.create()).put("test.key", "test.value");
+  @Test
+  public void testPut() {
+    Map<String, String> map = CConfigurationUtil.asMap(CConfiguration.create());
+    Assert.assertNull(map.put("test.key", "test.value"));
+    Assert.assertEquals("test.value", map.get("test.key"));
   }
 
-  @Test (expected = UnsupportedOperationException.class)
-  public void testDisallowPutAll() {
-    CConfigurationUtil.asMap(CConfiguration.create()).putAll(Collections.singletonMap("test.key", "test.value"));
+  @Test
+  public void testPutAll() {
+    Map<String, String> map = CConfigurationUtil.asMap(CConfiguration.create());
+    map.putAll(Collections.singletonMap("test.key", "test.value"));
+    Assert.assertEquals("test.value", map.get("test.key"));
   }
 
-  @Test (expected = UnsupportedOperationException.class)
-  public void testDisallowRemove() {
-    CConfigurationUtil.asMap(CConfiguration.create()).remove(Constants.CFG_LOCAL_DATA_DIR);
+  @Test
+  public void testRemove() {
+    Map<String, String> map = CConfigurationUtil.asMap(CConfiguration.create());
+    String oldValue = map.get(Constants.CFG_LOCAL_DATA_DIR);
+    Assert.assertNotNull(oldValue);
+    String removed = map.remove(Constants.CFG_LOCAL_DATA_DIR);
+    Assert.assertEquals(oldValue, removed);
+    Assert.assertNull(map.get(Constants.CFG_LOCAL_DATA_DIR));
   }
 
-  @Test (expected = UnsupportedOperationException.class)
-  public void testDisallowRemoveSet() {
-    CConfigurationUtil.asMap(CConfiguration.create()).entrySet().removeIf(e -> true);
+  @Test
+  public void testIteratorRemove() {
+    Map<String, String> map = CConfigurationUtil.asMap(CConfiguration.create());
+    Assert.assertNotNull(map.get(Constants.CFG_LOCAL_DATA_DIR));
+    map.entrySet().removeIf(e -> e.getKey().equals(Constants.CFG_LOCAL_DATA_DIR));
+    Assert.assertNull(map.get(Constants.CFG_LOCAL_DATA_DIR));
   }
 }