[ZEPPELIN-6394] Make the interpreter setting write safe under concurrent saves

zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/InterpreterSettingManager.java

@@ -131,6 +131,9 @@ public class InterpreterSettingManager implements NoteEventListener {
     new ConcurrentHashMap<>());
   private final Map<String, List<Meter>> interpreterSettingsMeters = new ConcurrentHashMap<>();
 
+  /** Guards {@link #saveToFile()}, the only writer of the interpreter setting file. */
+  private final Object saveLock = new Object();
+
   private final List<Repository> interpreterRepositories;
   private InterpreterOption defaultOption;
   private String defaultInterpreterGroup;
@@ -369,11 +372,19 @@ private void removeInterpreterSetting(String id) {
     }
   }
 
+  /**
+   * Snapshotting the settings and writing them out has to be one step. Dependency downloads
+   * save from a thread per interpreter setting, so without this an older snapshot can be
+   * written after a newer one and drop a setting that was added in between from the file,
+   * while it survives in memory until the next restart.
+   */
   public void saveToFile() throws IOException {
-    InterpreterInfoSaving info = new InterpreterInfoSaving();
-    info.interpreterSettings = new HashMap<>(interpreterSettings);
-    info.interpreterRepositories = interpreterRepositories;
-    configStorage.save(info);
+    synchronized (saveLock) {
+      InterpreterInfoSaving info = new InterpreterInfoSaving();
+      info.interpreterSettings = new HashMap<>(interpreterSettings);
+      info.interpreterRepositories = interpreterRepositories;
+      configStorage.save(info);
+    }
   }
 
   private void initMetrics() {

zeppelin-server/src/main/java/org/apache/zeppelin/util/FileUtils.java

@@ -57,8 +57,12 @@ public static void atomicWriteToFile(String content, File file, Set<PosixFilePer
     }
     try {
       file.getParentFile().mkdirs();
+      // A plain replace is not atomic, so two writers moving their own temp file onto the
+      // same destination can make one of them fail with NoSuchFileException and leave its
+      // temp file behind. The temp file is created in the destination directory, so this
+      // move never crosses file systems and the atomic move is always supported.
       Files.move(tempFile.toPath(), destinationFilePath,
-              StandardCopyOption.REPLACE_EXISTING); //StandardCopyOption.ATOMIC_MOVE);
+              StandardCopyOption.REPLACE_EXISTING, StandardCopyOption.ATOMIC_MOVE);
     } catch (IOException iox) {
       if (!tempFile.delete()) {
         tempFile.deleteOnExit();

zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/InterpreterSettingManagerSaveRaceTest.java

@@ -0,0 +1,181 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zeppelin.interpreter;
+
+import org.apache.zeppelin.conf.ZeppelinConfiguration;
+import org.apache.zeppelin.dep.Dependency;
+import org.apache.zeppelin.display.AngularObjectRegistryListener;
+import org.apache.zeppelin.helium.ApplicationEventListener;
+import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcessListener;
+import org.apache.zeppelin.notebook.NotebookAuthorizationInfoSaving;
+import org.apache.zeppelin.storage.ConfigStorage;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+
+/**
+ * Dependency downloads write the interpreter settings from a thread per interpreter setting,
+ * so {@link InterpreterSettingManager#saveToFile()} runs concurrently. Its snapshot and its
+ * write have to stay one step, otherwise a snapshot taken before a setting was added can be
+ * written after the save that carried it, and the file loses that setting while it survives
+ * in memory until the next restart.
+ */
+class InterpreterSettingManagerSaveRaceTest extends AbstractInterpreterTest {
+
+  private static final String NEW_SETTING_NAME = "added_during_save";
+  /** How long the first writer waits for a second one before it gives up and writes. */
+  private static final long HANDOVER_TIMEOUT_MS = 3000;
+
+  @Test
+  void testSaveDuringAnotherSaveIsNotOverwritten() throws Exception {
+    RecordingConfigStorage storage = new RecordingConfigStorage(zConf);
+    InterpreterSettingManager manager = new InterpreterSettingManager(zConf,
+        mock(AngularObjectRegistryListener.class), mock(RemoteInterpreterProcessListener.class),
+        mock(ApplicationEventListener.class), storage, pluginManager);
+    try {
+      storage.observe();
+
+      // The first writer holds its write until a second write has gone through, which is the
+      // window an unserialized saveToFile leaves open between its snapshot and its write.
+      List<Throwable> errors = new CopyOnWriteArrayList<>();
+      CountDownLatch firstWriterDone = new CountDownLatch(1);
+      Thread firstWriter = new Thread(() -> {
+        try {
+          manager.saveToFile();
+        } catch (Exception e) {
+          errors.add(e);
+        } finally {
+          firstWriterDone.countDown();
+        }
+      }, "saveToFile-holder");
+      firstWriter.setDaemon(true);
+      firstWriter.start();
+
+      assertTrue(storage.firstWriteEntered.await(10, TimeUnit.SECONDS),
+          "the first writer never reached the storage");
+
+      // Add a setting while the first writer is inside its write. createNewSetting() mutates
+      // the settings and saves them, so its content has to survive.
+      InterpreterOption option = new InterpreterOption();
+      option.setPerNote("scoped");
+      option.setPerUser("scoped");
+      Map<String, InterpreterProperty> properties = new HashMap<>();
+      properties.put("property_4", new InterpreterProperty("property_4", "value_4"));
+      manager.createNewSetting(NEW_SETTING_NAME, "test", new ArrayList<Dependency>(),
+          option, properties);
+
+      assertTrue(firstWriterDone.await(30, TimeUnit.SECONDS),
+          "the first writer did not finish within the timeout");
+      assertTrue(errors.isEmpty(), () -> "A writer threw: " + errors);
+
+      List<Set<String>> writes = storage.writtenSettingNames;
+      assertFalse(writes.isEmpty(), "nothing was written");
+      assertTrue(writes.get(writes.size() - 1).contains(NEW_SETTING_NAME),
+          () -> "the last write dropped " + NEW_SETTING_NAME + ", writes were " + writes);
+    } finally {
+      manager.close();
+    }
+  }
+
+  /**
+   * Records the interpreter names of every write. The first write is held back until a second
+   * one has been recorded, so that a save which is not serialized can overtake it.
+   */
+  private static class RecordingConfigStorage extends ConfigStorage {
+    private final List<Set<String>> writtenSettingNames = new CopyOnWriteArrayList<>();
+    private final CountDownLatch firstWriteEntered = new CountDownLatch(1);
+    private volatile boolean observing;
+    private volatile boolean firstWrite = true;
+
+    RecordingConfigStorage(ZeppelinConfiguration zConf) {
+      super(zConf);
+    }
+
+    /** Start recording, so that the writes of the constructor are left out. */
+    void observe() {
+      observing = true;
+    }
+
+    @Override
+    public void save(InterpreterInfoSaving settingInfos) throws IOException {
+      if (!observing) {
+        return;
+      }
+      Set<String> names = new HashSet<>();
+      for (InterpreterSetting setting : settingInfos.interpreterSettings.values()) {
+        names.add(setting.getName());
+      }
+      boolean holdForSecondWrite = firstWrite;
+      firstWrite = false;
+      if (holdForSecondWrite) {
+        firstWriteEntered.countDown();
+        // Serialized saves make this wait time out, which is the point: the second write
+        // cannot start before this one is done, so it lands last and keeps its content.
+        long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(HANDOVER_TIMEOUT_MS);
+        while (writtenSettingNames.isEmpty() && System.nanoTime() < deadline) {
+          try {
+            Thread.sleep(20);
+          } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            break;
+          }
+        }
+      }
+      writtenSettingNames.add(Collections.unmodifiableSet(names));
+    }
+
+    @Override
+    public InterpreterInfoSaving loadInterpreterSettings() throws IOException {
+      return null;
+    }
+
+    @Override
+    public void save(NotebookAuthorizationInfoSaving authorizationInfoSaving) throws IOException {
+      // not used by this test
+    }
+
+    @Override
+    public NotebookAuthorizationInfoSaving loadNotebookAuthorization() throws IOException {
+      return null;
+    }
+
+    @Override
+    public String loadCredentials() throws IOException {
+      return null;
+    }
+
+    @Override
+    public void saveCredentials(String credentials) throws IOException {
+      // not used by this test
+    }
+  }
+}

zeppelin-server/src/test/java/org/apache/zeppelin/util/FileUtilsTest.java

@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zeppelin.util;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class FileUtilsTest {
+
+  private static final int WRITERS = 16;
+  private static final int WRITES_PER_WRITER = 300;
+
+  /**
+   * Several threads write the same file, the way the interpreter settings are written when
+   * more than one interpreter finishes downloading its dependencies. A replace that is not
+   * atomic lets one of the moves fail with NoSuchFileException and leaves its temp file in
+   * the destination directory.
+   */
+  @Test
+  void testConcurrentWritesToTheSameFileSucceed(@TempDir Path tempDir) throws Exception {
+    File target = tempDir.resolve("interpreter.json").toFile();
+    ExecutorService pool = Executors.newFixedThreadPool(WRITERS);
+    CountDownLatch start = new CountDownLatch(1);
+    List<Throwable> failures = new CopyOnWriteArrayList<>();
+    try {
+      for (int writer = 0; writer < WRITERS; writer++) {
+        final int writerId = writer;
+        pool.submit(() -> {
+          try {
+            start.await();
+            for (int i = 0; i < WRITES_PER_WRITER; i++) {
+              FileUtils.atomicWriteToFile("{\"writer\":" + writerId + ",\"write\":" + i + "}",
+                  target);
+            }
+          } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+          } catch (Exception e) {
+            failures.add(e);
+          }
+        });
+      }
+      start.countDown();
+      pool.shutdown();
+      assertTrue(pool.awaitTermination(120, TimeUnit.SECONDS), "the writers did not finish");
+    } finally {
+      pool.shutdownNow();
+    }
+
+    assertTrue(failures.isEmpty(), () -> "writing the file failed: " + failures);
+    File[] leftovers = tempDir.toFile().listFiles((dir, name) -> name.endsWith(".tmp"));
+    assertEquals(0, leftovers == null ? 0 : leftovers.length,
+        "a failed move left its temp file behind");
+    assertTrue(FileUtils.readFromFile(target).startsWith("{\"writer\":"),
+        "the file does not hold a complete write");
+  }
+}