HADOOP-13500. Synchronizing iteration of Configuration properties object (#3775)

Signed-off-by: Akira Ajisaka <[email protected]>
(cherry picked from commit 4483607)
(cherry picked from commit 253b217)

Author: dbadaya1

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/conf/Configuration.java

@@ -2906,11 +2906,13 @@ public Iterator<Map.Entry<String, String>> iterator() {
     // methods that allow non-strings to be put into configurations are removed,
     // we could replace properties with a Map<String,String> and get rid of this
     // code.
-    Map<String,String> result = new HashMap<String,String>();
-    for(Map.Entry<Object,Object> item: getProps().entrySet()) {
-      if (item.getKey() instanceof String &&
-          item.getValue() instanceof String) {
+    Properties props = getProps();
+    Map<String, String> result = new HashMap<>();
+    synchronized (props) {
+      for (Map.Entry<Object, Object> item : props.entrySet()) {
+        if (item.getKey() instanceof String && item.getValue() instanceof String) {
           result.put((String) item.getKey(), (String) item.getValue());
+        }
       }
     }
     return result.entrySet().iterator();

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/conf/TestConfiguration.java

@@ -38,13 +38,15 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.ConcurrentModificationException;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import java.util.Random;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.regex.Pattern;
 import static java.util.concurrent.TimeUnit.*;
 
@@ -2590,4 +2592,31 @@ private static Configuration checkCDATA(byte[] bytes) {
     assertEquals("  prefix >cdata\nsuffix  ", conf.get("cdata-whitespace"));
     return conf;
   }
+
+  @Test
+  public void testConcurrentModificationDuringIteration() throws InterruptedException {
+    Configuration configuration = new Configuration();
+    new Thread(() -> {
+      while (true) {
+        configuration.set(String.valueOf(Math.random()), String.valueOf(Math.random()));
+      }
+    }).start();
+
+    AtomicBoolean exceptionOccurred = new AtomicBoolean(false);
+
+    new Thread(() -> {
+      while (true) {
+        try {
+          configuration.iterator();
+        } catch (final ConcurrentModificationException e) {
+          exceptionOccurred.set(true);
+          break;
+        }
+      }
+    }).start();
+
+    Thread.sleep(1000); //give enough time for threads to run
+
+    assertFalse("ConcurrentModificationException occurred", exceptionOccurred.get());
+  }
 }