[Draft] Simplified upgrades WIP branch.

cdap-app-fabric/src/main/java/io/cdap/cdap/app/guice/AppFabricServiceRuntimeModule.java

@@ -44,6 +44,7 @@
 import io.cdap.cdap.app.mapreduce.LocalMRJobInfoFetcher;
 import io.cdap.cdap.app.mapreduce.MRJobInfoFetcher;
 import io.cdap.cdap.app.store.Store;
+import io.cdap.cdap.app.upgrade.UpgradeModule;
 import io.cdap.cdap.common.conf.CConfiguration;
 import io.cdap.cdap.common.conf.Constants;
 import io.cdap.cdap.common.conf.Constants.AppFabric;
@@ -158,7 +159,9 @@
 import io.cdap.cdap.messaging.server.FetchHandler;
 import io.cdap.cdap.messaging.server.MetadataHandler;
 import io.cdap.cdap.messaging.server.StoreHandler;
+import io.cdap.cdap.metadata.DefaultMetadataAdmin;
 import io.cdap.cdap.metadata.LocalPreferencesFetcherInternal;
+import io.cdap.cdap.metadata.MetadataAdmin;
 import io.cdap.cdap.metadata.PreferencesFetcher;
 import io.cdap.cdap.pipeline.PipelineFactory;
 import io.cdap.cdap.scheduler.CoreSchedulerService;
