From 38e721bd6c2a523f8314116533187c39f786b26e Mon Sep 17 00:00:00 2001 From: HwangRock Date: Sun, 19 Jul 2026 16:15:32 +0900 Subject: [PATCH 1/3] [ZEPPELIN-5858] Add reproduction test for moveNote/saveNote race Port the reproduction attached to ZEPPELIN-5858 (2022) onto the current NotebookRepo API. VFSNotebookRepoWithDelay extends VFSNotebookRepo and injects a delay after save() resolves its destination file name and after move() starts, simulating slow repo I/O such as remote storage. The test runs renameNote and insertParagraph concurrently against it. The save thread derives the note file name from stale path data, so after the move completes it resurrects the note at its old path and the same noteId ends up with two .zpln files: Saving note 2MXKZT37A to folder_1/note_2MXKZT37A.zpln Move note 2MXKZT37A to /folder_2/note Saving note 2MXKZT37A to folder_1/note_2MXKZT37A.zpln -> [folder_2/note_2MXKZT37A.zpln, folder_1/note_2MXKZT37A.zpln] The test walks the notebook directory and currently fails with two .zpln files for the same noteId. --- .../repo/VFSNotebookRepoWithDelay.java | 82 +++++++++ .../NotebookServiceRaceConditionTest.java | 164 ++++++++++++++++++ 2 files changed, 246 insertions(+) create mode 100644 zeppelin-server/src/test/java/org/apache/zeppelin/notebook/repo/VFSNotebookRepoWithDelay.java create mode 100644 zeppelin-server/src/test/java/org/apache/zeppelin/service/NotebookServiceRaceConditionTest.java 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 00000000000..14fa241e363 --- /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/service/NotebookServiceRaceConditionTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/service/NotebookServiceRaceConditionTest.java new file mode 100644 index 00000000000..2affa776906 --- /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 zplnFiles = findZplnFiles(); + assertEquals(1, zplnFiles.size(), + () -> "Expected exactly one .zpln file, but found: " + zplnFiles); + } + + private List findZplnFiles() throws IOException { + Path notebookPath = notebookDir.toPath(); + try (Stream paths = Files.walk(notebookPath)) { + return paths + .filter(p -> p.toString().endsWith(".zpln")) + .map(p -> notebookPath.relativize(p).toString()) + .collect(Collectors.toList()); + } + } +} From b8a4c224c1b3d8e263c3719e8175f5c849edeb47 Mon Sep 17 00:00:00 2001 From: HwangRock Date: Sun, 19 Jul 2026 16:15:43 +0900 Subject: [PATCH 2/3] [ZEPPELIN-5858] Serialize NoteManager mutations to fix moveNote/saveNote race moveNote mutated the folder tree, the notesInfo mapping and the repo without holding any lock, while saveNote derived the target file name from the path carried by the Note object. A save racing with a move could therefore write the note back to its pre-move path, leaving the same noteId on disk twice. moveNote, removeNote and moveFolder now perform their tree, mapping and repo mutations inside the monitor saveNote already synchronizes on, so the mutual exclusion holds against the save path. moveNote updates the cached note path directly through the note cache instead of processNote: processNote acquires the note's readLock, and taking it while holding the monitor would invert the fixed lock order (readLock -> monitor) used by every save caller. Since save callers and moveNote share the cached Note instance, a save that lost the monitor to a concurrent move then derives its file name from the already-updated path. The rename-triggered resave stays outside the monitor, reuses saveNote, and realigns the reloaded note with the move target first: a note evicted from the cache is reloaded from disk at that point, and the on-disk JSON still carries the pre-move name. --- .../apache/zeppelin/notebook/NoteManager.java | 114 ++++++++++-------- 1 file changed, 63 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 0635fde994c..d07b628da01 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 void saveNote(Note note, AuthenticationInfo subject) throws IOException { 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 void saveNote(Note note) throws IOException { * @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,42 +235,53 @@ public void moveNote(String noteId, 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 instead of going + // through processNote/loadAndProcessNote, which would acquire the note's + // readLock while holding this monitor and invert the fixed lock order + // (readLock -> monitor). + 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. + // Done outside the monitor: processNote acquires the readLock and saveNote then + // acquires the monitor, matching the fixed lock order (readLock -> monitor). String oldNoteName = getNoteName(notePath); String newNoteName = getNoteName(newNotePath); if (!StringUtils.equals(oldNoteName, newNoteName)) { processNote(noteId, note -> { - this.notebookRepo.save(note, subject); + // A note evicted from the cache is reloaded from disk here, and the on-disk + // JSON still carries the pre-move name, so realign the reloaded note with + // the move target before persisting it. + note.setPath(newNotePath); + saveNote(note, subject); return null; }); } @@ -279,20 +290,21 @@ public void moveNote(String noteId, 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()); + } } } From e3f715b19a9d4e0641f407360edd324ec0cdf945 Mon Sep 17 00:00:00 2001 From: HwangRock Date: Tue, 4 Aug 2026 00:54:40 +0900 Subject: [PATCH 3/3] [ZEPPELIN-5858] Fix stale destination path in moveNote's rename resave The rename-triggered resave in moveNote read newNotePath, an argument fixed at method entry, and used it to set the reloaded note's path before saving. If a second moveNote call for the same noteId completed while the first call's resave was still reloading the note from disk, the first call resumed with a destination that was no longer current and saved the note back at that stale path, leaving two .zpln files for the same noteId. The resave now reads the note's current path from the notesInfo mapping instead of trusting the method argument, and does the read and the save inside the same monitor section so no concurrent move can land between them. It also guards against note being null, which processNote passes in when the noteId has already left the mapping (e.g. a concurrent removeNote/removeFolder). Also corrects a comment above the direct cache access in moveNote: it claimed processNote's readLock acquisition would invert the fixed lock order against the monitor, but a reverse edge (monitor -> note writeLock) already exists via saveNote -> noteCache.putNote() -> LRU eviction. That edge is harmless only because every note writeLock acquisition in NoteCache uses tryLock(), which never blocks. The comment now records this invariant instead of the inaccurate one. Adds NoteManagerMoveResaveRaceTest, which reproduces the race deterministically with VFSNotebookRepoWithGetGate by pinning the first moveNote call's reload right after it reads the note from disk. --- .../apache/zeppelin/notebook/NoteManager.java | 27 ++- .../NoteManagerMoveResaveRaceTest.java | 196 ++++++++++++++++++ .../repo/VFSNotebookRepoWithGetGate.java | 86 ++++++++ 3 files changed, 298 insertions(+), 11 deletions(-) create mode 100644 zeppelin-server/src/test/java/org/apache/zeppelin/notebook/NoteManagerMoveResaveRaceTest.java create mode 100644 zeppelin-server/src/test/java/org/apache/zeppelin/notebook/repo/VFSNotebookRepoWithGetGate.java 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 d07b628da01..c31cad72b8c 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 @@ -257,10 +257,9 @@ public void moveNote(String noteId, // update notebookrepo this.notebookRepo.move(noteId, notePath, newNotePath, subject); - // Update path of the note. Access the cache directly instead of going - // through processNote/loadAndProcessNote, which would acquire the note's - // readLock while holding this monitor and invert the fixed lock order - // (readLock -> monitor). + // 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) { @@ -270,18 +269,24 @@ public void moveNote(String noteId, } // save note if note name is changed, because we need to update the note field in note json. - // Done outside the monitor: processNote acquires the readLock and saveNote then - // acquires the monitor, matching the fixed lock order (readLock -> monitor). String oldNoteName = getNoteName(notePath); String newNoteName = getNoteName(newNotePath); if (!StringUtils.equals(oldNoteName, newNoteName)) { processNote(noteId, note -> { - // A note evicted from the cache is reloaded from disk here, and the on-disk - // JSON still carries the pre-move name, so realign the reloaded note with - // the move target before persisting it. - note.setPath(newNotePath); - saveNote(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; }); } 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 00000000000..36e05fac336 --- /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. + * + *

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 thread1Errors = Collections.synchronizedList(new ArrayList<>()); + List 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 zplnFilesForNote = findZplnFilesForNote(noteId); + assertEquals(1, zplnFilesForNote.size(), + () -> "Expected exactly one .zpln file for note " + noteId + ", but found: " + + zplnFilesForNote); + } + + private List findZplnFilesForNote(String noteId) throws IOException { + Path notebookPath = notebookDir.toPath(); + try (Stream 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/VFSNotebookRepoWithGetGate.java b/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/repo/VFSNotebookRepoWithGetGate.java new file mode 100644 index 00000000000..66524847ea3 --- /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; + } +}