cdapio · vsethi09 · Jan 12, 2025 · Sep 24, 2024
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
@@ -46,6 +46,7 @@
 import io.cdap.cdap.common.conf.CConfiguration;
 import io.cdap.cdap.common.conf.Constants;
 import io.cdap.cdap.common.conf.Constants.AppFabric;
+import io.cdap.cdap.common.conf.Constants.Service;
 import io.cdap.cdap.common.feature.DefaultFeatureFlagsProvider;
 import io.cdap.cdap.common.guice.RemoteAuthenticatorModules;
 import io.cdap.cdap.common.runtime.RuntimeModule;
@@ -220,6 +221,11 @@ protected void configure() {
                     Names.named("appfabric.services.names"));
             servicesNamesBinder.addBinding().toInstance(Constants.Service.APP_FABRIC_HTTP);
 
+            Multibinder<String> processorNamesBinder =
+                Multibinder.newSetBinder(binder(), String.class,
+                    Names.named("appfabric.processor.services.names"));
+            processorNamesBinder.addBinding().toInstance(Service.APP_FABRIC_PROCESSOR);
+
             // TODO: Uncomment after CDAP-7688 is resolved
             // servicesNamesBinder.addBinding().toInstance(Constants.Service.MESSAGING_SERVICE);
 
@@ -230,6 +236,18 @@ protected void configure() {
 
             // TODO: Uncomment after CDAP-7688 is resolved
             // handlerHookNamesBinder.addBinding().toInstance(Constants.Service.MESSAGING_SERVICE);
+
+            // TODO (CDAP-21112): Move HTTP handler from Appfabric processor to server after fixing
+            //  ProgramRuntimeService and RunRecordMonitorService.
+            // Remove additional handlers added to server for in-memory module.
+            Multibinder<HttpHandler> handlerBinder = Multibinder.newSetBinder(
+                binder(), HttpHandler.class, Names.named(Constants.AppFabric.SERVER_HANDLERS_BINDING));
+            handlerBinder.addBinding().to(BootstrapHttpHandler.class);
+            handlerBinder.addBinding().to(AppLifecycleHttpHandler.class);
+            handlerBinder.addBinding().to(AppLifecycleHttpHandlerInternal.class);
+            handlerBinder.addBinding().to(ProgramLifecycleHttpHandler.class);
+            handlerBinder.addBinding().to(ProgramLifecycleHttpHandlerInternal.class);
+            handlerBinder.addBinding().to(WorkflowHttpHandler.class);
           }
         });
   }
@@ -270,6 +288,11 @@ protected void configure() {
             servicesNamesBinder.addBinding().toInstance(Constants.Service.TRANSACTION_HTTP);
             servicesNamesBinder.addBinding().toInstance(Constants.Service.RUNTIME);
 
+            Multibinder<String> processorNamesBinder =
+                Multibinder.newSetBinder(binder(), String.class,
+                    Names.named("appfabric.processor.services.names"));
+            processorNamesBinder.addBinding().toInstance(Service.APP_FABRIC_PROCESSOR);
+
             // TODO: Uncomment after CDAP-7688 is resolved
             // servicesNamesBinder.addBinding().toInstance(Constants.Service.MESSAGING_SERVICE);
 
@@ -330,6 +353,11 @@ protected void configure() {
                     Names.named("appfabric.handler.hooks"));
             handlerHookNamesBinder.addBinding().toInstance(Constants.Service.APP_FABRIC_HTTP);
             servicesNamesBinder.addBinding().toInstance(Constants.Service.SECURE_STORE_SERVICE);
+
+            Multibinder<String> processorNamesBinder =
+                Multibinder.newSetBinder(binder(), String.class,
+                    Names.named("appfabric.processor.services.names"));
+            processorNamesBinder.addBinding().toInstance(Service.APP_FABRIC_PROCESSOR);
           }
         });
   }
@@ -435,7 +463,7 @@ protected void configure() {
       bind(MetricsProvider.class).to(SparkProgramStatusMetricsProvider.class);
 
       Multibinder<HttpHandler> handlerBinder = Multibinder.newSetBinder(
-          binder(), HttpHandler.class, Names.named(Constants.AppFabric.HANDLERS_BINDING));
+          binder(), HttpHandler.class, Names.named(Constants.AppFabric.SERVER_HANDLERS_BINDING));
 
       CommonHandlers.add(handlerBinder);
       handlerBinder.addBinding().to(ConfigHandler.class);