@@ -239,6 +242,7 @@ public Module getInMemoryModules() {
         new MasterCredentialProviderModule(),
         new OperationModule(),
         new DataStorageAeadEncryptionModule(),
+        new UpgradeModule(),
         BootstrapModules.getInMemoryModule(),
         new AbstractModule() {
           @Override
@@ -299,6 +303,7 @@ public Module getStandaloneModules() {
         new MasterCredentialProviderModule(),
         new OperationModule(),
         new DataStorageAeadEncryptionModule(),
+        new UpgradeModule(),
         serviceTypes.contains(ServiceType.PROCESSOR) ? BootstrapModules.getFileBasedModule() :
             BootstrapModules.getNoOpModule(),
         new AbstractModule() {
@@ -360,6 +365,7 @@ public Module getDistributedModules() {
         new MasterCredentialProviderModule(),
         new OperationModule(),
         new DataStorageAeadEncryptionModule(),
+        new UpgradeModule(),
         serviceTypes.contains(ServiceType.PROCESSOR) ? BootstrapModules.getFileBasedModule() :
             BootstrapModules.getNoOpModule(),
         new AbstractModule() {
@@ -374,6 +380,7 @@ protected void configure() {
             bind(StorageProviderNamespaceAdmin.class)
                 .to(DistributedStorageProviderNamespaceAdmin.class);
             bind(UGIProvider.class).toProvider(UgiProviderProvider.class);
+            bind(MetadataAdmin.class).to(DefaultMetadataAdmin.class);
 
             bind(ProgramRunDispatcher.class).to(RemoteProgramRunDispatcher.class)
                 .in(Scopes.SINGLETON);

cdap-app-fabric/src/main/java/io/cdap/cdap/app/preview/PreviewRunnerModule.java

@@ -28,6 +28,7 @@
 import io.cdap.cdap.app.deploy.ManagerFactory;
 import io.cdap.cdap.app.guice.AppFabricServiceRuntimeModule;
 import io.cdap.cdap.app.store.Store;
+import io.cdap.cdap.app.upgrade.UpgradeModule;
 import io.cdap.cdap.common.conf.CConfiguration;
 import io.cdap.cdap.common.guice.LocalLocationModule;
 import io.cdap.cdap.common.namespace.NamespaceAdmin;
@@ -204,6 +205,7 @@ protected void configure() {
     expose(OwnerAdmin.class);
 
     bind(CapabilityReader.class).to(CapabilityStatusStore.class);
+    install(new UpgradeModule());
   }
 
   /**

cdap-app-fabric/src/main/java/io/cdap/cdap/app/upgrade/UpgradeManager.java

@@ -0,0 +1,36 @@
+/*
+ * 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.app.upgrade;
+
+import io.cdap.cdap.proto.id.NamespaceId;
+import io.cdap.cdap.proto.upgrade.ApplicationUpgradeDetail;
+import java.util.List;
+
+/**
+ * Manager for all upgrade related operations.
+ */
+public interface UpgradeManager {
+
+  /**
+   * Lists upgrade details for applications in a namespace.
+   *
+   * @param namespace the namespace in which applications are searched.
+   * @return A list of {@link ApplicationUpgradeDetail} containing application and current/latest
+   *     version details for the application artifact and the plugins.
+   */
+  List<ApplicationUpgradeDetail> listUpgrades(NamespaceId namespace) throws Exception;
+}

cdap-app-fabric/src/main/java/io/cdap/cdap/app/upgrade/UpgradeModule.java

@@ -0,0 +1,36 @@
+/*
+ * 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.app.upgrade;
+
+import com.google.inject.AbstractModule;
+import com.google.inject.Scopes;
+import io.cdap.cdap.internal.app.upgrade.ApplicationPluginMappingFetcher;
+import io.cdap.cdap.internal.app.upgrade.DefaultUpgradeManager;
+import io.cdap.cdap.internal.app.upgrade.MetadataApplicationPluginMappingFetcher;
+
+/**
+ * Module that configures Upgrade related classes.
+ */
+public class UpgradeModule extends AbstractModule {
+
+  @Override
+  protected void configure() {
+    bind(UpgradeManager.class).to(DefaultUpgradeManager.class).in(Scopes.SINGLETON);
+    bind(ApplicationPluginMappingFetcher.class).to(MetadataApplicationPluginMappingFetcher.class)
+        .in(Scopes.SINGLETON);
+  }
+}

cdap-app-fabric/src/main/java/io/cdap/cdap/gateway/handlers/AppLifecycleHttpHandler.java

@@ -69,6 +69,8 @@
 import io.cdap.cdap.proto.id.KerberosPrincipalId;
 import io.cdap.cdap.proto.id.NamespaceId;
 import io.cdap.cdap.proto.security.StandardPermission;
+import io.cdap.cdap.proto.upgrade.ListUpgradeRequest;
+import io.cdap.cdap.proto.upgrade.ListUpgradeResponse;
 import io.cdap.cdap.security.spi.authentication.AuthenticationContext;
 import io.cdap.cdap.security.spi.authorization.AccessEnforcer;
 import io.cdap.cdap.security.spi.authorization.UnauthorizedException;
@@ -305,7 +307,7 @@
     }
     if (nameFilter != null && !nameFilter.isEmpty()) {
       if (nameFilterType != null) {
         switch (nameFilterType) {
           case EQUALS:
             builder.setApplicationReference(new ApplicationReference(namespaceId, nameFilter));
             break;
@@ -498,7 +500,7 @@
   /**
    * Gets count for all application the namespace.
    *
    * <p>This API returns the count for latest applications only.</>
    */
   @GET
   @Path("/apps/count")
@@ -641,6 +643,30 @@
     }
   }
 
+  /**
+   * Lists upgrade details for all pipelines in the namespace.
+   *
+   * <p>
+   * The response will be of type {@link ListUpgradeResponse} which contains a list of
+   * {@link io.cdap.cdap.proto.upgrade.ApplicationUpgradeDetail} containing pipeline and plugin
+   * upgrade details.
+   * </p>
+   */
+  @POST
+  @Path("/upgrade/list")
+  @AuditPolicy(AuditDetail.REQUEST_BODY)
+  public void listApplicationUpgrades(FullHttpRequest request, HttpResponder responder,
+      @PathParam("namespace-id") String namespace) throws Exception {
+    NamespaceId namespaceId = validateNamespace(namespace);
+    ListUpgradeRequest upgradeRequest =
+        DECODE_GSON.fromJson(request.content().toString(StandardCharsets.UTF_8),
+            new TypeToken<ListUpgradeRequest>() {
+            }.getType());
+    LOG.info("received list upgrade request {}", upgradeRequest);
+    responder.sendJson(HttpResponseStatus.OK,
+        GSON.toJson(applicationLifecycleService.listUpgradeResponse(namespaceId)));
+  }
+
   /**
    * Gets {@link ApplicationDetail} for a set of applications. It expects a post body as a array of
    * object, with each object specifying the applciation id and an optional version. E.g.

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/services/ApplicationLifecycleService.java

@@ -52,6 +52,7 @@
 import io.cdap.cdap.app.store.ApplicationFilter;
 import io.cdap.cdap.app.store.ScanApplicationsRequest;
 import io.cdap.cdap.app.store.Store;
+import io.cdap.cdap.app.upgrade.UpgradeManager;
 import io.cdap.cdap.common.ApplicationNotFoundException;
 import io.cdap.cdap.common.ArtifactAlreadyExistsException;
 import io.cdap.cdap.common.ArtifactNotFoundException;
@@ -113,6 +114,7 @@
 import io.cdap.cdap.proto.security.Principal;
 import io.cdap.cdap.proto.security.StandardPermission;
 import io.cdap.cdap.proto.sourcecontrol.SourceControlMeta;
+import io.cdap.cdap.proto.upgrade.ListUpgradeResponse;
 import io.cdap.cdap.security.impersonation.EntityImpersonator;
 import io.cdap.cdap.security.impersonation.Impersonator;
 import io.cdap.cdap.security.impersonation.OwnerAdmin;
@@ -182,6 +184,7 @@ public class ApplicationLifecycleService extends AbstractIdleService {
   private final int batchSize;
   private final MetricsCollectionService metricsCollectionService;
   private final FeatureFlagsProvider featureFlagsProvider;
+  private final UpgradeManager upgradeManager;
 
   /**
    * Construct the ApplicationLifeCycleService with service factory and cConf coming from guice
@@ -197,7 +200,7 @@ public ApplicationLifecycleService(CConfiguration cConf,
       AccessEnforcer accessEnforcer, AuthenticationContext authenticationContext,
       MessagingService messagingService, Impersonator impersonator,
       CapabilityReader capabilityReader,
-      MetricsCollectionService metricsCollectionService) {
+      MetricsCollectionService metricsCollectionService, UpgradeManager upgradeManager) {
     this.cConf = cConf;
     this.appUpdateSchedules = cConf.getBoolean(Constants.AppFabric.APP_UPDATE_SCHEDULES,
         Constants.AppFabric.DEFAULT_APP_UPDATE_SCHEDULES);
@@ -215,6 +218,7 @@ public ApplicationLifecycleService(CConfiguration cConf,
     this.authenticationContext = authenticationContext;
     this.impersonator = impersonator;
     this.capabilityReader = capabilityReader;
+    this.upgradeManager = upgradeManager;
     this.adminEventPublisher = new AdminEventPublisher(cConf,
         new MultiThreadMessagingContext(messagingService));
     this.metricsCollectionService = metricsCollectionService;
@@ -665,6 +669,17 @@ public ApplicationId upgradeApplication(ApplicationId appId,
     return updateApplicationByArtifact(appId, currentSpec, allowedArtifactScopes, allowSnapshot);
   }
 
+  /**
+   * Lists the upgrade details for all applications in the namespace.
+   *
+   * @param namespace namespace for which upgrades are computed.
+   * @return A {@link ListUpgradeResponse} object that will be returned from the HTTP client.
+   * @throws Exception if there was any exception while listing upgrades.
+   */
+  public ListUpgradeResponse listUpgradeResponse(NamespaceId namespace) throws Exception {
+    return new ListUpgradeResponse(upgradeManager.listUpgrades(namespace));
+  }
+
   /**
    * Upgrades the latest version of an existing application by upgrading application artifact
    * versions and plugin artifact versions.

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/upgrade/ApplicationPluginMapping.java

@@ -0,0 +1,62 @@
+/*
+ * 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.upgrade;
+
+import io.cdap.cdap.proto.id.ApplicationId;
+import io.cdap.cdap.proto.id.PluginId;
+import java.util.Objects;
+
+/**
+ * Mapping for the latest application version and its plugin.
+ */
+public class ApplicationPluginMapping {
+
+  // Version less application Id.
+  private final ApplicationId applicationId;
+  private final PluginId pluginId;
+
+  public ApplicationPluginMapping(ApplicationId applicationId, PluginId pluginId) {
+    this.applicationId = applicationId;
+    this.pluginId = pluginId;
+  }
+
+  public ApplicationId getApplicationId() {
+    return applicationId;
+  }
+
+  public PluginId getPluginId() {
+    return pluginId;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof ApplicationPluginMapping)) {
+      return false;
+    }
+    ApplicationPluginMapping that = (ApplicationPluginMapping) o;
+    return Objects.equals(applicationId, that.applicationId) && Objects.equals(
+        pluginId, that.pluginId);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(applicationId, pluginId);
+  }
+}

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/upgrade/ApplicationPluginMappingFetcher.java

@@ -0,0 +1,38 @@
+/*
+ * 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.upgrade;
+
+import io.cdap.cdap.proto.id.NamespaceId;
+import java.util.List;
+
+/**
+ * Fetcher for application to plugin mapping.
+ */
+public interface ApplicationPluginMappingFetcher {
+
+  /**
+   * Fetches application to plugin mapping for a namespace.
+   *
+   * @param namespace the namespace for which mapping will be fetched.
+   * @return a list of type {@link ApplicationPluginMapping}. An empty list will be returned if no
+   *     applications are found.
+   * @throws Exception the exception during fetching of application plugin mapping.
+   */
+  List<ApplicationPluginMapping> fetchApplicationPluginMapping(NamespaceId namespace)
+      throws Exception;
+
+}