This is an automated email from the ASF dual-hosted git repository.
tbonelee pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/zeppelin.git
The following commit(s) were added to refs/heads/master by this push:
new 6f60984f4e [ZEPPELIN-6394] Make the interpreter setting write safe
under concurrent saves
6f60984f4e is described below
commit 6f60984f4e74344de200bf36715cc533b56c4f7f
Author: 김동환 <[email protected]>
AuthorDate: Sat Sep 26 00:53:17 2026 +0900
[ZEPPELIN-6394] Make the interpreter setting write safe under concurrent
saves
### What is this PR for?
When several interpreters finish downloading their dependencies at the same
time, each of them saves the interpreter settings from its own thread, so
`conf/interpreter.json` is written concurrently. The reporter's log shows those
writes failing:
```
ERROR ({Thread-16}) InterpreterSetting.java[run]:1003 - Fail to save
interpreter.json
java.nio.file.NoSuchFileException:
/data/soft/zeppelin-0.11.2/conf/interpreter.json
at java.nio.file.Files.move(Files.java:1395)
at
org.apache.zeppelin.util.FileUtils.atomicWriteToFile(FileUtils.java:60)
at
org.apache.zeppelin.storage.LocalConfigStorage.save(LocalConfigStorage.java:54)
at
org.apache.zeppelin.interpreter.InterpreterSettingManager.saveToFile(InterpreterSettingManager.java:350)
at
org.apache.zeppelin.interpreter.InterpreterSetting$1.run(InterpreterSetting.java:1001)
```
There are two problems behind it.
**1. The write itself is not atomic.** `FileUtils.atomicWriteToFile` writes
a temp file and then moves it over the destination with `REPLACE_EXISTING`
alone. That replace is not a single operation, so when two writers move their
own temp file onto the same destination, one of them can fail with
`NoSuchFileException` and leave its temp file behind in the config directory.
The failure is swallowed as an `ERROR` log, so the settings silently do not
reach disk.
Measured on the current code with 16 threads writing the same file 300
times each:
| move options | failed writes (of 4800) | leftover temp files |
| --- | --- | --- |
| `REPLACE_EXISTING` | 9 – 24, reproduced on every run | same count |
| `REPLACE_EXISTING, ATOMIC_MOVE` | 0 | 0 |
This PR requests `ATOMIC_MOVE` as well. The temp file is created in the
destination directory (`Files.createTempFile(destinationDirectory, ...)`), so
the move never crosses file systems and the atomic move is always supported.
The option was present as a trailing comment from the first version of this
method in ZEPPELIN-4907 and was never enabled; there is no recorded reason for
leaving it off.
**2. The snapshot and the write are two steps.**
`InterpreterSettingManager.saveToFile` copies the settings and then writes
them, without holding anything in between, so an older snapshot can be written
after a newer one. When that happens, a setting that was added in between is
dropped from the file while it survives in memory, and the loss only shows up
after the next restart. This is the serialization the Jira asks for; it is done
with a private lock rather than `synchronized` on t [...]
Note for the Jira description: on the default `LocalConfigStorage` the
concurrent writes do not corrupt the file, because every write goes to its own
temp file. What they do is fail, and lose an update.
**Relation to #5144.** The `saveToFile()` serialization in this PR overlaps
#5144, which was opened by the reporter and already carries an LGTM that a
later rebase dismissed. That PR does not cover the move itself, which is what
produces the `NoSuchFileException` in the report. If it is preferred to land
#5144 for the serialization, I am happy to drop the `InterpreterSettingManager`
change from here and leave only the `FileUtils` fix.
### What type of PR is it?
Bug Fix
### What is the Jira issue?
* https://issues.apache.org/jira/browse/ZEPPELIN-6394
### How should this be tested?
Two new tests, both verified to fail without the corresponding change:
* `FileUtilsTest.testConcurrentWritesToTheSameFileSucceed` — 16 threads
write the same file 300 times each and no write may fail or leave a temp file
behind. Fails on master on every run with the `NoSuchFileException` above.
*
`InterpreterSettingManagerSaveRaceTest.testSaveDuringAnotherSaveIsNotOverwritten`
— a setting is added while another save is in flight, and the last write has
to carry it. Fails on master with `the last write dropped added_during_save`.
Existing suites run: `InterpreterSettingManagerTest`,
`InterpreterSettingTest`, `InterpreterFactoryTest`,
`InterpreterInfoSavingTest`, `LocalConfigStorageTest`, `CredentialsTest`,
`LocalRecoveryStorageTest`.
### Questions:
* Does the license files need to update? No
* Is there breaking changes for older versions? No
* Does this needs documentation? No
Closes #5426 from dev-donghwan/ZEPPELIN-6394.
Signed-off-by: ChanHo Lee <[email protected]>
---
.../interpreter/InterpreterSettingManager.java | 19 ++-
.../java/org/apache/zeppelin/util/FileUtils.java | 6 +-
.../InterpreterSettingManagerSaveRaceTest.java | 181 +++++++++++++++++++++
.../org/apache/zeppelin/util/FileUtilsTest.java | 83 ++++++++++
4 files changed, 284 insertions(+), 5 deletions(-)
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 d0cff37f2b..1a85f82d5d 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<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 @@ public class InterpreterSettingManager implements
NoteEventListener {
}
}
+ /**
+ * 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 2698ffcd06..fdf9ac6898 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 class FileUtils {
}
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();
diff --git
a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/InterpreterSettingManagerSaveRaceTest.java
b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/InterpreterSettingManagerSaveRaceTest.java
new file mode 100644
index 0000000000..270c9b488f
--- /dev/null
+++
b/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
+ }
+ }
+}
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 0000000000..a43d3dc63e
--- /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<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");
+ }
+}