@@ -444,17 +472,12 @@ protected void configure() {
       handlerBinder.addBinding().to(InstanceOperationHttpHandler.class);
       handlerBinder.addBinding().to(NamespaceHttpHandler.class);
       handlerBinder.addBinding().to(SourceControlManagementHttpHandler.class);
-      handlerBinder.addBinding().to(AppLifecycleHttpHandler.class);
-      handlerBinder.addBinding().to(AppLifecycleHttpHandlerInternal.class);
-      handlerBinder.addBinding().to(ProgramLifecycleHttpHandler.class);
-      handlerBinder.addBinding().to(ProgramLifecycleHttpHandlerInternal.class);
       // TODO: [CDAP-13355] Move OperationsDashboardHttpHandler into report generation app
       handlerBinder.addBinding().to(OperationsDashboardHttpHandler.class);
       handlerBinder.addBinding().to(PreferencesHttpHandler.class);
       handlerBinder.addBinding().to(PreferencesHttpHandlerInternal.class);
       handlerBinder.addBinding().to(ConsoleSettingsHttpHandler.class);
       handlerBinder.addBinding().to(TransactionHttpHandler.class);
-      handlerBinder.addBinding().to(WorkflowHttpHandler.class);
       handlerBinder.addBinding().to(ArtifactHttpHandler.class);
       handlerBinder.addBinding().to(ArtifactHttpHandlerInternal.class);
       handlerBinder.addBinding().to(WorkflowStatsSLAHttpHandler.class);
@@ -464,7 +487,6 @@ protected void configure() {
       handlerBinder.addBinding().to(OperationalStatsHttpHandler.class);
       handlerBinder.addBinding().to(ProfileHttpHandler.class);
       handlerBinder.addBinding().to(ProvisionerHttpHandler.class);
-      handlerBinder.addBinding().to(BootstrapHttpHandler.class);
       handlerBinder.addBinding().to(FileFetcherHttpHandlerInternal.class);
       handlerBinder.addBinding().to(TetheringHandler.class);
       handlerBinder.addBinding().to(TetheringServerHandler.class);
@@ -483,6 +505,18 @@ protected void configure() {
       for (Class<? extends HttpHandler> handlerClass : handlerClasses) {
         handlerBinder.addBinding().to(handlerClass);
       }
+
+      Multibinder<HttpHandler> processorHandlerBinder = Multibinder.newSetBinder(
+          binder(), HttpHandler.class, Names.named(AppFabric.PROCESSOR_HANDLERS_BINDING));
+      CommonHandlers.add(processorHandlerBinder);
+      // TODO (CDAP-21112): Move HTTP handler from Appfabric processor to server after fixing
+      //  ProgramRuntimeService and RunRecordMonitorService.
+      processorHandlerBinder.addBinding().to(BootstrapHttpHandler.class);
+      processorHandlerBinder.addBinding().to(AppLifecycleHttpHandler.class);
+      processorHandlerBinder.addBinding().to(AppLifecycleHttpHandlerInternal.class);
+      processorHandlerBinder.addBinding().to(ProgramLifecycleHttpHandler.class);
+      processorHandlerBinder.addBinding().to(ProgramLifecycleHttpHandlerInternal.class);
+      processorHandlerBinder.addBinding().to(WorkflowHttpHandler.class);
     }
 
     @Provides

diff --git a/cdap-app-fabric/src/main/java/io/cdap/cdap/app/guice/MonitorHandlerModule.java b/cdap-app-fabric/src/main/java/io/cdap/cdap/app/guice/MonitorHandlerModule.java
@@ -40,8 +40,10 @@
 import io.cdap.cdap.data2.dataset2.lib.kv.LevelDBKVTableDefinition;
 import io.cdap.cdap.gateway.handlers.DatasetServiceStore;
 import io.cdap.cdap.gateway.handlers.MonitorHandler;
+import io.cdap.cdap.internal.app.runtime.distributed.AppFabricProcessorManager;
 import io.cdap.cdap.internal.app.runtime.distributed.AppFabricServiceManager;
 import io.cdap.cdap.internal.app.runtime.distributed.TransactionServiceManager;
+import io.cdap.cdap.internal.app.runtime.monitor.NonHadoopAppFabricProcessorManager;
 import io.cdap.cdap.internal.app.runtime.monitor.NonHadoopAppFabricServiceManager;
 import io.cdap.cdap.internal.app.runtime.monitor.RuntimeServiceManager;
 import io.cdap.cdap.internal.app.services.AppFabricServer;
@@ -92,7 +94,7 @@ protected void configure() {
     });
 
     Multibinder<HttpHandler> handlerBinder = Multibinder.newSetBinder(
-        binder(), HttpHandler.class, Names.named(Constants.AppFabric.HANDLERS_BINDING));
+        binder(), HttpHandler.class, Names.named(Constants.AppFabric.SERVER_HANDLERS_BINDING));
 
     handlerBinder.addBinding().to(MonitorHandler.class);
   }
