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 796c8c094f [ZEPPELIN-6340] Add ZeppelinEventBus and update
NotebookServer to handle NoteRemoveEvent
796c8c094f is described below
commit 796c8c094f0324a12495a133deec255ea25be7ac
Author: SeungYoung Oh <[email protected]>
AuthorDate: Sat Oct 17 02:26:57 2026 +0900
[ZEPPELIN-6340] Add ZeppelinEventBus and update NotebookServer to handle
NoteRemoveEvent
### What is this PR for?
This PR is the first step toward migrating Zeppelin’s listener-based event
handling to an EventBus model, as outlined in the
[proposal](https://cwiki.apache.org/confluence/x/vAsGFw). It sets up the
EventBus infrastructure and migrates `NoteRemoveEvent` handling to it.
Key changes:
* Added the `ZeppelinEventBus` class
* Added RxJava 3 as a dependency of the `zeppelin-zengine` module
* Added the `zeppelin.eventbus.enabled` feature flag
* Updated `NotebookServer` to handle `NoteRemoveEvent` through
`ZeppelinEventBus` when the feature flag is enabled
This PR supersedes #5085. The original branch had fallen behind the current
codebase, and the implementation has since changed, so I opened a new PR.
### What type of PR is it?
Improvement
### Todos
* [ ] Refactor other NoteEventListener methods and additional listeners to
EventBus
### What is the Jira issue?
* Open an issue on Jira https://issues.apache.org/jira/browse/ZEPPELIN-6340
### How should this be tested?
- Run the existing test suite against the updated code
- Verify the note deletion flow through the UI
### Screenshots (if appropriate)
### Questions:
* Does the license files need to update? N
* Is there breaking changes for older versions? N
* Does this needs documentation? N
Closes #5441 from seung-00/ZEPPELIN-6340.
Signed-off-by: ChanHo Lee <[email protected]>
---
conf/zeppelin-site.xml.template | 6 ++
zeppelin-server/pom.xml | 6 ++
.../zeppelin/conf/ZeppelinConfiguration.java | 5 +-
.../org/apache/zeppelin/eventbus/EventBus.java | 34 +++++++++
.../org/apache/zeppelin/eventbus/NoOpEventBus.java | 47 +++++++++++++
.../org/apache/zeppelin/eventbus/NoteEvent.java | 28 ++++++++
.../apache/zeppelin/eventbus/NoteRemovedEvent.java | 43 ++++++++++++
.../apache/zeppelin/eventbus/ZeppelinEvent.java | 21 ++++++
.../apache/zeppelin/eventbus/ZeppelinEventBus.java | 63 +++++++++++++++++
.../org/apache/zeppelin/notebook/Notebook.java | 23 ++++--
.../org/apache/zeppelin/server/ZeppelinServer.java | 9 +++
.../org/apache/zeppelin/socket/NotebookServer.java | 59 ++++++++++++----
.../zeppelin/eventbus/ZeppelinEventBusTest.java | 82 ++++++++++++++++++++++
.../helium/HeliumApplicationFactoryTest.java | 3 +-
.../interpreter/AbstractInterpreterTest.java | 6 +-
.../org/apache/zeppelin/notebook/NotebookTest.java | 4 +-
.../notebook/repo/NotebookRepoSyncTest.java | 6 +-
.../zeppelin/service/NotebookServiceTest.java | 8 +--
.../apache/zeppelin/socket/NotebookServerTest.java | 30 ++++++++
19 files changed, 457 insertions(+), 26 deletions(-)
diff --git a/conf/zeppelin-site.xml.template b/conf/zeppelin-site.xml.template
index 6cf0e071d6..f30b346d66 100755
--- a/conf/zeppelin-site.xml.template
+++ b/conf/zeppelin-site.xml.template
@@ -872,4 +872,10 @@
<description>Directory for assistant conversations.</description>
</property>
+<property>
+ <name>zeppelin.eventbus.enabled</name>
+ <value>false</value>
+ <description>Enables the new event-driven architecture using an in-process
EventBus</description>
+</property>
+
</configuration>
diff --git a/zeppelin-server/pom.xml b/zeppelin-server/pom.xml
index 7191c0bb01..12699c9ce3 100644
--- a/zeppelin-server/pom.xml
+++ b/zeppelin-server/pom.xml
@@ -97,6 +97,12 @@
<artifactId>jakarta.inject-api</artifactId>
</dependency>
+ <dependency>
+ <groupId>io.reactivex.rxjava3</groupId>
+ <artifactId>rxjava</artifactId>
+ <version>3.1.12</version>
+ </dependency>
+
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
diff --git
a/zeppelin-server/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
b/zeppelin-server/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
index 8b161d174f..781e7d9f57 100644
---
a/zeppelin-server/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
+++
b/zeppelin-server/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
@@ -965,6 +965,8 @@ public class ZeppelinConfiguration {
return getBoolean(ConfVars.ZEPPELIN_METRIC_ENABLE_PROMETHEUS);
}
+ public boolean isEventBusEnabled() { return
getBoolean(ConfVars.ZEPPELIN_EVENTBUS_ENABLED); }
+
public DEFAULT_UI getDefaultUi() {
return
DEFAULT_UI.valueOf(getString(ConfVars.ZEPPELIN_DEFAULT_UI).toUpperCase());
}
@@ -1222,7 +1224,8 @@ public class ZeppelinConfiguration {
),
ZEPPELIN_ASSISTANT_OPENAI_MODEL(
"zeppelin.assistant.openai.model", "gpt-6-luna"
- );
+ ),
+ ZEPPELIN_EVENTBUS_ENABLED("zeppelin.eventbus.enabled", false);
private String varName;
private Class<?> varClass;
diff --git
a/zeppelin-server/src/main/java/org/apache/zeppelin/eventbus/EventBus.java
b/zeppelin-server/src/main/java/org/apache/zeppelin/eventbus/EventBus.java
new file mode 100644
index 0000000000..3be7559776
--- /dev/null
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/eventbus/EventBus.java
@@ -0,0 +1,34 @@
+/*
+ * 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.eventbus;
+
+import io.reactivex.rxjava3.disposables.Disposable;
+import io.reactivex.rxjava3.functions.Consumer;
+
+public interface EventBus {
+
+ void post(ZeppelinEvent event);
+
+ /**
+ * Handlers may not run on the same thread that calls post().
+ */
+ <T extends ZeppelinEvent> Disposable subscribe(
+ Class<T> eventType,
+ Consumer<? super T> handler
+ );
+}
diff --git
a/zeppelin-server/src/main/java/org/apache/zeppelin/eventbus/NoOpEventBus.java
b/zeppelin-server/src/main/java/org/apache/zeppelin/eventbus/NoOpEventBus.java
new file mode 100644
index 0000000000..fe6198790b
--- /dev/null
+++
b/zeppelin-server/src/main/java/org/apache/zeppelin/eventbus/NoOpEventBus.java
@@ -0,0 +1,47 @@
+/*
+ * 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.eventbus;
+
+import io.reactivex.rxjava3.disposables.Disposable;
+import io.reactivex.rxjava3.functions.Consumer;
+import jakarta.inject.Inject;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class NoOpEventBus implements EventBus {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(NoOpEventBus.class);
+
+ @Inject
+ public NoOpEventBus() {
+ LOGGER.info("Starting NoOpEventBus");
+ }
+
+ @Override
+ public void post(ZeppelinEvent event) {
+ LOGGER.debug("Posting event: {}", event.getClass().getName());
+ }
+
+ @Override
+ public <T extends ZeppelinEvent> Disposable subscribe(
+ Class<T> eventType,
+ Consumer<? super T> handler
+ ) {
+ return Disposable.disposed();
+ }
+}
diff --git
a/zeppelin-server/src/main/java/org/apache/zeppelin/eventbus/NoteEvent.java
b/zeppelin-server/src/main/java/org/apache/zeppelin/eventbus/NoteEvent.java
new file mode 100644
index 0000000000..2b6b806a14
--- /dev/null
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/eventbus/NoteEvent.java
@@ -0,0 +1,28 @@
+/*
+ * 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.eventbus;
+
+import org.apache.zeppelin.notebook.Note;
+import org.apache.zeppelin.user.AuthenticationInfo;
+
+public interface NoteEvent extends ZeppelinEvent {
+
+ Note getNote();
+
+ AuthenticationInfo getSubject();
+}
diff --git
a/zeppelin-server/src/main/java/org/apache/zeppelin/eventbus/NoteRemovedEvent.java
b/zeppelin-server/src/main/java/org/apache/zeppelin/eventbus/NoteRemovedEvent.java
new file mode 100644
index 0000000000..de7ca88749
--- /dev/null
+++
b/zeppelin-server/src/main/java/org/apache/zeppelin/eventbus/NoteRemovedEvent.java
@@ -0,0 +1,43 @@
+/*
+ * 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.eventbus;
+
+import org.apache.zeppelin.notebook.Note;
+import org.apache.zeppelin.user.AuthenticationInfo;
+
+public class NoteRemovedEvent implements NoteEvent {
+
+ private final Note note;
+
+ private final AuthenticationInfo subject;
+
+ public NoteRemovedEvent(Note note, AuthenticationInfo subject) {
+ this.note = note;
+ this.subject = subject;
+ }
+
+ @Override
+ public Note getNote() {
+ return this.note;
+ }
+
+ @Override
+ public AuthenticationInfo getSubject() {
+ return subject;
+ }
+}
diff --git
a/zeppelin-server/src/main/java/org/apache/zeppelin/eventbus/ZeppelinEvent.java
b/zeppelin-server/src/main/java/org/apache/zeppelin/eventbus/ZeppelinEvent.java
new file mode 100644
index 0000000000..2f181ef7e8
--- /dev/null
+++
b/zeppelin-server/src/main/java/org/apache/zeppelin/eventbus/ZeppelinEvent.java
@@ -0,0 +1,21 @@
+/*
+ * 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.eventbus;
+
+public interface ZeppelinEvent {
+}
diff --git
a/zeppelin-server/src/main/java/org/apache/zeppelin/eventbus/ZeppelinEventBus.java
b/zeppelin-server/src/main/java/org/apache/zeppelin/eventbus/ZeppelinEventBus.java
new file mode 100644
index 0000000000..1ae66db7a0
--- /dev/null
+++
b/zeppelin-server/src/main/java/org/apache/zeppelin/eventbus/ZeppelinEventBus.java
@@ -0,0 +1,63 @@
+/*
+ * 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.eventbus;
+
+import io.reactivex.rxjava3.disposables.Disposable;
+import io.reactivex.rxjava3.functions.Consumer;
+import io.reactivex.rxjava3.subjects.PublishSubject;
+import io.reactivex.rxjava3.subjects.Subject;
+import jakarta.inject.Inject;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ZeppelinEventBus implements EventBus {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(ZeppelinEventBus.class);
+
+ private final Subject<ZeppelinEvent> eventBus;
+
+ @Inject
+ public ZeppelinEventBus() {
+ LOGGER.info("Starting ZeppelinEventBus");
+
+ eventBus = PublishSubject.<ZeppelinEvent>create().toSerialized();
+ }
+
+ @Override
+ public void post(ZeppelinEvent event) {
+ LOGGER.debug("Posting event: {}", event.getClass().getName());
+
+ eventBus.onNext(event);
+ }
+
+ @Override
+ public <T extends ZeppelinEvent> Disposable subscribe(
+ Class<T> eventType,
+ Consumer<? super T> handler
+ ) {
+ return eventBus.ofType(eventType).subscribe(event -> {
+ try {
+ handler.accept(event);
+ } catch (Exception e) {
+ LOGGER.error("Failed to handle event: {}", eventType.getName(), e);
+ }
+ }, error -> LOGGER.error(
+ "Event subscription terminated: {}", eventType.getName(), error)
+ );
+ }
+}
diff --git
a/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/Notebook.java
b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/Notebook.java
index 10c2abca77..9805300ab2 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/Notebook.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/Notebook.java
@@ -44,6 +44,8 @@ import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.conf.ZeppelinConfiguration.ConfVars;
import org.apache.zeppelin.display.AngularObject;
import org.apache.zeppelin.display.AngularObjectRegistry;
+import org.apache.zeppelin.eventbus.EventBus;
+import org.apache.zeppelin.eventbus.NoteRemovedEvent;
import org.apache.zeppelin.interpreter.Interpreter;
import org.apache.zeppelin.interpreter.InterpreterFactory;
import org.apache.zeppelin.interpreter.InterpreterGroup;
@@ -86,11 +88,13 @@ public class Notebook {
private Credentials credentials;
private final List<Consumer<String>> initConsumers;
private ExecutorService initExecutor;
+ private EventBus eventBus;
/**
* Main constructor \w manual Dependency Injection
*
* @throws IOException
+ *
* @throws SchedulerException
*/
public Notebook(
@@ -100,8 +104,8 @@ public class Notebook {
NoteManager noteManager,
InterpreterFactory replFactory,
InterpreterSettingManager interpreterSettingManager,
- Credentials credentials)
- {
+ Credentials credentials,
+ EventBus eventBus) {
this.zConf = zConf;
this.authorizationService = authorizationService;
this.noteManager = noteManager;
@@ -111,6 +115,7 @@ public class Notebook {
// TODO(zjffdu) cycle refer, not a good solution
this.interpreterSettingManager.setNotebook(this);
this.credentials = credentials;
+ this.eventBus = eventBus;
addNotebookEventListener(this.interpreterSettingManager);
initConsumers = new LinkedList<>();
}
@@ -221,7 +226,8 @@ public class Notebook {
InterpreterFactory replFactory,
InterpreterSettingManager interpreterSettingManager,
Credentials credentials,
- NoteEventListener noteEventListener)
+ NoteEventListener noteEventListener,
+ EventBus eventBus)
throws IOException {
this(
zConf,
@@ -230,7 +236,8 @@ public class Notebook {
noteManager,
replFactory,
interpreterSettingManager,
- credentials);
+ credentials,
+ eventBus);
if (null != noteEventListener) {
addNotebookEventListener(noteEventListener);
}
@@ -432,7 +439,11 @@ public class Notebook {
note.setRemoved(true);
noteManager.removeNote(note.getId(), subject);
authorizationService.removeNoteAuth(note.getId());
+
fireNoteRemoveEvent(note, subject);
+ if (zConf.isEventBusEnabled()) {
+ eventBus.post(new NoteRemovedEvent(note, subject));
+ }
}
public void removeCorruptedNote(String noteId, AuthenticationInfo subject)
throws IOException {
@@ -564,6 +575,9 @@ public class Notebook {
note.setRemoved(true);
authorizationService.removeNoteAuth(note.getId());
fireNoteRemoveEvent(note, subject);
+ if (zConf.isEventBusEnabled()) {
+ eventBus.post(new NoteRemovedEvent(note, subject));
+ }
}
return null;
});
@@ -843,6 +857,7 @@ public class Notebook {
}
}
+
private void fireNoteRemoveEvent(Note note, AuthenticationInfo subject) {
for (NoteEventListener listener : noteEventListeners) {
listener.onNoteRemove(note, subject);
diff --git
a/zeppelin-server/src/main/java/org/apache/zeppelin/server/ZeppelinServer.java
b/zeppelin-server/src/main/java/org/apache/zeppelin/server/ZeppelinServer.java
index af255aa2c6..06924121b4 100644
---
a/zeppelin-server/src/main/java/org/apache/zeppelin/server/ZeppelinServer.java
+++
b/zeppelin-server/src/main/java/org/apache/zeppelin/server/ZeppelinServer.java
@@ -64,6 +64,9 @@ import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.conf.ZeppelinConfiguration.ConfVars;
import org.apache.zeppelin.conf.ZeppelinConfiguration.DEFAULT_UI;
import org.apache.zeppelin.display.AngularObjectRegistryListener;
+import org.apache.zeppelin.eventbus.EventBus;
+import org.apache.zeppelin.eventbus.NoOpEventBus;
+import org.apache.zeppelin.eventbus.ZeppelinEventBus;
import org.apache.zeppelin.healthcheck.HealthChecks;
import org.apache.zeppelin.helium.ApplicationEventListener;
import org.apache.zeppelin.helium.Helium;
@@ -182,6 +185,11 @@ public class ZeppelinServer implements AutoCloseable {
bind(storage).to(ConfigStorage.class);
bindAsContract(PluginManager.class).in(Singleton.class);
bind(GsonNoteParser.class).to(NoteParser.class).in(Singleton.class);
+ if (zConf.isEventBusEnabled()) {
+
bind(ZeppelinEventBus.class).to(EventBus.class).in(Singleton.class);
+ } else {
+ bind(NoOpEventBus.class).to(EventBus.class).in(Singleton.class);
+ }
bindAsContract(InterpreterFactory.class).in(Singleton.class);
bindAsContract(NotebookRepoSync.class).to(NotebookRepo.class).in(Singleton.class);
bindAsContract(Helium.class).in(Singleton.class);
@@ -388,6 +396,7 @@ public class ZeppelinServer implements AutoCloseable {
if (!zConf.isRecoveryEnabled()) {
sharedServiceLocator.getService(InterpreterSettingManager.class).close();
}
+ sharedServiceLocator.getService(NotebookServer.class).close();
sharedServiceLocator.getService(Notebook.class).close();
}
} catch (Exception e) {
diff --git
a/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java
b/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java
index 63ce877f86..ca1c1756f1 100644
---
a/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java
+++
b/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java
@@ -43,6 +43,8 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
+
+import io.reactivex.rxjava3.disposables.CompositeDisposable;
import jakarta.inject.Inject;
import jakarta.inject.Provider;
import jakarta.websocket.CloseReason;
@@ -66,6 +68,8 @@ import org.apache.zeppelin.display.AngularObjectRegistry;
import org.apache.zeppelin.display.AngularObjectRegistryListener;
import org.apache.zeppelin.display.GUI;
import org.apache.zeppelin.display.Input;
+import org.apache.zeppelin.eventbus.EventBus;
+import org.apache.zeppelin.eventbus.NoteRemovedEvent;
import org.apache.zeppelin.helium.ApplicationEventListener;
import org.apache.zeppelin.helium.HeliumPackage;
import org.apache.zeppelin.interpreter.InterpreterGroup;
@@ -123,7 +127,8 @@ public class NotebookServer implements
AngularObjectRegistryListener,
RemoteInterpreterProcessListener,
ApplicationEventListener,
ParagraphJobListener,
- NoteEventListener {
+ NoteEventListener,
+ AutoCloseable {
/**
* Job manager service type.
@@ -170,6 +175,7 @@ public class NotebookServer implements
AngularObjectRegistryListener,
private AuthorizationService authorizationService;
private Provider<JobManagerService> jobManagerServiceProvider;
private Provider<Assistant> assistantProvider;
+ private CompositeDisposable subscriptions;
public NotebookServer() {
NotebookServer.self.set(this);
@@ -181,6 +187,26 @@ public class NotebookServer implements
AngularObjectRegistryListener,
this.zConf = zConf;
}
+ @Inject
+ public void registerEventBus(EventBus eventBus, ZeppelinConfiguration zConf)
{
+ if (!zConf.isEventBusEnabled()) {
+ LOGGER.debug("ZeppelinEventBus is disabled");
+ return;
+ }
+ subscriptions = new CompositeDisposable();
+
+ subscriptions.add(eventBus.subscribe(
+ NoteRemovedEvent.class, event -> handleNoteRemove(event.getNote()))
+ );
+ }
+
+ @Override
+ public void close() {
+ if (subscriptions != null && !subscriptions.isDisposed()) {
+ subscriptions.dispose();
+ }
+ }
+
@Inject
public void setNoteParser(Provider<NoteParser> noteParser) {
this.noteParser = noteParser;
@@ -2080,19 +2106,12 @@ public class NotebookServer implements
AngularObjectRegistryListener,
@Override
public void onNoteRemove(Note note, AuthenticationInfo subject) {
- try {
- broadcastUpdateNoteJobInfo(note, System.currentTimeMillis() - 5000);
- } catch (IOException e) {
- LOGGER.warn("can not broadcast for job manager: {}", e.getMessage(), e);
- }
-
- try {
- getJobManagerService().removeNoteJobInfo(note.getId(), null,
- new JobManagerServiceCallback());
- } catch (IOException e) {
- LOGGER.warn("can not broadcast for job manager: {}", e.getMessage(), e);
+ if (zConf.isEventBusEnabled()) {
+ LOGGER.debug("ZeppelinEventBus is enabled");
+ return;
}
+ handleNoteRemove(note);
}
@Override
@@ -2549,4 +2568,20 @@ public class NotebookServer implements
AngularObjectRegistryListener,
}
}
}
+
+ private void handleNoteRemove(Note note) {
+ try {
+ broadcastUpdateNoteJobInfo(note, System.currentTimeMillis() - 5000);
+ } catch (IOException e) {
+ LOGGER.warn("can not broadcast for job manager: {}", e.getMessage(), e);
+ }
+
+ try {
+ getJobManagerService().removeNoteJobInfo(
+ note.getId(), null, new JobManagerServiceCallback()
+ );
+ } catch (IOException e) {
+ LOGGER.warn("can not broadcast for job manager: {}", e.getMessage(), e);
+ }
+ }
}
diff --git
a/zeppelin-server/src/test/java/org/apache/zeppelin/eventbus/ZeppelinEventBusTest.java
b/zeppelin-server/src/test/java/org/apache/zeppelin/eventbus/ZeppelinEventBusTest.java
new file mode 100644
index 0000000000..3f9f83aebc
--- /dev/null
+++
b/zeppelin-server/src/test/java/org/apache/zeppelin/eventbus/ZeppelinEventBusTest.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.eventbus;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import io.reactivex.rxjava3.disposables.Disposable;
+
+class ZeppelinEventBusTest {
+
+ @Test
+ void testSubscription() {
+ EventBus bus = new ZeppelinEventBus();
+ List<String> received = new ArrayList<>();
+ Disposable subscription = bus.subscribe(MockEvent.class, event ->
received.add(event.payload));
+
+ bus.post(new ZeppelinEvent() { });
+ bus.post(new MockEvent("received"));
+ assertEquals(List.of("received"), received);
+
+ subscription.dispose();
+ bus.post(new MockEvent("ignored"));
+ assertEquals(List.of("received"), received);
+ }
+
+ @Test
+ void testContinueAfterException() {
+ EventBus bus = new ZeppelinEventBus();
+ List<String> received = new ArrayList<>();
+ Disposable subscription = bus.subscribe(MockEvent.class, event -> {
+ if ("checked".equals(event.payload)) {
+ throw new IOException("Handler failed");
+ }
+ if ("runtime".equals(event.payload)) {
+ throw new IllegalStateException("Handler failed");
+ }
+ received.add(event.payload);
+ });
+
+ try {
+ bus.post(new MockEvent("checked"));
+ bus.post(new MockEvent("after checked"));
+ bus.post(new MockEvent("runtime"));
+ bus.post(new MockEvent("after runtime"));
+
+ assertFalse(subscription.isDisposed());
+ assertEquals(List.of("after checked", "after runtime"), received);
+ } finally {
+ subscription.dispose();
+ }
+ }
+
+ private static class MockEvent implements ZeppelinEvent {
+ private final String payload;
+
+ MockEvent(String payload) {
+ this.payload = payload;
+ }
+ }
+}
diff --git
a/zeppelin-server/src/test/java/org/apache/zeppelin/helium/HeliumApplicationFactoryTest.java
b/zeppelin-server/src/test/java/org/apache/zeppelin/helium/HeliumApplicationFactoryTest.java
index 2f356281ab..d66662ebfa 100644
---
a/zeppelin-server/src/test/java/org/apache/zeppelin/helium/HeliumApplicationFactoryTest.java
+++
b/zeppelin-server/src/test/java/org/apache/zeppelin/helium/HeliumApplicationFactoryTest.java
@@ -70,7 +70,8 @@ class HeliumApplicationFactoryTest extends
AbstractInterpreterTest {
new NoteManager(notebookRepo, ZeppelinConfiguration.load()),
interpreterFactory,
interpreterSettingManager,
- new Credentials());
+ new Credentials(),
+ eventBus);
heliumAppFactory = new HeliumApplicationFactory(notebook, null);
diff --git
a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/AbstractInterpreterTest.java
b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/AbstractInterpreterTest.java
index 0bfd19014b..2f7b6a65cb 100644
---
a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/AbstractInterpreterTest.java
+++
b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/AbstractInterpreterTest.java
@@ -21,6 +21,7 @@ package org.apache.zeppelin.interpreter;
import org.apache.commons.io.FileUtils;
import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.display.AngularObjectRegistryListener;
+import org.apache.zeppelin.eventbus.ZeppelinEventBus;
import org.apache.zeppelin.helium.ApplicationEventListener;
import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcessListener;
import org.apache.zeppelin.notebook.AuthorizationService;
@@ -65,6 +66,7 @@ public abstract class AbstractInterpreterTest {
protected ZeppelinConfiguration zConf;
protected ConfigStorage storage;
protected PluginManager pluginManager;
+ protected ZeppelinEventBus eventBus;
@BeforeEach
public void setUp() throws Exception {
@@ -100,12 +102,12 @@ public abstract class AbstractInterpreterTest {
zConf.setProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_GROUP_DEFAULT.getVarName(),
"test");
-
NotebookRepo notebookRepo = new InMemoryNotebookRepo();
NoteManager noteManager = new NoteManager(notebookRepo, zConf);
noteParser = new GsonNoteParser(zConf);
storage = ConfigStorage.createConfigStorage(zConf);
pluginManager = new PluginManager(zConf);
+ eventBus = new ZeppelinEventBus();
AuthorizationService authorizationService =
new AuthorizationService(noteManager, zConf, storage);
@@ -114,7 +116,7 @@ public abstract class AbstractInterpreterTest {
mock(ApplicationEventListener.class), storage, pluginManager);
interpreterFactory = new InterpreterFactory(interpreterSettingManager);
Credentials credentials = new Credentials(zConf, storage);
- notebook = new Notebook(zConf, authorizationService, notebookRepo,
noteManager, interpreterFactory, interpreterSettingManager, credentials);
+ notebook = new Notebook(zConf, authorizationService, notebookRepo,
noteManager, interpreterFactory, interpreterSettingManager, credentials,
eventBus);
interpreterSettingManager.setNotebook(notebook);
}
diff --git
a/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/NotebookTest.java
b/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/NotebookTest.java
index f7113944f8..dece3335a4 100644
---
a/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/NotebookTest.java
+++
b/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/NotebookTest.java
@@ -21,6 +21,7 @@ import org.apache.commons.io.IOUtils;
import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.conf.ZeppelinConfiguration.ConfVars;
import org.apache.zeppelin.display.AngularObjectRegistry;
+import org.apache.zeppelin.eventbus.ZeppelinEventBus;
import org.apache.zeppelin.interpreter.AbstractInterpreterTest;
import org.apache.zeppelin.interpreter.ExecutionContext;
import org.apache.zeppelin.interpreter.InterpreterException;
@@ -107,7 +108,8 @@ class NotebookTest extends AbstractInterpreterTest
implements ParagraphJobListen
authorizationService = new AuthorizationService(noteManager, zConf,
storage);
credentials = new Credentials(zConf, storage);
- notebook = new Notebook(zConf, authorizationService, notebookRepo,
noteManager, interpreterFactory, interpreterSettingManager, credentials, null);
+ ZeppelinEventBus eventBus = new ZeppelinEventBus();
+ notebook = new Notebook(zConf, authorizationService, notebookRepo,
noteManager, interpreterFactory, interpreterSettingManager, credentials,
eventBus);
notebook.setParagraphJobListener(this);
schedulerService = new QuartzSchedulerService(zConf, notebook);
notebook.initNotebook();
diff --git
a/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/repo/NotebookRepoSyncTest.java
b/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/repo/NotebookRepoSyncTest.java
index 3a90e4f603..b7cb75729e 100644
---
a/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/repo/NotebookRepoSyncTest.java
+++
b/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/repo/NotebookRepoSyncTest.java
@@ -39,6 +39,8 @@ import org.apache.commons.io.FileUtils;
import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.conf.ZeppelinConfiguration.ConfVars;
import org.apache.zeppelin.display.AngularObjectRegistryListener;
+import org.apache.zeppelin.eventbus.EventBus;
+import org.apache.zeppelin.eventbus.ZeppelinEventBus;
import org.apache.zeppelin.helium.ApplicationEventListener;
import org.apache.zeppelin.interpreter.InterpreterFactory;
import org.apache.zeppelin.interpreter.InterpreterSettingManager;
@@ -76,6 +78,7 @@ class NotebookRepoSyncTest {
private InterpreterFactory factory;
private InterpreterSettingManager interpreterSettingManager;
private Credentials credentials;
+ private EventBus eventBus;
private AuthenticationInfo anonymous;
private NoteManager noteManager;
private AuthorizationService authorizationService;
@@ -116,7 +119,8 @@ class NotebookRepoSyncTest {
noteManager = new NoteManager(notebookRepoSync, zConf);
authorizationService = new AuthorizationService(noteManager, zConf,
storage);
credentials = new Credentials(zConf, storage);
- notebook = new Notebook(zConf, authorizationService, notebookRepoSync,
noteManager, factory, interpreterSettingManager, credentials, null);
+ eventBus = new ZeppelinEventBus();
+ notebook = new Notebook(zConf, authorizationService, notebookRepoSync,
noteManager, factory, interpreterSettingManager, credentials, eventBus);
anonymous = new AuthenticationInfo("anonymous");
}
diff --git
a/zeppelin-server/src/test/java/org/apache/zeppelin/service/NotebookServiceTest.java
b/zeppelin-server/src/test/java/org/apache/zeppelin/service/NotebookServiceTest.java
index 1fb67977bc..d9c23718e9 100644
---
a/zeppelin-server/src/test/java/org/apache/zeppelin/service/NotebookServiceTest.java
+++
b/zeppelin-server/src/test/java/org/apache/zeppelin/service/NotebookServiceTest.java
@@ -49,6 +49,8 @@ import java.util.stream.IntStream;
import org.apache.commons.lang3.StringUtils;
import org.apache.zeppelin.conf.ZeppelinConfiguration;
+import org.apache.zeppelin.eventbus.EventBus;
+import org.apache.zeppelin.eventbus.ZeppelinEventBus;
import org.apache.zeppelin.interpreter.Interpreter;
import org.apache.zeppelin.interpreter.Interpreter.FormType;
import org.apache.zeppelin.interpreter.InterpreterFactory;
@@ -84,9 +86,6 @@ import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.junit.jupiter.api.TestInfo;
-
-import com.google.gson.Gson;
-
class NotebookServiceTest {
private static NotebookService notebookService;
@@ -149,6 +148,7 @@ class NotebookServiceTest {
Credentials credentials = new Credentials();
NoteManager noteManager = new NoteManager(notebookRepo, zConf);
authorizationService = new AuthorizationService(noteManager, zConf,
storage);
+ EventBus eventBus = new ZeppelinEventBus();
notebook =
new Notebook(
zConf,
@@ -158,7 +158,7 @@ class NotebookServiceTest {
mockInterpreterFactory,
mockInterpreterSettingManager,
credentials,
- null);
+ eventBus);
searchService = new LuceneSearch(zConf, notebook);
QuartzSchedulerService schedulerService = new
QuartzSchedulerService(zConf, notebook);
notebook.initNotebook();
diff --git
a/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerTest.java
b/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerTest.java
index 1708fa2044..6aae0af52a 100644
---
a/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerTest.java
+++
b/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerTest.java
@@ -56,6 +56,9 @@ import org.apache.thrift.TException;
import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.display.AngularObject;
import org.apache.zeppelin.display.AngularObjectBuilder;
+import org.apache.zeppelin.eventbus.EventBus;
+import org.apache.zeppelin.eventbus.NoteRemovedEvent;
+import org.apache.zeppelin.eventbus.ZeppelinEventBus;
import org.apache.zeppelin.interpreter.InterpreterGroup;
import org.apache.zeppelin.interpreter.InterpreterSetting;
import org.apache.zeppelin.interpreter.remote.RemoteAngularObjectRegistry;
@@ -74,6 +77,7 @@ import org.apache.zeppelin.common.Message.OP;
import org.apache.zeppelin.rest.AbstractTestRestApi;
import org.apache.zeppelin.scheduler.Job;
import org.apache.zeppelin.scheduler.Job.Status;
+import org.apache.zeppelin.service.JobManagerService;
import org.apache.zeppelin.service.NotebookService;
import org.apache.zeppelin.service.ServiceContext;
import org.apache.zeppelin.user.AuthenticationInfo;
@@ -197,6 +201,32 @@ class NotebookServerTest extends AbstractTestRestApi {
}
}
+ @Test
+ void testEventBusHandlesNoteRemovedEvent() throws IOException {
+ ZeppelinConfiguration conf = ZeppelinConfiguration.load();
+
conf.setProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_EVENTBUS_ENABLED.getVarName(),
"true");
+
+ EventBus eventBus = new ZeppelinEventBus();
+ NotebookServer server = new NotebookServer();
+ AuthorizationService mockAuthorizationService =
mock(AuthorizationService.class);
+ JobManagerService mockJobManagerService = mock(JobManagerService.class);
+ Note note = new Note();
+
+ when(mockAuthorizationService.getOwners(note.getId())).thenReturn(new
HashSet<>());
+
+ server.setZeppelinConfiguration(conf);
+ server.setAuthorizationService(mockAuthorizationService);
+ server.setJobManagerService(() -> mockJobManagerService);
+ server.registerEventBus(eventBus, conf);
+
+ eventBus.post(new NoteRemovedEvent(note, AuthenticationInfo.ANONYMOUS));
+
+ verify(mockJobManagerService).getNoteJobInfoByUnixTime(Mockito.anyLong(),
Mockito.any(), Mockito.any());
+ verify(mockJobManagerService).removeNoteJobInfo(eq(note.getId()),
Mockito.isNull(), Mockito.any());
+
+ server.close();
+ }
+
@Test
void testOnParagraphCreate_whenJobManagerDisabled() {
boolean originalFlag = disableJobManagerAndBackupFlag();
