diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/InterpreterSettingManager.java b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/InterpreterSettingManager.java index 8b4c2fe55be..21544dd0780 100644 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/InterpreterSettingManager.java +++ b/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> interpreterSettingsMeters = new ConcurrentHashMap<>(); + /** Guards {@link #saveToFile()}, the only writer of the interpreter setting file. */ + private final Object saveLock = new Object(); + private final List 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() { diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/util/FileUtils.java b/zeppelin-server/src/main/java/org/apache/zeppelin/util/FileUtils.java index 2698ffcd06d..fdf9ac6898a 100644 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/util/FileUtils.java +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/util/FileUtils.java @@ -57,8 +57,12 @@ public static void atomicWriteToFile(String content, File file, Set 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 properties = new HashMap<>(); + properties.put("property_4", new InterpreterProperty("property_4", "value_4")); + manager.createNewSetting(NEW_SETTING_NAME, "test", new ArrayList(), + 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> 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> 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 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 + } + } +} diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/util/FileUtilsTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/util/FileUtilsTest.java new file mode 100644 index 00000000000..a43d3dc63ea --- /dev/null +++ b/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 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"); + } +}