@@ -117,6 +119,9 @@ private void addNonHadoopBindings(Binder binder) {
     mapBinder.addBinding(Constants.Service.APP_FABRIC_HTTP)
         .toProvider(
             new NonHadoopMasterServiceManagerProvider(NonHadoopAppFabricServiceManager.class));
+    mapBinder.addBinding(Constants.Service.APP_FABRIC_PROCESSOR)
+        .toProvider(
+            new NonHadoopMasterServiceManagerProvider(NonHadoopAppFabricProcessorManager.class));
     mapBinder.addBinding(Constants.Service.DATASET_EXECUTOR)
         .toProvider(new NonHadoopMasterServiceManagerProvider(DatasetExecutorServiceManager.class));
     mapBinder.addBinding(Constants.Service.METADATA_SERVICE)
@@ -144,6 +149,7 @@ private void addHadoopBindings(Binder binder) {
         .to(MetricsProcessorStatusServiceManager.class);
     mapBinder.addBinding(Constants.Service.METRICS).to(MetricsServiceManager.class);
     mapBinder.addBinding(Constants.Service.APP_FABRIC_HTTP).to(AppFabricServiceManager.class);
+    mapBinder.addBinding(Constants.Service.APP_FABRIC_PROCESSOR).to(AppFabricProcessorManager.class);
     mapBinder.addBinding(Constants.Service.DATASET_EXECUTOR)
         .to(DatasetExecutorServiceManager.class);
     mapBinder.addBinding(Constants.Service.METADATA_SERVICE).to(MetadataServiceManager.class);

diff --git a/...rc/main/java/io/cdap/cdap/internal/app/runtime/distributed/AppFabricProcessorManager.java b/...rc/main/java/io/cdap/cdap/internal/app/runtime/distributed/AppFabricProcessorManager.java
@@ -0,0 +1,42 @@
+/*
+ * Copyright © 2025 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.app.runtime.distributed;
+
+import com.google.inject.Inject;
+import io.cdap.cdap.common.conf.CConfiguration;
+import io.cdap.cdap.common.conf.Constants;
+import io.cdap.cdap.common.conf.Constants.AppFabric;
+import io.cdap.cdap.common.twill.AbstractMasterServiceManager;
+import org.apache.twill.api.TwillRunner;
+import org.apache.twill.discovery.DiscoveryServiceClient;
+
+/**
+ * App Fabric Processor Service Management in Distributed Mode.
+ */
+public class AppFabricProcessorManager extends AbstractMasterServiceManager {
+
+  @Inject
+  AppFabricProcessorManager(CConfiguration cConf, TwillRunner twillRunner,
+      DiscoveryServiceClient discoveryClient) {
+    super(cConf, discoveryClient, Constants.Service.APP_FABRIC_PROCESSOR, twillRunner);
+  }
+
+  @Override
+  public String getDescription() {
+    return AppFabric.PROCESSOR_DESCRIPTION;
+  }
+}
diff --git a/...in/java/io/cdap/cdap/internal/app/runtime/monitor/NonHadoopAppFabricProcessorManager.java b/...in/java/io/cdap/cdap/internal/app/runtime/monitor/NonHadoopAppFabricProcessorManager.java
@@ -0,0 +1,41 @@
+/*
+ * Copyright © 2025 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.app.runtime.monitor;
+
+import com.google.inject.Inject;
+import io.cdap.cdap.common.conf.CConfiguration;
+import io.cdap.cdap.common.conf.Constants;
+import io.cdap.cdap.common.twill.AbstractMasterServiceManager;
+import org.apache.twill.api.TwillRunner;
+import org.apache.twill.discovery.DiscoveryServiceClient;
+
+/**
+ * Service for managing app fabric processor service.
+ */
+public class NonHadoopAppFabricProcessorManager extends AbstractMasterServiceManager {
+
+  @Inject
+  NonHadoopAppFabricProcessorManager(CConfiguration cConf, DiscoveryServiceClient discoveryClient,
+      TwillRunner twillRunner) {
+    super(cConf, discoveryClient, Constants.Service.APP_FABRIC_PROCESSOR, twillRunner);
+  }
+
+  @Override
+  public String getDescription() {
+    return Constants.AppFabric.PROCESSOR_DESCRIPTION;
+  }
+}