From b412a2644543e8473953abd415a4d8389c3c3fc3 Mon Sep 17 00:00:00 2001
From: Peter Chacko <peterchacko@google.com>
Date: Fri, 7 Jul 2023 23:29:26 +0000
Subject: [PATCH] Event Subscriber Infrastructure

---
 cdap-app-fabric/pom.xml                       |  7 +-
 .../guice/AppFabricServiceRuntimeModule.java  |  7 +-
 .../events/DefaultEventReaderContext.java     | 47 ++++++++++
 .../internal/events/EventReaderProvider.java  | 33 +++++++
 .../cdap/internal/events/EventSubscriber.java | 33 +++++++
 .../events/EventSubscriberManager.java        | 72 +++++++++++++++
 .../events/dummy/DummyEventReader.java        | 87 +++++++++++++++++++
 .../DummyEventReaderExtensionProvider.java    | 45 ++++++++++
 .../io/cdap/cdap/common/conf/Constants.java   |  2 +
 .../src/main/resources/cdap-default.xml       | 10 +++
 .../java/io/cdap/cdap/features/Feature.java   |  5 +-
 .../java/io/cdap/cdap/StandaloneMain.java     |  5 ++
 12 files changed, 350 insertions(+), 3 deletions(-)
 create mode 100644 cdap-app-fabric/src/main/java/io/cdap/cdap/internal/events/DefaultEventReaderContext.java
 create mode 100644 cdap-app-fabric/src/main/java/io/cdap/cdap/internal/events/EventReaderProvider.java
 create mode 100644 cdap-app-fabric/src/main/java/io/cdap/cdap/internal/events/EventSubscriber.java
 create mode 100644 cdap-app-fabric/src/main/java/io/cdap/cdap/internal/events/EventSubscriberManager.java
 create mode 100644 cdap-app-fabric/src/main/java/io/cdap/cdap/internal/events/dummy/DummyEventReader.java
 create mode 100644 cdap-app-fabric/src/main/java/io/cdap/cdap/internal/events/dummy/DummyEventReaderExtensionProvider.java

diff --git a/cdap-app-fabric/pom.xml b/cdap-app-fabric/pom.xml
index e9793a1f6c7f..b43fee7793c3 100644
--- a/cdap-app-fabric/pom.xml
+++ b/cdap-app-fabric/pom.xml
@@ -1,6 +1,6 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <!--
-  Copyright © 2014-2021 Cask Data, Inc.
+  Copyright © 2014-2023 Cask Data, Inc.
 
   Licensed 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
@@ -269,6 +269,11 @@
       <artifactId>cdap-source-control</artifactId>
       <version>${project.version}</version>
     </dependency>
+      <dependency>
+          <groupId>io.cdap.cdap</groupId>
+          <artifactId>cdap-event-reader-spi</artifactId>
+          <version>${project.version}</version>
+      </dependency>
   </dependencies>
 
   <build>
diff --git a/cdap-app-fabric/src/main/java/io/cdap/cdap/app/guice/AppFabricServiceRuntimeModule.java b/cdap-app-fabric/src/main/java/io/cdap/cdap/app/guice/AppFabricServiceRuntimeModule.java
index 6442af24e92e..70e6e3d5d24b 100644
--- a/cdap-app-fabric/src/main/java/io/cdap/cdap/app/guice/AppFabricServiceRuntimeModule.java
+++ b/cdap-app-fabric/src/main/java/io/cdap/cdap/app/guice/AppFabricServiceRuntimeModule.java
@@ -120,6 +120,8 @@
 import io.cdap.cdap.internal.credential.handler.CredentialProviderHttpHandlerInternal;
 import io.cdap.cdap.internal.events.EventPublishManager;
 import io.cdap.cdap.internal.events.EventPublisher;
+import io.cdap.cdap.internal.events.EventSubscriber;
+import io.cdap.cdap.internal.events.EventSubscriberManager;
 import io.cdap.cdap.internal.events.EventWriterExtensionProvider;
 import io.cdap.cdap.internal.events.EventWriterProvider;
 import io.cdap.cdap.internal.events.MetricsProvider;
