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

Signed-off-by: Akira Ajisaka <aajisaka@apache.org>

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

@@ -2696,11 +2696,13 @@ public class Configuration implements Iterable<Map.Entry<String,String>>,
     // 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

@@ -36,12 +36,14 @@ import java.nio.charset.StandardCharsets;
 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.Random;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.regex.Pattern;
 import static java.util.concurrent.TimeUnit.*;
 
@@ -2133,6 +2135,39 @@ public class TestConfiguration extends TestCase {
     checkCDATA(os.toByteArray());
   }
 
+  @Test
+  public void testConcurrentModificationDuringIteration() throws InterruptedException {
+    final Configuration configuration = new Configuration();
+    new Thread(new Runnable() {
+      @Override
+      public void run() {
+        while (true) {
+          configuration.set(String.valueOf(Math.random()), String.valueOf(Math.random()));
+        }
+      }
+    }).start();
+
+    final AtomicBoolean exceptionOccurred = new AtomicBoolean(false);
+
+    new Thread(new Runnable() {
+      @Override
+      public void run() {
+        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());
+  }
+
   private static Configuration checkCDATA(byte[] bytes) {
     Configuration conf = new Configuration(false);
     conf.addResource(new ByteArrayInputStream(bytes));