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..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 @@ -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,33 +235,37 @@ 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 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 void moveNote(String noteId, 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 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()); + } } } 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/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/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; + } +} 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()); + } + } +}