@@ -398,6 +400,9 @@ protected void configure() {
           Multibinder.newSetBinder(binder(), EventPublisher.class);
       eventPublishersBinder.addBinding().to(ProgramStatusEventPublisher.class);
       bind(EventPublishManager.class).in(Scopes.SINGLETON);
+      Multibinder<EventSubscriber> eventSubscribersBinder =
+              Multibinder.newSetBinder(binder(), EventSubscriber.class);
+      bind(EventSubscriberManager.class).in(Scopes.SINGLETON);
       bind(EventWriterProvider.class).to(EventWriterExtensionProvider.class);
       bind(MetricsProvider.class).to(SparkProgramStatusMetricsProvider.class);
 
@@ -488,7 +493,7 @@ public synchronized org.quartz.Scheduler get() {
      * @return an instance of {@link org.quartz.Scheduler}
      */
     private org.quartz.Scheduler getScheduler(JobStore store,
-        CConfiguration cConf) throws SchedulerException {
+                                              CConfiguration cConf) throws SchedulerException {
 
       int threadPoolSize = cConf.getInt(Constants.Scheduler.CFG_SCHEDULER_MAX_THREAD_POOL_SIZE);
       ExecutorThreadPool threadPool = new ExecutorThreadPool(threadPoolSize);
diff --git a/cdap-app-fabric/src/main/java/io/cdap/cdap/internal/events/DefaultEventReaderContext.java b/cdap-app-fabric/src/main/java/io/cdap/cdap/internal/events/DefaultEventReaderContext.java
new file mode 100644
index 000000000000..0a5c21d19170
--- /dev/null
+++ b/cdap-app-fabric/src/main/java/io/cdap/cdap/internal/events/DefaultEventReaderContext.java
@@ -0,0 +1,47 @@
+/*
+ * Copyright © 2023 Cask Data, Inc.
+ * Licensed 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 io.cdap.cdap.internal.events;
+
+import io.cdap.cdap.common.conf.CConfiguration;
+import io.cdap.cdap.common.conf.Constants;
+import io.cdap.cdap.spi.events.EventReaderContext;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Provides an initialized default context for EventReader implementing {@link EventReaderContext}.
+ */
+public class DefaultEventReaderContext implements EventReaderContext {
+
+  private final Map<String, String> properties;
+  private final static int DEFAULT_ACK_DEADLINE = 600;
+
+  /**
+   * Construct the default Event reader context.
+   *
+   * @param prefix prefix for specific event reader
+   * @param cConf        An instance of an injected ${@link CConfiguration}.
+   */
+  DefaultEventReaderContext(String prefix, CConfiguration cConf) {
+    Map<String, String> mutableProperties = new HashMap<>(cConf.getPropsWithPrefix(prefix));
+    mutableProperties.put("ackDeadline", cConf.get(Constants.Event.EVENT_READER_ACK_DEADLINE,
+            String.valueOf(DEFAULT_ACK_DEADLINE)));
+    this.properties = Collections.unmodifiableMap(mutableProperties);
+  }
+
+  @Override
+  public Map<String, String> getProperties() {
+    return properties;
+  }
+}
diff --git a/cdap-app-fabric/src/main/java/io/cdap/cdap/internal/events/EventReaderProvider.java b/cdap-app-fabric/src/main/java/io/cdap/cdap/internal/events/EventReaderProvider.java
new file mode 100644
index 000000000000..38fa8369796b
--- /dev/null
+++ b/cdap-app-fabric/src/main/java/io/cdap/cdap/internal/events/EventReaderProvider.java
@@ -0,0 +1,33 @@
+/*
+ * Copyright © 2023 Cask Data, Inc.
+ *
+ * Licensed 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 io.cdap.cdap.internal.events;
+
+import io.cdap.cdap.spi.events.Event;
+import io.cdap.cdap.spi.events.EventReader;
+import java.util.Map;
+
+/**
+ * This interface provides Event Readers ({@link EventReader}) in order to load them as an
+ * extension.
+ */
+public interface EventReaderProvider<T extends Event> {
+
+  /**
+   * Method which retrieve a {@link Map} of {@link EventReader} and the Extension Type.
+   */
+  Map<String, EventReader<T>> loadEventReaders();
+}
diff --git a/cdap-app-fabric/src/main/java/io/cdap/cdap/internal/events/EventSubscriber.java b/cdap-app-fabric/src/main/java/io/cdap/cdap/internal/events/EventSubscriber.java
new file mode 100644
index 000000000000..9f54c3accb87
--- /dev/null
+++ b/cdap-app-fabric/src/main/java/io/cdap/cdap/internal/events/EventSubscriber.java
@@ -0,0 +1,33 @@
+/*
+ * Copyright © 2023 Cask Data, Inc.
+ *
+ * Licensed 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 io.cdap.cdap.internal.events;
+
+import com.google.common.util.concurrent.AbstractScheduledService;
+
+/**
+ * Abstract class for an event subscriber.
+ * Each EventSubscriber will receive and process incoming events.
+ */
+public abstract class EventSubscriber extends AbstractScheduledService {
+
+  /**
+   * Initialize this handler.
+   *
+   * @return true if successful
+   */
+  public abstract boolean initialize();
+}
diff --git a/cdap-app-fabric/src/main/java/io/cdap/cdap/internal/events/EventSubscriberManager.java b/cdap-app-fabric/src/main/java/io/cdap/cdap/internal/events/EventSubscriberManager.java
new file mode 100644
index 000000000000..9d0228d67e25
--- /dev/null
+++ b/cdap-app-fabric/src/main/java/io/cdap/cdap/internal/events/EventSubscriberManager.java
@@ -0,0 +1,72 @@
+/*
+ * Copyright © 2023 Cask Data, Inc.
+ *
+ * Licensed 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 io.cdap.cdap.internal.events;
+
+import com.google.common.util.concurrent.AbstractIdleService;
+import com.google.inject.Inject;
+import io.cdap.cdap.common.conf.CConfiguration;
+import io.cdap.cdap.common.feature.DefaultFeatureFlagsProvider;
+import io.cdap.cdap.features.Feature;
+import java.util.Set;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * EventSubscriberManager is responsible for starting all the event subscriber threads.
+ */
+public class EventSubscriberManager extends AbstractIdleService {
+
+  private static final Logger LOG = LoggerFactory.getLogger(EventSubscriberManager.class);
+
+  private final boolean enabled;
+  private final Set<EventSubscriber> eventSubscribers;
+
+  @Inject
+  EventSubscriberManager(CConfiguration cConf, Set<EventSubscriber> eventSubscribers) {
+    this.enabled = Feature.EVENT_READER.isEnabled(new DefaultFeatureFlagsProvider(cConf));
+    this.eventSubscribers = eventSubscribers;
+  }
+
+  @Override
+  protected void startUp() throws Exception {
+    if (!enabled) {
+      return; // If not enabled, don't start
+    }
+    eventSubscribers.forEach(eventSubscriber -> {
+      // Loading the event readers from provider
+      // Initialize the event subscribers with all the event readers provided by provider
+      if (eventSubscriber.initialize()) {
+        eventSubscriber.startAndWait();
+        LOG.info("Successfully initialized eventSubscriber: {}",
+                eventSubscriber.getClass().getSimpleName());
+      } else {
+        LOG.error("Failed to initialize eventSubscriber: {}",
+                eventSubscriber.getClass().getSimpleName());
+      }
+    });
+  }
+
+  @Override
+  protected void shutDown() throws Exception {
+    if (!enabled) {
+      return; // If not enabled, don't shut down
+    }
+    eventSubscribers.forEach(eventSubscriber -> {
+      eventSubscriber.stopAndWait();
+    });
+  }
+}
diff --git a/cdap-app-fabric/src/main/java/io/cdap/cdap/internal/events/dummy/DummyEventReader.java b/cdap-app-fabric/src/main/java/io/cdap/cdap/internal/events/dummy/DummyEventReader.java
new file mode 100644
index 000000000000..14d1e30f88e4
--- /dev/null
+++ b/cdap-app-fabric/src/main/java/io/cdap/cdap/internal/events/dummy/DummyEventReader.java
@@ -0,0 +1,87 @@
+/*
+ * Copyright © 2023 Cask Data, Inc.
+ * Licensed 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 io.cdap.cdap.internal.events.dummy;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.inject.Inject;
+import io.cdap.cdap.spi.events.Event;
+import io.cdap.cdap.spi.events.EventReader;
+import io.cdap.cdap.spi.events.EventReaderContext;
+import io.cdap.cdap.spi.events.EventResult;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.function.Consumer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Dummy implementation of {@link EventReader} mainly for test proposals.
+ */
+public class DummyEventReader<T extends Event> implements EventReader<T> {
+
+  private static final Logger logger = LoggerFactory.getLogger(DummyEventReader.class);
+
+  @Inject
+  public DummyEventReader() {
+  }
+
+  @Override
+  public void initialize(EventReaderContext eventReaderContext) {
+    logger.info("Initializing DummyEventReader...");
+  }
+
+  @Override
+  public EventResult<T> pull(int maxMessages) {
+    ArrayList<T> sentMessages = new ArrayList<>();
+    Iterator<T> it = getMessages().iterator();
+    int i = 0;
+    while (i < maxMessages && it.hasNext()) {
+      sentMessages.add(it.next());
+      i++;
+    }
+    return new DummyEventResult(sentMessages);
+  }
+
+  @VisibleForTesting
+  public Collection<T> getMessages() {
+   return new ArrayList<>();
+  }
+
+  @Override
+  public void close() throws Exception {
+    logger.info("Closing dummy reader");
+  }
+
+  class DummyEventResult implements EventResult<T> {
+    final Collection<T> events;
+
+    DummyEventResult(Collection<T> events) {
+      this.events = events;
+    }
+
+    @Override
+    public void consumeEvents(Consumer<T> consumer) {
+      for (T evt : events) {
+        try {
+          consumer.accept(evt);
+        } catch (Exception e) {
+          logger.error("Error on consuming event {}", evt.getVersion(), e);
+        }
+      }
+    }
+
+    @Override
+    public void close() throws Exception {}
+  }
+}
diff --git a/cdap-app-fabric/src/main/java/io/cdap/cdap/internal/events/dummy/DummyEventReaderExtensionProvider.java b/cdap-app-fabric/src/main/java/io/cdap/cdap/internal/events/dummy/DummyEventReaderExtensionProvider.java
new file mode 100644
index 000000000000..3a58b3ba35f3
--- /dev/null
+++ b/cdap-app-fabric/src/main/java/io/cdap/cdap/internal/events/dummy/DummyEventReaderExtensionProvider.java
@@ -0,0 +1,45 @@
+/*
+ * Copyright © 2023 Cask Data, Inc.
+ *
+ * Licensed 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 io.cdap.cdap.internal.events.dummy;
+
+import com.google.inject.Inject;
+import io.cdap.cdap.internal.events.EventReaderProvider;
+import io.cdap.cdap.spi.events.Event;
+import io.cdap.cdap.spi.events.EventReader;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Dummy implementation for {@link EventReaderProvider} for test proposals.
+ */
+public class DummyEventReaderExtensionProvider<T extends Event> implements EventReaderProvider {
+
+  private final DummyEventReader eventReader;
+
+  @Inject
+  public DummyEventReaderExtensionProvider(DummyEventReader eventReader) {
+    this.eventReader = eventReader;
+  }
+
+  @Override
+  public Map<String, EventReader> loadEventReaders() {
+    Map<String, EventReader> map = new HashMap<>();
+    map.put(this.eventReader.getClass().getName(), this.eventReader);
+    return Collections.unmodifiableMap(map);
+  }
+}
diff --git a/cdap-common/src/main/java/io/cdap/cdap/common/conf/Constants.java b/cdap-common/src/main/java/io/cdap/cdap/common/conf/Constants.java
index 222f4b64fffc..6ef49c24f914 100644
--- a/cdap-common/src/main/java/io/cdap/cdap/common/conf/Constants.java
+++ b/cdap-common/src/main/java/io/cdap/cdap/common/conf/Constants.java
@@ -2219,6 +2219,8 @@ public static final class Event {
     public static final String EVENTS_WRITER_EXTENSIONS_DIR = "events.writer.extensions.dir";
 
     public static final String EVENTS_WRITER_EXTENSIONS_ENABLED_LIST = "events.writer.extensions.enabled.list";
+    public static final String EVENT_READER_ACK_DEADLINE = "events.reader.ackdeadline";
+
   }
 
   /**
diff --git a/cdap-common/src/main/resources/cdap-default.xml b/cdap-common/src/main/resources/cdap-default.xml
index ad52ead56ba3..fcee0ba7f80a 100644
--- a/cdap-common/src/main/resources/cdap-default.xml
+++ b/cdap-common/src/main/resources/cdap-default.xml
@@ -5051,6 +5051,16 @@
     </description>
   </property>
 
+  <!-- event subscriber -->
+  <property>
+    <name>feature.event.reader.enabled</name>
+    <value>false</value>
+    <final>false</final>
+    <description>
+      Enable event subscribing in CDAP
+    </description>
+  </property>
+
   <!-- Spark on k8s  -->
   <property>
     <name>artifact.fetcher.bind.port</name>
diff --git a/cdap-features/src/main/java/io/cdap/cdap/features/Feature.java b/cdap-features/src/main/java/io/cdap/cdap/features/Feature.java
index e604c573db11..ba0e38d98387 100644
--- a/cdap-features/src/main/java/io/cdap/cdap/features/Feature.java
+++ b/cdap-features/src/main/java/io/cdap/cdap/features/Feature.java
@@ -28,6 +28,7 @@
 public enum Feature {
   REPLICATION_TRANSFORMATIONS("6.6.0"),
   EVENT_PUBLISH("6.7.0", false),
+  EVENT_READER("6.10.0", false),
   PIPELINE_COMPOSITE_TRIGGERS("6.8.0"),
   PUSHDOWN_TRANSFORMATION_GROUPBY("6.7.0"),
   PUSHDOWN_TRANSFORMATION_DEDUPLICATE("6.7.0"),
@@ -71,7 +72,9 @@ public boolean isEnabled(FeatureFlagsProvider featureFlagsProvider) {
   }
 
   /**
-   * @return string that identifies the feature flag.
+   * Retrieve the string that identifies the feature flag.
+   *
+   * @return feature flag string
    */
   public String getFeatureFlagString() {
     return featureFlagString;
diff --git a/cdap-standalone/src/main/java/io/cdap/cdap/StandaloneMain.java b/cdap-standalone/src/main/java/io/cdap/cdap/StandaloneMain.java
index ad1f8c8651e6..ee618ea4e978 100644
--- a/cdap-standalone/src/main/java/io/cdap/cdap/StandaloneMain.java
+++ b/cdap-standalone/src/main/java/io/cdap/cdap/StandaloneMain.java
@@ -70,6 +70,7 @@
 import io.cdap.cdap.internal.app.services.AppFabricServer;
 import io.cdap.cdap.internal.app.worker.sidecar.ArtifactLocalizerService;
 import io.cdap.cdap.internal.events.EventPublishManager;
+import io.cdap.cdap.internal.events.EventSubscriberManager;
 import io.cdap.cdap.logging.LoggingUtil;
 import io.cdap.cdap.logging.appender.LogAppenderInitializer;
 import io.cdap.cdap.logging.framework.LogPipelineLoader;
@@ -155,6 +156,7 @@ public class StandaloneMain {
   private final RuntimeServer runtimeServer;
   private final ArtifactLocalizerService artifactLocalizerService;
   private final EventPublishManager eventPublishManager;
+  private final EventSubscriberManager eventSubscriberManager;
 
   private ExternalAuthenticationServer externalAuthenticationServer;
 
@@ -187,6 +189,7 @@ private StandaloneMain(List<Module> modules, CConfiguration cConf) {
     cConf.setInt(Constants.ArtifactLocalizer.PORT, 0);
     artifactLocalizerService = injector.getInstance(ArtifactLocalizerService.class);
     eventPublishManager = injector.getInstance(EventPublishManager.class);
+    eventSubscriberManager = injector.getInstance(EventSubscriberManager.class);
 
     if (cConf.getBoolean(Constants.Transaction.TX_ENABLED)) {
       txService = injector.getInstance(InMemoryTransactionService.class);
@@ -301,6 +304,7 @@ public void startUp() throws Exception {
     secureStoreService.startAndWait();
     supportBundleInternalService.startAndWait();
     eventPublishManager.startAndWait();
+    eventSubscriberManager.startAndWait();
 
     String protocol = sslEnabled ? "https" : "http";
     int dashboardPort = sslEnabled
@@ -339,6 +343,7 @@ public void shutDown() {
       previewHttpServer.stopAndWait();
       artifactLocalizerService.stopAndWait();
       eventPublishManager.stopAndWait();
+      eventSubscriberManager.stopAndWait();
       // app fabric will also stop all programs
       appFabricServer.stopAndWait();
       runtimeServer.stopAndWait();