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 0e72120aae [ZEPPELIN-5858] Fix moveNote/saveNote race that duplicates
notes
0e72120aae is described below
commit 0e72120aaecaa13993b862fbd8b714cc6e5e2b54
Author: HwangRock <[email protected]>
AuthorDate: Mon Aug 10 23:55:20 2026 +0900
[ZEPPELIN-5858] Fix moveNote/saveNote race that duplicates notes
### What is this PR for?
`moveNote` mutates the folder tree, the `notesInfo` mapping and the
notebook repo without holding any lock, while `saveNote` derives its target
file name from the path carried by the `Note` object at save time. A save
racing with a move therefore writes the note back to its pre-move path, and the
same noteId ends up on disk twice. ZEPPELIN-5858 describes exactly this and
ships a reproduction; this PR ports that reproduction onto the current API and
fixes the race.
Reproduced on current master with the ported test
(`VFSNotebookRepoWithDelay` extends `VFSNotebookRepo` and injects repo latency,
e.g. remote storage):
```
Saving note 2MXKZT37A to folder_1/note_2MXKZT37A.zpln
Move note 2MXKZT37A to /folder_2/note
Move note 2MXKZT37A from /folder_1/note to /folder_2/note
Saving note 2MXKZT37A to folder_1/note_2MXKZT37A.zpln <- stale path,
resurrects the old file
Expected exactly one .zpln file, but found:
[folder_2/note_2MXKZT37A.zpln, folder_1/note_2MXKZT37A.zpln]
```
The fix serializes NoteManager's persistent mutations: `moveNote`,
`removeNote` and `moveFolder` now run their tree, mapping and repo mutations
inside the monitor `saveNote` already synchronizes on. Save callers and
`moveNote` share the cached `Note` instance, so a save that loses the monitor
to a concurrent move derives its file name from the already-updated path once
it enters — no note is written to its pre-move location. `saveNote`'s existing
contract (the `Note` object's path is [...]
Two deliberate details:
- The fixed lock order in this code base is note `readLock` -> NoteManager
monitor, because every save caller runs inside `processNote` holding the note's
readLock. `moveNote` therefore updates the cached note path directly via the
note cache instead of calling `processNote` while holding the monitor, and the
rename-triggered resave stays outside the monitor and reuses `saveNote`. This
keeps the lock order consistent on every path.
- The rename-triggered resave realigns the reloaded note with the move
target before saving: a note evicted from the cache is reloaded from disk at
that point, and the on-disk JSON still carries the pre-move name.
Verified the causality both ways: the reproduction test fails with two
`.zpln` files for the same noteId on master, passes with this change, and fails
again if only the `NoteManager` change is reverted.
### What type of PR is it?
Bug Fix
### Todos
* [x] Port the reproduction attached to ZEPPELIN-5858 onto the current
NotebookRepo API
* [x] Serialize NoteManager mutations
### What is the Jira issue?
https://issues.apache.org/jira/browse/ZEPPELIN-5858
### How should this be tested?
`mvn test -pl zeppelin-server -Dtest=NotebookServiceRaceConditionTest` —
the test runs `renameNote` and `insertParagraph` concurrently against a
delay-injecting `VFSNotebookRepo` subclass, then walks the notebook directory
and asserts exactly one `.zpln` file remains. Without the `NoteManager` change
it finds the same noteId at both the old and the new path.
Regression: `NoteManagerTest`, `NotebookServiceTest`, `NotebookTest`,
`LuceneSearchTest`, `ZeppelinRestApiTest` (86 tests) all pass.
### Questions:
* Does the license files need to update? No
* Is there breaking changes for older versions? No
* Does this needs documentation? No
Closes #5325 from HwangRock/ZEPPELIN-5858.
Signed-off-by: ChanHo Lee <[email protected]>
---
.../org/apache/zeppelin/notebook/NoteManager.java | 119 +++++++------
.../notebook/NoteManagerMoveResaveRaceTest.java | 196 +++++++++++++++++++++
.../notebook/repo/VFSNotebookRepoWithDelay.java | 82 +++++++++
.../notebook/repo/VFSNotebookRepoWithGetGate.java | 86 +++++++++
.../service/NotebookServiceRaceConditionTest.java | 164 +++++++++++++++++
5 files changed, 596 insertions(+), 51 deletions(-)
diff --git
a/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/NoteManager.java
b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/NoteManager.java
index 0635fde994..c31cad72b8 100644
---
a/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/NoteManager.java
+++
b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/NoteManager.java
@@ -187,11 +187,9 @@ public class NoteManager {
if (note.isRemoved()) {
LOGGER.warn("Try to save note: {} when it is removed", note.getId());
} else {
- addOrUpdateNoteNode(this.noteTree, new NoteInfo(note), false);
- noteCache.putNote(note);
- // Make sure to execute `notebookRepo.save()` successfully in concurrent
context
- // Otherwise, the NullPointerException will be thrown when invoking
notebookRepo.get() in the following operations.
synchronized (this) {
+ addOrUpdateNoteNode(this.noteTree, new NoteInfo(note), false);
+ noteCache.putNote(note);
this.notebookRepo.save(note, subject);
}
}
@@ -220,12 +218,14 @@ public class NoteManager {
* @throws IOException
*/
public void removeNote(String noteId, AuthenticationInfo subject) throws
IOException {
- NoteTree tree = this.noteTree;
- String notePath = tree.notesInfo.remove(noteId);
- Folder folder = getOrCreateFolder(tree, getFolderName(notePath));
- folder.removeNote(getNoteName(notePath));
- noteCache.removeNote(noteId);
- this.notebookRepo.remove(noteId, notePath, subject);
+ synchronized (this) {
+ NoteTree tree = this.noteTree;
+ String notePath = tree.notesInfo.remove(noteId);
+ Folder folder = getOrCreateFolder(tree, getFolderName(notePath));
+ folder.removeNote(getNoteName(notePath));
+ noteCache.removeNote(noteId);
+ this.notebookRepo.remove(noteId, notePath, subject);
+ }
}
public void moveNote(String noteId,
@@ -235,33 +235,37 @@ public class NoteManager {
throw new IOException("No metadata found for this note: " + noteId);
}
- NoteTree tree = this.noteTree;
- if (!isNotePathAvailable(tree, newNotePath)) {
- throw new NotePathAlreadyExistsException("Note '" + newNotePath + "'
existed");
- }
-
- // move the old NoteNode from notePath to newNotePath
- String notePath = tree.notesInfo.get(noteId);
- NoteNode noteNode = getNoteNode(tree, notePath);
- noteNode.getParent().removeNote(getNoteName(notePath));
- noteNode.setNotePath(newNotePath);
- String newParent = getFolderName(newNotePath);
- Folder newFolder = getOrCreateFolder(tree, newParent);
- newFolder.addNoteNode(noteNode);
-
- // update noteInfo mapping
- tree.notesInfo.put(noteId, newNotePath);
-
- // update notebookrepo
- this.notebookRepo.move(noteId, notePath, newNotePath, subject);
+ String notePath;
+ synchronized (this) {
+ NoteTree tree = this.noteTree;
+ if (!isNotePathAvailable(tree, newNotePath)) {
+ throw new NotePathAlreadyExistsException("Note '" + newNotePath + "'
existed");
+ }
- // Update path of the note
- if (!StringUtils.equals(notePath, newNotePath)) {
- processNote(noteId,
- note -> {
- note.setPath(newNotePath);
- return null;
- });
+ // move the old NoteNode from notePath to newNotePath
+ notePath = tree.notesInfo.get(noteId);
+ NoteNode noteNode = getNoteNode(tree, notePath);
+ noteNode.getParent().removeNote(getNoteName(notePath));
+ noteNode.setNotePath(newNotePath);
+ String newParent = getFolderName(newNotePath);
+ Folder newFolder = getOrCreateFolder(tree, newParent);
+ newFolder.addNoteNode(noteNode);
+
+ // update noteInfo mapping
+ tree.notesInfo.put(noteId, newNotePath);
+
+ // update notebookrepo
+ this.notebookRepo.move(noteId, notePath, newNotePath, subject);
+
+ // Update path of the note. Access the cache directly to avoid the
readLock and the
+ // disk load that processNote would add while we hold this monitor. The
reverse edge
+ // via noteCache.putNote() -> LRU eviction is safe: NoteCache only ever
tryLock()s.
+ if (!StringUtils.equals(notePath, newNotePath)) {
+ Note cachedNote = noteCache.getNote(noteId);
+ if (cachedNote != null) {
+ cachedNote.setPath(newNotePath);
+ }
+ }
}
// save note if note name is changed, because we need to update the note
field in note json.
@@ -270,7 +274,19 @@ public class NoteManager {
if (!StringUtils.equals(oldNoteName, newNoteName)) {
processNote(noteId,
note -> {
- this.notebookRepo.save(note, subject);
+ // null when the noteId already left the mapping, e.g. a concurrent
remove.
+ if (note == null) {
+ return null;
+ }
+ // newNotePath was fixed at method entry, so re-read the current
path and save
+ // it under the same monitor to keep a concurrent move out of the
gap.
+ synchronized (this) {
+ String currentPath = this.noteTree.notesInfo.get(noteId);
+ if (currentPath != null) {
+ note.setPath(currentPath);
+ saveNote(note, subject);
+ }
+ }
return null;
});
}
@@ -279,20 +295,21 @@ public class NoteManager {
public void moveFolder(String folderPath,
String newFolderPath,
AuthenticationInfo subject) throws IOException {
-
- // update notebookrepo
- this.notebookRepo.move(folderPath, newFolderPath, subject);
-
- // update filesystem tree
- NoteTree tree = this.noteTree;
- Folder folder = getFolder(tree, folderPath);
- folder.getParent().removeFolder(folder.getName(), subject);
- Folder newFolder = getOrCreateFolder(tree, newFolderPath);
- newFolder.getParent().addFolder(newFolder.getName(), folder);
-
- // update notesInfo
- for (NoteInfo noteInfo : folder.getNoteInfoRecursively()) {
- tree.notesInfo.put(noteInfo.getId(), noteInfo.getPath());
+ synchronized (this) {
+ // update notebookrepo
+ this.notebookRepo.move(folderPath, newFolderPath, subject);
+
+ // update filesystem tree
+ NoteTree tree = this.noteTree;
+ Folder folder = getFolder(tree, folderPath);
+ folder.getParent().removeFolder(folder.getName(), subject);
+ Folder newFolder = getOrCreateFolder(tree, newFolderPath);
+ newFolder.getParent().addFolder(newFolder.getName(), folder);
+
+ // update notesInfo
+ for (NoteInfo noteInfo : folder.getNoteInfoRecursively()) {
+ tree.notesInfo.put(noteInfo.getId(), noteInfo.getPath());
+ }
}
}
diff --git
a/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/NoteManagerMoveResaveRaceTest.java
b/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/NoteManagerMoveResaveRaceTest.java
new file mode 100644
index 0000000000..36e05fac33
--- /dev/null
+++
b/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/NoteManagerMoveResaveRaceTest.java
@@ -0,0 +1,196 @@
+/*
+ * 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.notebook;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import org.apache.zeppelin.conf.ZeppelinConfiguration;
+import org.apache.zeppelin.conf.ZeppelinConfiguration.ConfVars;
+import org.apache.zeppelin.interpreter.InterpreterFactory;
+import org.apache.zeppelin.interpreter.InterpreterSettingManager;
+import org.apache.zeppelin.notebook.repo.NotebookRepo;
+import org.apache.zeppelin.notebook.repo.VFSNotebookRepoWithGetGate;
+import org.apache.zeppelin.storage.ConfigStorage;
+import org.apache.zeppelin.user.AuthenticationInfo;
+import org.apache.zeppelin.user.Credentials;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Reproduction test for ZEPPELIN-5858. {@link NoteManager#moveNote} only
re-saves a note (to
+ * refresh the {@code path} field baked into its JSON) when the move changes
the note's leaf
+ * name, and it re-saves using the destination path that was passed into that
specific
+ * {@code moveNote} call, captured before the (possibly slow) reload from
{@link NotebookRepo}.
+ * If a second {@code moveNote} call for the same note (with the same leaf
name, so it takes no
+ * re-save path of its own) completes while the first call is still reloading
the note, the
+ * first call resumes and saves the note back at its own, now-stale
destination path -- leaving
+ * behind two {@code .zpln} files for the same noteId.
+ *
+ * <p>The scenario is pinned deterministically with {@link
VFSNotebookRepoWithGetGate}, which
+ * parks the reloading {@code get()} call after it has read the note from
disk, and with the
+ * note cache threshold lowered to 1 (evicting the target note via a filler
note) so the reload
+ * actually happens.
+ */
+class NoteManagerMoveResaveRaceTest {
+
+ private static final String DEFAULT_INTERPRETER_GROUP = "test";
+ private static final long JOIN_TIMEOUT_MILLIS = 30_000L;
+ private static final long GATE_ARRIVAL_TIMEOUT_SECONDS = 30L;
+
+ private File notebookDir;
+ private Notebook notebook;
+ private NoteManager noteManager;
+ private VFSNotebookRepoWithGetGate notebookRepo;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ notebookDir =
Files.createTempDirectory("notebookDir").toAbsolutePath().toFile();
+ ZeppelinConfiguration zConf = ZeppelinConfiguration.load();
+
zConf.setProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_NOTEBOOK_DIR.getVarName(),
+ notebookDir.getAbsolutePath());
+ // Must be set before NoteManager is constructed, since NoteCache reads
the threshold once
+ // at construction time.
+ zConf.setProperty(ConfVars.ZEPPELIN_NOTE_CACHE_THRESHOLD.getVarName(),
"1");
+
+ NoteParser noteParser = new GsonNoteParser(zConf);
+ ConfigStorage storage = ConfigStorage.createConfigStorage(zConf);
+ notebookRepo = new VFSNotebookRepoWithGetGate();
+ notebookRepo.init(zConf, noteParser);
+
+ InterpreterSettingManager mockInterpreterSettingManager =
mock(InterpreterSettingManager.class);
+ InterpreterFactory mockInterpreterFactory = mock(InterpreterFactory.class);
+ Credentials credentials = new Credentials();
+ noteManager = new NoteManager(notebookRepo, zConf);
+ AuthorizationService authorizationService =
+ new AuthorizationService(noteManager, zConf, storage);
+ notebook =
+ new Notebook(
+ zConf,
+ authorizationService,
+ notebookRepo,
+ noteManager,
+ mockInterpreterFactory,
+ mockInterpreterSettingManager,
+ credentials,
+ null);
+ notebook.initNotebook();
+ notebook.waitForFinishInit(1, TimeUnit.MINUTES);
+ }
+
+ @AfterEach
+ void tearDown() {
+ notebookDir.delete();
+ }
+
+ /**
+ * Given a note evicted from the (threshold=1) note cache, when a second,
unrelated-looking
+ * {@code moveNote} call (same leaf name, so no re-save of its own) runs to
completion while
+ * the first {@code moveNote} call's re-save is still reloading the note
from the repo, then
+ * the first call must not resurrect a {@code .zpln} file at its own,
now-stale destination.
+ */
+ @Test
+ void testConcurrentMoveNoteResaveRace() throws Exception {
+ String noteId = notebook.createNote(
+ "/folder_0/note", DEFAULT_INTERPRETER_GROUP,
AuthenticationInfo.ANONYMOUS, true);
+
+ // A filler note pushes the target note out of the (threshold=1) cache,
forcing the re-save
+ // path in moveNote to reload it from the repo.
+ notebook.createNote(
+ "/filler", DEFAULT_INTERPRETER_GROUP, AuthenticationInfo.ANONYMOUS,
true);
+ assertEquals(1, noteManager.getCacheSize(),
+ "creating the filler note should have evicted the target note from the
cache; "
+ + "the race scenario depends on a cache miss during moveNote's
re-save");
+
+ notebookRepo.armGate();
+
+ List<Throwable> thread1Errors = Collections.synchronizedList(new
ArrayList<>());
+ List<Throwable> thread2Errors = Collections.synchronizedList(new
ArrayList<>());
+
+ // Thread 1: rename note -> renamed. Leaf name changes, so moveNote
reloads (cache miss)
+ // and parks inside the gated get() call, having already read the (still
current) note
+ // path from disk.
+ Thread thread1 = new Thread(() -> {
+ try {
+ notebook.moveNote(noteId, "/folder_1/renamed",
AuthenticationInfo.ANONYMOUS);
+ } catch (Throwable t) {
+ thread1Errors.add(t);
+ }
+ }, "move-note-race-thread-1");
+ thread1.start();
+
+ assertTrue(
+ notebookRepo.awaitArrival(GATE_ARRIVAL_TIMEOUT_SECONDS,
TimeUnit.SECONDS),
+ "Thread 1's gated get() call never arrived. The scenario did not pin
as expected: "
+ + "either the target note was not evicted from the cache, or
moveNote's re-save "
+ + "path was not entered.");
+
+ // Thread 2: rename renamed -> renamed (different folder, same leaf name),
while thread 1
+ // is parked. Leaf name is unchanged, so this move takes no re-save path
of its own and
+ // runs to completion using only the (fast) synchronized block in moveNote.
+ Thread thread2 = new Thread(() -> {
+ try {
+ notebook.moveNote(noteId, "/folder_2/renamed",
AuthenticationInfo.ANONYMOUS);
+ } catch (Throwable t) {
+ thread2Errors.add(t);
+ }
+ }, "move-note-race-thread-2");
+ thread2.start();
+ thread2.join(JOIN_TIMEOUT_MILLIS);
+ assertFalse(thread2.isAlive(), "Thread 2's moveNote did not finish within
the timeout");
+
+ // Only now let thread 1 resume: it will save the reloaded note back at
its own, stale
+ // destination path ("/folder_1/renamed"), even though thread 2 already
moved the note to
+ // "/folder_2/renamed".
+ notebookRepo.release();
+ thread1.join(JOIN_TIMEOUT_MILLIS);
+ assertFalse(thread1.isAlive(), "Thread 1's moveNote did not finish within
the timeout");
+
+ assertTrue(thread1Errors.isEmpty(), () -> "Thread 1 threw: " +
thread1Errors);
+ assertTrue(thread2Errors.isEmpty(), () -> "Thread 2 threw: " +
thread2Errors);
+
+ List<String> zplnFilesForNote = findZplnFilesForNote(noteId);
+ assertEquals(1, zplnFilesForNote.size(),
+ () -> "Expected exactly one .zpln file for note " + noteId + ", but
found: "
+ + zplnFilesForNote);
+ }
+
+ private List<String> findZplnFilesForNote(String noteId) throws IOException {
+ Path notebookPath = notebookDir.toPath();
+ try (Stream<Path> paths = Files.walk(notebookPath)) {
+ return paths
+ .filter(p -> p.toString().endsWith("_" + noteId + ".zpln"))
+ .map(p -> notebookPath.relativize(p).toString())
+ .collect(Collectors.toList());
+ }
+ }
+}
diff --git
a/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/repo/VFSNotebookRepoWithDelay.java
b/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/repo/VFSNotebookRepoWithDelay.java
new file mode 100644
index 0000000000..14fa241e36
--- /dev/null
+++
b/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/repo/VFSNotebookRepoWithDelay.java
@@ -0,0 +1,82 @@
+/*
+ * 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.notebook.repo;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.vfs2.FileObject;
+import org.apache.commons.vfs2.NameScope;
+import org.apache.zeppelin.conf.ZeppelinConfiguration.ConfVars;
+import org.apache.zeppelin.notebook.Note;
+import org.apache.zeppelin.user.AuthenticationInfo;
+
+/**
+ * Test-only subclass of {@link VFSNotebookRepo} that injects an artificial
delay after the
+ * destination file name has been resolved in {@code save()} and after {@code
move()} starts.
+ * This reproduces the ZEPPELIN-5858 moveNote/saveNote race condition: a
concurrent move
+ * (rename) and save on the same note can both write a {@code
{oldPath}_{noteId}.zpln} and a
+ * {@code {newPath}_{noteId}.zpln} file, leaving a duplicated noteId in the
repo.
+ */
+public class VFSNotebookRepoWithDelay extends VFSNotebookRepo {
+
+ private final long delayInMillis;
+
+ public VFSNotebookRepoWithDelay(long delayInMillis) {
+ this.delayInMillis = delayInMillis;
+ }
+
+ @Override
+ public synchronized void save(Note note, AuthenticationInfo subject) throws
IOException {
+ // write to tmp file first, then rename it to the
{note_name}_{note_id}.zpln
+ FileObject noteJson = rootNotebookFileObject.resolveFile(
+ buildNoteTempFileName(note), NameScope.DESCENDENT);
+ OutputStream out = null;
+ try {
+ out = noteJson.getContent().getOutputStream(false);
+
IOUtils.write(note.toJson().getBytes(zConf.getString(ConfVars.ZEPPELIN_ENCODING)),
out);
+ } finally {
+ if (out != null) {
+ out.close();
+ }
+ }
+ // Destination file name is captured before the delay, simulating a
network round trip
+ // that happens after the note path has already been read. This ordering
is the essence
+ // of the race: capturing after the delay would not reproduce it.
+ String noteFileName = buildNoteFileName(note);
+ delay();
+ noteJson.moveTo(rootNotebookFileObject.resolveFile(noteFileName,
NameScope.DESCENDENT));
+ }
+
+ @Override
+ public void move(String noteId, String notePath, String newNotePath,
+ AuthenticationInfo subject) throws IOException {
+ // Delay at the start simulates a slow remote repo, widening the window
for a concurrent
+ // save to race with this move.
+ delay();
+ super.move(noteId, notePath, newNotePath, subject);
+ }
+
+ private void delay() {
+ try {
+ Thread.sleep(delayInMillis);
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ }
+ }
+}
diff --git
a/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/repo/VFSNotebookRepoWithGetGate.java
b/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/repo/VFSNotebookRepoWithGetGate.java
new file mode 100644
index 0000000000..66524847ea
--- /dev/null
+++
b/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/repo/VFSNotebookRepoWithGetGate.java
@@ -0,0 +1,86 @@
+/*
+ * 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.notebook.repo;
+
+import java.io.IOException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.zeppelin.notebook.Note;
+import org.apache.zeppelin.user.AuthenticationInfo;
+
+/**
+ * Test-only subclass of {@link VFSNotebookRepo} that parks the first {@code
get()} call after
+ * arming, once the note has already been read from disk. This reproduces the
reload path taken
+ * by {@code NoteManager#moveNote} when the re-save block (leaf name changed)
misses the note
+ * cache: {@code loadAndProcessNote} calls {@code NotebookRepo#get()} to
reload the note before
+ * re-saving it at the (possibly stale) target path passed into the outer
{@code moveNote} call.
+ * Parking here, after the disk read, lets a second, concurrent {@code
moveNote} call for the
+ * same note run to completion (including its own re-save skip, when the leaf
name did not
+ * change) before the parked call resumes and saves using its now-stale
destination path.
+ */
+public class VFSNotebookRepoWithGetGate extends VFSNotebookRepo {
+
+ private static final long GATE_SELF_TIMEOUT_SECONDS = 30;
+
+ private final AtomicBoolean armed = new AtomicBoolean(false);
+ private volatile CountDownLatch arrivedLatch;
+ private volatile CountDownLatch releaseLatch;
+
+ /**
+ * Arm the gate. Only the next {@code get()} call parks; every call
afterwards passes
+ * through untouched, so filler note loads and repeated reads do not get
caught by mistake.
+ */
+ public void armGate() {
+ arrivedLatch = new CountDownLatch(1);
+ releaseLatch = new CountDownLatch(1);
+ armed.set(true);
+ }
+
+ /**
+ * Wait for the gated {@code get()} call to arrive and park. Returns false,
instead of
+ * blocking forever, if it never arrives within the timeout so the caller
can fail the test
+ * with a clear message rather than hang.
+ */
+ public boolean awaitArrival(long timeout, TimeUnit unit) throws
InterruptedException {
+ return arrivedLatch.await(timeout, unit);
+ }
+
+ /**
+ * Let the parked {@code get()} call resume and return to its caller.
+ */
+ public void release() {
+ releaseLatch.countDown();
+ }
+
+ @Override
+ public Note get(String noteId, String notePath, AuthenticationInfo subject)
throws IOException {
+ Note note = super.get(noteId, notePath, subject);
+ if (armed.compareAndSet(true, false)) {
+ arrivedLatch.countDown();
+ try {
+ // Self-timeout so a test bug (forgetting to call release()) fails
fast instead of
+ // hanging the build forever.
+ releaseLatch.await(GATE_SELF_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ return note;
+ }
+}
diff --git
a/zeppelin-server/src/test/java/org/apache/zeppelin/service/NotebookServiceRaceConditionTest.java
b/zeppelin-server/src/test/java/org/apache/zeppelin/service/NotebookServiceRaceConditionTest.java
new file mode 100644
index 0000000000..2affa77690
--- /dev/null
+++
b/zeppelin-server/src/test/java/org/apache/zeppelin/service/NotebookServiceRaceConditionTest.java
@@ -0,0 +1,164 @@
+/*
+ * 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.service;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import org.apache.zeppelin.conf.ZeppelinConfiguration;
+import org.apache.zeppelin.interpreter.InterpreterFactory;
+import org.apache.zeppelin.interpreter.InterpreterSettingManager;
+import org.apache.zeppelin.notebook.AuthorizationService;
+import org.apache.zeppelin.notebook.GsonNoteParser;
+import org.apache.zeppelin.notebook.NoteManager;
+import org.apache.zeppelin.notebook.NoteParser;
+import org.apache.zeppelin.notebook.Notebook;
+import org.apache.zeppelin.notebook.repo.NotebookRepo;
+import org.apache.zeppelin.notebook.repo.VFSNotebookRepoWithDelay;
+import org.apache.zeppelin.notebook.scheduler.NoSchedulerService;
+import org.apache.zeppelin.storage.ConfigStorage;
+import org.apache.zeppelin.user.AuthenticationInfo;
+import org.apache.zeppelin.user.Credentials;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Reproduction test for ZEPPELIN-5858: a concurrent 'move' (rename) and
'save' (insert
+ * paragraph) on the same note can race in the notebook repo, leaving two
{@code .zpln} files
+ * for the same noteId (old path + new path) behind. {@link
VFSNotebookRepoWithDelay} injects
+ * an artificial delay to widen the race window.
+ */
+class NotebookServiceRaceConditionTest {
+
+ private static NotebookService notebookService;
+
+ private File notebookDir;
+ private Notebook notebook;
+ private NotebookRepo notebookRepo;
+ private ServiceContext context =
+ new ServiceContext(AuthenticationInfo.ANONYMOUS, new HashSet<>());
+
+ private ServiceCallback callback = mock(ServiceCallback.class);
+
+ @BeforeEach
+ void setUp() throws Exception {
+ notebookDir =
Files.createTempDirectory("notebookDir").toAbsolutePath().toFile();
+ ZeppelinConfiguration zConf = ZeppelinConfiguration.load();
+
zConf.setProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_NOTEBOOK_DIR.getVarName(),
+ notebookDir.getAbsolutePath());
+ NoteParser noteParser = new GsonNoteParser(zConf);
+ ConfigStorage storage = ConfigStorage.createConfigStorage(zConf);
+ notebookRepo = new VFSNotebookRepoWithDelay(5000L);
+ notebookRepo.init(zConf, noteParser);
+
+ InterpreterSettingManager mockInterpreterSettingManager =
mock(InterpreterSettingManager.class);
+ InterpreterFactory mockInterpreterFactory = mock(InterpreterFactory.class);
+ Credentials credentials = new Credentials();
+ NoteManager noteManager = new NoteManager(notebookRepo, zConf);
+ AuthorizationService authorizationService =
+ new AuthorizationService(noteManager, zConf, storage);
+ notebook =
+ new Notebook(
+ zConf,
+ authorizationService,
+ notebookRepo,
+ noteManager,
+ mockInterpreterFactory,
+ mockInterpreterSettingManager,
+ credentials,
+ null);
+ notebook.initNotebook();
+ notebook.waitForFinishInit(1, TimeUnit.MINUTES);
+ notebookService =
+ new NotebookService(
+ notebook, authorizationService, zConf, new NoSchedulerService());
+ }
+
+ @AfterEach
+ void tearDown() {
+ notebookDir.delete();
+ }
+
+ /**
+ * Concurrent 'insertParagraph' (save) and 'renameNote' (move) on the same
note. The delayed
+ * repo widens the window between reading a note's path and writing to it,
so both operations
+ * can write a {@code .zpln} file for the same noteId: one at the old path,
one at the new
+ * path. Thread 2 starts the move first (delay simulates a slow remote
write); thread 1 saves
+ * shortly after, while the move is still in flight.
+ */
+ @Test
+ void testConcurrentMoveAndSave() throws IOException, InterruptedException {
+ // given a note
+ String noteId = notebookService.createNote("/folder_1/note", "test", true,
context, callback);
+
+ // when executing 'move' (renameNote) and 'save' (insertParagraph)
concurrently
+ CountDownLatch latch = new CountDownLatch(2);
+ ExecutorService threadPool = Executors.newFixedThreadPool(2);
+ threadPool.execute(() -> {
+ try {
+ // ensure we 'save' after 'move' has started processing, but before
'move' has finished
+ Thread.sleep(1000L);
+ notebookService.insertParagraph(noteId, 1, Collections.emptyMap(),
context, callback);
+ latch.countDown();
+ } catch (IOException | InterruptedException ex) {
+ // ignore
+ }
+ });
+ threadPool.execute(() -> {
+ try {
+ notebookService.renameNote(noteId, "/folder_2/note", false, context,
callback);
+ latch.countDown();
+ } catch (IOException ex) {
+ // ignore
+ }
+ });
+ assertTrue(latch.await(100, TimeUnit.SECONDS));
+ threadPool.shutdown();
+
+ // then only a single .zpln file exists for this note under notebookDir
+ List<String> zplnFiles = findZplnFiles();
+ assertEquals(1, zplnFiles.size(),
+ () -> "Expected exactly one .zpln file, but found: " + zplnFiles);
+ }
+
+ private List<String> findZplnFiles() throws IOException {
+ Path notebookPath = notebookDir.toPath();
+ try (Stream<Path> paths = Files.walk(notebookPath)) {
+ return paths
+ .filter(p -> p.toString().endsWith(".zpln"))
+ .map(p -> notebookPath.relativize(p).toString())
+ .collect(Collectors.toList());
+ }
+ }
+}
