[ZEPPELIN-6340] Add ZeppelinEventBus and update NotebookServer to handle NoteRemovedEvent

# Conflicts:
#	zeppelin-server/src/test/java/org/apache/zeppelin/service/NotebookServiceTest.java

Author: seung-00

conf/zeppelin-site.xml.template

@@ -840,4 +840,10 @@
   <description>fields to be excluded from being saved in note files, with Paragraph prefix mean the fields in Paragraph, e.g. Paragraph.results</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>

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>

zeppelin-server/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java

@@ -941,6 +941,8 @@ public boolean isPrometheusMetricEnabled() {
     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());
   }
@@ -1187,7 +1189,8 @@ public enum ConfVars {
     ZEPPELIN_SPARK_ONLY_YARN_CLUSTER("zeppelin.spark.only_yarn_cluster", false),
     ZEPPELIN_SESSION_CHECK_INTERVAL("zeppelin.session.check_interval", 60 * 10 * 1000),
     ZEPPELIN_NOTE_CACHE_THRESHOLD("zeppelin.note.cache.threshold", 50),
-    ZEPPELIN_NOTE_FILE_EXCLUDE_FIELDS("zeppelin.note.file.exclude.fields", "");
+    ZEPPELIN_NOTE_FILE_EXCLUDE_FIELDS("zeppelin.note.file.exclude.fields", ""),
+    ZEPPELIN_EVENTBUS_ENABLED("zeppelin.eventbus.enabled", false);
 
     private String varName;
     private Class<?> varClass;

zeppelin-server/src/main/java/org/apache/zeppelin/eventbus/EventBus.java

@@ -0,0 +1,27 @@
+/*
+ * 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.core.Observable;
+
+public interface EventBus {
+
+  void post(ZeppelinEvent event);
+
+  <T extends ZeppelinEvent> Observable<T> observe(Class<T> eventType);
+}

zeppelin-server/src/main/java/org/apache/zeppelin/eventbus/NoOpEventBus.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 io.reactivex.rxjava3.core.Observable;
+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> Observable<T> observe(Class<T> eventType) {
+    return Observable.empty();
+  }
+}

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();
+}

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;
+  }
+}

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 {
+}

zeppelin-server/src/main/java/org/apache/zeppelin/eventbus/ZeppelinEventBus.java

@@ -0,0 +1,53 @@
+/*
+ * 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.core.Observable;
+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> Observable<T> observe(Class<T> eventType) {
+    LOGGER.debug("Observing event: {}", eventType.getName());
+
+    return eventBus.ofType(eventType);
+  }
+}

zeppelin-server/src/main/java/org/apache/zeppelin/notebook/Notebook.java

@@ -44,6 +44,8 @@
 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 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 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 Notebook(
       InterpreterFactory replFactory,
       InterpreterSettingManager interpreterSettingManager,
       Credentials credentials,
-      NoteEventListener noteEventListener)
+      NoteEventListener noteEventListener,
+      EventBus eventBus)
       throws IOException {
     this(
         zConf,
@@ -230,7 +236,8 @@ public Notebook(
         noteManager,
         replFactory,
         interpreterSettingManager,
-        credentials);
+        credentials,
+        eventBus);
     if (null != noteEventListener) {
       addNotebookEventListener(noteEventListener);
     }
@@ -432,7 +439,11 @@ private void removeNote(Note note, AuthenticationInfo subject) throws IOExceptio
     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 void removeFolder(String folderPath, AuthenticationInfo subject) throws I
             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 @@ private void fireNoteUpdateEvent(Note note, AuthenticationInfo subject) {
     }
   }
 
+
   private void fireNoteRemoveEvent(Note note, AuthenticationInfo subject) {
     for (NoteEventListener listener : noteEventListeners) {
       listener.onNoteRemove(note, subject);