This is an automated email from the ASF dual-hosted git repository.
jongyoul 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 d8c43cb156 [ZEPPELIN-6579] Make notebook tree reload safe for
concurrent note operations
d8c43cb156 is described below
commit d8c43cb1561bddc296f9a158eafe8477c488625f
Author: dae won <[email protected]>
AuthorDate: Sat Aug 1 13:16:51 2026 +0900
[ZEPPELIN-6579] Make notebook tree reload safe for concurrent note
operations
### What is this PR for?
`NoteManager` locates a note through two separate pieces of state:
`notesInfo` maps a note id to its path, and `root` holds the folder tree that
the path is walked against. A lookup uses both in sequence, so the two have to
agree.
`reloadNotes()` replaced them one at a time:
```java
public void reloadNotes() throws IOException {
this.root = new Folder("/", notebookRepo, noteCache, zConf); // (1) tree
becomes empty
this.trash = this.root.getOrCreateFolder(TRASH_FOLDER);
init(); // (2) new
mapping, (3) refill tree
}
```
Neither field is `volatile` and nothing is held while they are swapped, so
a concurrent `processNote()` can observe a mapping and a tree that belong to
different generations:
| time | reloading thread | note request thread | state |
|---|---|---|---|
| t1 | installs an empty tree | | mapping: old (complete) / tree: **empty**
|
| t2 | | `notesInfo.containsKey(noteId)` passes | the id is still in the
old mapping |
| t3 | | walks the path in the tree, finds nothing | **throws** |
| t4 | installs the new mapping | | |
| t5 | refills the tree, one note at a time | | notes not inserted yet
still fail |
The guard in `processNote()` only checks `notesInfo`, so it passes and the
failure surfaces one line later in `getNoteNode()`:
```
java.io.IOException: Can not find note: /E2E_TEST_FOLDER/TestNotebook_...
at org.apache.zeppelin.notebook.NoteManager.getNoteNode
at org.apache.zeppelin.notebook.NoteManager.processNote
at org.apache.zeppelin.rest.NotebookRestApi.updateParagraph
```
`IOException` is not mapped to a specific status, so
`WebApplicationExceptionMapper` turns it into **HTTP 500** for a note that was
never removed. Everything that goes through `processNote()` is affected:
reading a note, updating a paragraph, creating, deleting and moving notes, and
listing the notebook.
This PR holds the tree, the trash folder and the mapping in one immutable
`NoteTree` and publishes it with a single `volatile` write. `buildNoteTree()`
fills the new tree locally and returns it; only then is it assigned. The
tree-walking helpers (`getNoteNode`, `getFolder`, `getOrCreateFolder`,
`isNotePathAvailable`) take the tree as a parameter, and callers that need both
pieces of state read the reference once, so a lookup resolves the mapping and
the tree against the same generatio [...]
### Scope and related issues
**#5325** (`[ZEPPELIN-5858]`) is open against the same class and
restructures `removeNote`, `moveNote` and `moveFolder` with `synchronized
(this)`. It targets a different race (two mutators duplicating a note) and its
monitor does not cover `reloadNotes()`, so neither change subsumes the other.
Whichever merges second will need a rebase.
### What type of PR is it?
Bug Fix
### Todos
* [x] - Build the new tree, trash folder and mapping in `buildNoteTree()`
before publishing them
* [x] - Hold the three in an immutable `NoteTree` published through a
single `volatile` write
* [x] - Pass the tree into the tree-walking helpers so one lookup uses one
generation
* [x] - Add a regression test that reloads while other threads read notes
* [x] - Confirm the test fails without the fix and passes with it
### What is the Jira issue?
* [ZEPPELIN-6579](https://issues.apache.org/jira/browse/ZEPPELIN-6579)
### How should this be tested?
New test `NoteManagerTest#testConcurrentReloadAndProcessNote`: it saves 50
notes, then runs `reloadNotes()` in a loop on one thread while four threads
keep calling `processNote()` for every note, and asserts that no lookup fails
or returns nothing.
```bash
export JAVA_HOME=$(/usr/libexec/java_home -v 11)
./mvnw package -pl zeppelin-server --am -Dtest=NoteManagerTest
-DfailIfNoTests=false
```
Result with the fix: `Tests run: 7, Failures: 0, Errors: 0`.
Reverting only the production change makes the new test fail on every
reader thread with `java.io.IOException: Can not find note: /prod/note_0`
thrown from `NoteManager.getNoteNode` via `NoteManager.processNote`, which is
the stack from the ticket; with the fix it passes.
Also run, to cover the callers of the reload path:
```bash
./mvnw package -pl zeppelin-server --am \
-Dtest='NotebookTest#testReloadAllNotes+testReloadAndSetInterpreter'
-DfailIfNoTests=false
```
Result: `Tests run: 2, Failures: 0, Errors: 0`.
Not verified locally: the full `NotebookTest` and `NotebookServerTest`
classes, which start real remote interpreter processes and time out in my
environment, and `NotebookRepoSyncTest`. Those are left to CI.
### Screenshots (if appropriate)
N/A
### Questions:
* Does the license files need to update? No
* Is there breaking changes for older versions? No
* Does this needs documentation? No
Closes #5357 from big-cir/ZEPPELIN-6579.
Signed-off-by: Jongyoul Lee <[email protected]>
---
.../org/apache/zeppelin/notebook/NoteManager.java | 155 +++++++++++++--------
.../apache/zeppelin/notebook/NoteManagerTest.java | 59 ++++++++
2 files changed, 158 insertions(+), 56 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 fffb49d8ca..0635fde994 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
@@ -59,54 +59,62 @@ import io.micrometer.core.instrument.Tags;
public class NoteManager {
private static final Logger LOGGER =
LoggerFactory.getLogger(NoteManager.class);
public static final String TRASH_FOLDER = "~Trash";
- private Folder root;
- private Folder trash;
-
private NotebookRepo notebookRepo;
private NoteCache noteCache;
- // noteId -> notePath
- private Map<String, String> notesInfo;
private final ZeppelinConfiguration zConf;
+ /**
+ * The folder tree and the noteId -> notePath mapping. They are held
together in one
+ * immutable reference so that a reload publishes both at once and
concurrent note
+ * operations never observe a tree and a mapping that belong to different
generations.
+ */
+ private volatile NoteTree noteTree;
+
@Inject
public NoteManager(NotebookRepo notebookRepo, ZeppelinConfiguration zConf)
throws IOException {
this.zConf = zConf;
this.notebookRepo = notebookRepo;
this.noteCache = new NoteCache(zConf.getNoteCacheThreshold());
- this.root = new Folder("/", notebookRepo, noteCache, zConf);
- this.trash = this.root.getOrCreateFolder(TRASH_FOLDER);
- init();
+ this.noteTree = buildNoteTree();
}
- // build the tree structure of notes
- private void init() throws IOException {
- this.notesInfo =
notebookRepo.list(AuthenticationInfo.ANONYMOUS).values().stream()
- .collect(Collectors.toConcurrentMap(NoteInfo::getId,
NoteInfo::getPath));
- this.notesInfo.entrySet().stream()
- .forEach(entry ->
- {
- try {
- addOrUpdateNoteNode(new NoteInfo(entry.getKey(),
entry.getValue()));
- } catch (IOException e) {
- LOGGER.warn(e.getMessage());
- }
- });
+ /**
+ * Build the tree structure of notes from the NotebookRepo. The tree is
fully populated
+ * before it is returned, and it is not reachable by other threads until the
caller
+ * publishes it to {@link #noteTree}.
+ */
+ private NoteTree buildNoteTree() throws IOException {
+ Folder newRoot = new Folder("/", notebookRepo, noteCache, zConf);
+ Folder newTrash = newRoot.getOrCreateFolder(TRASH_FOLDER);
+ Map<String, String> newNotesInfo =
+ notebookRepo.list(AuthenticationInfo.ANONYMOUS).values().stream()
+ .collect(Collectors.toConcurrentMap(NoteInfo::getId,
NoteInfo::getPath));
+ NoteTree newNoteTree = new NoteTree(newRoot, newTrash, newNotesInfo);
+ for (Map.Entry<String, String> entry : newNotesInfo.entrySet()) {
+ try {
+ addOrUpdateNoteNode(newNoteTree, new NoteInfo(entry.getKey(),
entry.getValue()), false);
+ } catch (IOException e) {
+ LOGGER.warn(e.getMessage());
+ }
+ }
+ return newNoteTree;
}
public Map<String, String> getNotesInfo() {
- return notesInfo;
+ return this.noteTree.notesInfo;
}
/**
+ * Rebuild the notebook metadata from the NotebookRepo. The new tree is
built completely
+ * before it replaces the current one, so a concurrent note operation sees
either the
+ * previous tree or the new one, never a partially rebuilt tree.
*
* @throws IOException
*/
public void reloadNotes() throws IOException {
- this.root = new Folder("/", notebookRepo, noteCache, zConf);
- this.trash = this.root.getOrCreateFolder(TRASH_FOLDER);
- init();
+ this.noteTree = buildNoteTree();
}
/**
@@ -117,15 +125,16 @@ public class NoteManager {
return this.noteCache.getSize();
}
- private void addOrUpdateNoteNode(NoteInfo noteInfo, boolean checkDuplicates)
throws IOException {
+ private void addOrUpdateNoteNode(NoteTree tree, NoteInfo noteInfo, boolean
checkDuplicates)
+ throws IOException {
String notePath = noteInfo.getPath();
- if (checkDuplicates && !isNotePathAvailable(notePath)) {
+ if (checkDuplicates && !isNotePathAvailable(tree, notePath)) {
throw new NotePathAlreadyExistsException("Note '" + notePath + "'
existed");
}
String[] tokens = notePath.split("/");
- Folder curFolder = root;
+ Folder curFolder = tree.root;
for (int i = 0; i < tokens.length - 1; ++i) {
if (!StringUtils.isBlank(tokens[i])) {
curFolder = curFolder.getOrCreateFolder(tokens[i]);
@@ -133,11 +142,7 @@ public class NoteManager {
}
curFolder.addNote(tokens[tokens.length -1], noteInfo);
- this.notesInfo.put(noteInfo.getId(), noteInfo.getPath());
- }
-
- private void addOrUpdateNoteNode(NoteInfo noteInfo) throws IOException {
- addOrUpdateNoteNode(noteInfo, false);
+ tree.notesInfo.put(noteInfo.getId(), noteInfo.getPath());
}
/**
@@ -182,7 +187,7 @@ public class NoteManager {
if (note.isRemoved()) {
LOGGER.warn("Try to save note: {} when it is removed", note.getId());
} else {
- addOrUpdateNoteNode(new NoteInfo(note));
+ 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.
@@ -193,7 +198,7 @@ public class NoteManager {
}
public void addNote(Note note, AuthenticationInfo subject) throws
IOException {
- addOrUpdateNoteNode(new NoteInfo(note), true);
+ addOrUpdateNoteNode(this.noteTree, new NoteInfo(note), true);
noteCache.putNote(note);
}
@@ -215,8 +220,9 @@ public class NoteManager {
* @throws IOException
*/
public void removeNote(String noteId, AuthenticationInfo subject) throws
IOException {
- String notePath = this.notesInfo.remove(noteId);
- Folder folder = getOrCreateFolder(getFolderName(notePath));
+ 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);
@@ -229,21 +235,22 @@ public class NoteManager {
throw new IOException("No metadata found for this note: " + noteId);
}
- if (!isNotePathAvailable(newNotePath)) {
+ NoteTree tree = this.noteTree;
+ if (!isNotePathAvailable(tree, newNotePath)) {
throw new NotePathAlreadyExistsException("Note '" + newNotePath + "'
existed");
}
// move the old NoteNode from notePath to newNotePath
- String notePath = this.notesInfo.get(noteId);
- NoteNode noteNode = getNoteNode(notePath);
+ 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(newParent);
+ Folder newFolder = getOrCreateFolder(tree, newParent);
newFolder.addNoteNode(noteNode);
// update noteInfo mapping
- this.notesInfo.put(noteId, newNotePath);
+ tree.notesInfo.put(noteId, newNotePath);
// update notebookrepo
this.notebookRepo.move(noteId, notePath, newNotePath, subject);
@@ -277,14 +284,15 @@ public class NoteManager {
this.notebookRepo.move(folderPath, newFolderPath, subject);
// update filesystem tree
- Folder folder = getFolder(folderPath);
+ NoteTree tree = this.noteTree;
+ Folder folder = getFolder(tree, folderPath);
folder.getParent().removeFolder(folder.getName(), subject);
- Folder newFolder = getOrCreateFolder(newFolderPath);
+ Folder newFolder = getOrCreateFolder(tree, newFolderPath);
newFolder.getParent().addFolder(newFolder.getName(), folder);
// update notesInfo
for (NoteInfo noteInfo : folder.getNoteInfoRecursively()) {
- notesInfo.put(noteInfo.getId(), noteInfo.getPath());
+ tree.notesInfo.put(noteInfo.getId(), noteInfo.getPath());
}
}
@@ -313,12 +321,13 @@ public class NoteManager {
this.notebookRepo.remove(folderPath, subject);
// update filesystem tree
- Folder folder = getFolder(folderPath);
+ NoteTree tree = this.noteTree;
+ Folder folder = getFolder(tree, folderPath);
List<NoteInfo> noteInfos =
folder.getParent().removeFolder(folder.getName(), subject);
// update notesInfo and evict the deleted notes from the cache, mirroring
removeNote
for (NoteInfo noteInfo : noteInfos) {
- this.notesInfo.remove(noteInfo.getId());
+ tree.notesInfo.remove(noteInfo.getId());
this.noteCache.removeNote(noteInfo.getId());
}
@@ -336,11 +345,14 @@ public class NoteManager {
*/
public <T> T processNote(String noteId, boolean reload, NoteProcessor<T>
noteProcessor)
throws IOException {
- if (this.notesInfo == null || noteId == null ||
!this.notesInfo.containsKey(noteId)) {
+ // Read the tree once, so that the mapping lookup below and the tree
traversal that
+ // follows it are both resolved against the same generation of the
metadata.
+ NoteTree tree = this.noteTree;
+ if (tree == null || noteId == null || !tree.notesInfo.containsKey(noteId))
{
return noteProcessor.process(null);
}
- String notePath = this.notesInfo.get(noteId);
- NoteNode noteNode = getNoteNode(notePath);
+ String notePath = tree.notesInfo.get(noteId);
+ NoteNode noteNode = getNoteNode(tree, notePath);
return noteNode.loadAndProcessNote(reload, noteProcessor);
}
@@ -362,8 +374,12 @@ public class NoteManager {
* @return
*/
public Folder getOrCreateFolder(String folderName) {
+ return getOrCreateFolder(this.noteTree, folderName);
+ }
+
+ private static Folder getOrCreateFolder(NoteTree tree, String folderName) {
String[] tokens = folderName.split("/");
- Folder curFolder = root;
+ Folder curFolder = tree.root;
for (int i = 0; i < tokens.length; ++i) {
if (!StringUtils.isBlank(tokens[i])) {
curFolder = curFolder.getOrCreateFolder(tokens[i]);
@@ -373,8 +389,12 @@ public class NoteManager {
}
private NoteNode getNoteNode(String notePath) throws IOException {
+ return getNoteNode(this.noteTree, notePath);
+ }
+
+ private static NoteNode getNoteNode(NoteTree tree, String notePath) throws
IOException {
String[] tokens = notePath.split("/");
- Folder curFolder = root;
+ Folder curFolder = tree.root;
for (int i = 0; i < tokens.length - 1; ++i) {
if (!StringUtils.isBlank(tokens[i])) {
curFolder = curFolder.getFolder(tokens[i]);
@@ -391,8 +411,12 @@ public class NoteManager {
}
private Folder getFolder(String folderPath) throws IOException {
+ return getFolder(this.noteTree, folderPath);
+ }
+
+ private static Folder getFolder(NoteTree tree, String folderPath) throws
IOException {
String[] tokens = folderPath.split("/");
- Folder curFolder = root;
+ Folder curFolder = tree.root;
for (int i = 0; i < tokens.length; ++i) {
if (!StringUtils.isBlank(tokens[i])) {
curFolder = curFolder.getFolder(tokens[i]);
@@ -405,7 +429,7 @@ public class NoteManager {
}
public Folder getTrashFolder() {
- return this.trash;
+ return this.noteTree.trash;
}
private String getFolderName(String notePath) {
@@ -418,9 +442,9 @@ public class NoteManager {
return notePath.substring(pos + 1);
}
- private boolean isNotePathAvailable(String notePath) {
+ private static boolean isNotePathAvailable(NoteTree tree, String notePath) {
String[] tokens = notePath.split("/");
- Folder curFolder = root;
+ Folder curFolder = tree.root;
for (int i = 0; i < tokens.length - 1; ++i) {
if (!StringUtils.isBlank(tokens[i])) {
curFolder = curFolder.getFolder(tokens[i]);
@@ -441,6 +465,25 @@ public class NoteManager {
return noteNode.getNoteId();
}
+ /**
+ * The two indexes that together locate a note: the folder tree and the
noteId -> notePath
+ * mapping. A note lookup resolves the id through the mapping and then walks
the tree, so
+ * the two must belong to the same generation. Holding them in one immutable
reference lets
+ * a reload replace both of them in a single assignment.
+ */
+ private static class NoteTree {
+ private final Folder root;
+ private final Folder trash;
+ // noteId -> notePath
+ private final Map<String, String> notesInfo;
+
+ NoteTree(Folder root, Folder trash, Map<String, String> notesInfo) {
+ this.root = root;
+ this.trash = trash;
+ this.notesInfo = notesInfo;
+ }
+ }
+
/**
* Represent one folder that could contains sub folders and note files.
*/
diff --git
a/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/NoteManagerTest.java
b/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/NoteManagerTest.java
index cb23ea8f16..eaed222f9e 100644
---
a/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/NoteManagerTest.java
+++
b/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/NoteManagerTest.java
@@ -25,15 +25,20 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -208,6 +213,60 @@ class NoteManagerTest {
threadPool.shutdown();
}
+ @Test
+ void testConcurrentReloadAndProcessNote() throws Exception {
+ int noteNum = 50, readerNum = 4, reloadRounds = 30;
+ Map<Integer, String> notes = new ConcurrentHashMap<>();
+ for (int i = 0; i < noteNum; i++) {
+ Note note = createNote(String.format("/prod/note_%s", i));
+ noteManager.saveNote(note);
+ notes.put(i, note.getId());
+ }
+
+ List<Throwable> failures = Collections.synchronizedList(new ArrayList<>());
+ AtomicBoolean reloading = new AtomicBoolean(true);
+ ExecutorService threadPool = Executors.newFixedThreadPool(readerNum + 1);
+ CountDownLatch done = new CountDownLatch(readerNum + 1);
+
+ // Reload the whole note tree repeatedly while other threads read the notes
+ threadPool.execute(() -> {
+ try {
+ for (int i = 0; i < reloadRounds; i++) {
+ noteManager.reloadNotes();
+ }
+ } catch (Throwable t) {
+ failures.add(t);
+ } finally {
+ reloading.set(false);
+ done.countDown();
+ }
+ });
+
+ for (int i = 0; i < readerNum; i++) {
+ threadPool.execute(() -> {
+ try {
+ while (reloading.get()) {
+ for (String noteId : notes.values()) {
+ assertNotNull(noteManager.processNote(noteId, note -> note),
+ "processNote() found no note for an existing noteId during
reload");
+ }
+ }
+ } catch (Throwable t) {
+ failures.add(t);
+ } finally {
+ done.countDown();
+ }
+ });
+ }
+
+ assertTrue(done.await(60, TimeUnit.SECONDS), "Concurrent reload did not
finish in time");
+ threadPool.shutdown();
+ if (!failures.isEmpty()) {
+ throw new AssertionError(failures.size()
+ + " note operation(s) failed while the note tree was being
reloaded", failures.get(0));
+ }
+ }
+
abstract class ConcurrentTask {
private ExecutorService threadPool;
private int noteNum;
