tests fixes

cdap-app-fabric-tests/pom.xml

@@ -29,6 +29,11 @@ the License.
   <name>CDAP App Fabric Tests</name>
 
   <dependencies>
+    <dependency>
+      <groupId>io.cdap.cdap</groupId>
+      <artifactId>cdap-common</artifactId>
+      <version>${project.version}</version>
+    </dependency>
     <dependency>
       <groupId>io.cdap.cdap</groupId>
       <artifactId>cdap-api</artifactId>

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

@@ -47,6 +47,8 @@
 import io.cdap.cdap.security.spi.authorization.UnauthorizedException;
 import io.cdap.http.BodyConsumer;
 import io.cdap.http.HttpResponder;
+import io.netty.handler.codec.http.DefaultHttpHeaders;
+import io.netty.handler.codec.http.HttpHeaders;
 import io.netty.handler.codec.http.HttpResponseStatus;
 import java.io.File;
 import java.io.FileReader;
@@ -172,7 +174,8 @@ protected ApplicationRecord getApplicationRecord(ApplicationWithPrograms deploye
 
   protected BodyConsumer deployAppFromArtifact(
       final ApplicationId appId,
-      final boolean skipMarkingLatest)
+      final boolean skipMarkingLatest,
+      final AppDeployStrategy appDeployStrategy)
       throws IOException {
     return new AbstractBodyConsumer(
         File.createTempFile("apprequest-" + appId, ".json", tmpDir)) {
@@ -183,8 +186,16 @@ protected void onFinish(HttpResponder responder, File uploadedFile) {
 
           try {
             ApplicationWithPrograms app = applicationLifecycleService.deployApp(appId, appRequest,
-                null, createProgramTerminator(), skipMarkingLatest);
-            responder.sendJson(HttpResponseStatus.OK, GSON.toJson(getApplicationRecord(app)));
+                null, createProgramTerminator(), skipMarkingLatest, appDeployStrategy);
+
+            if (app.isDeploySkipped()) {
+              LOG.debug("Application {} is already deployed", appId);
+            }
+
+            HttpHeaders headers = new DefaultHttpHeaders()
+                .add(Constants.Gateway.APP_DEPLOYMENT_SKIPPED_HEADER,
+                String.valueOf(app.isDeploySkipped()));
+            responder.sendString(HttpResponseStatus.OK, GSON.toJson(getApplicationRecord(app)), headers);
           } catch (DatasetManagementException e) {
             if (e.getCause() instanceof UnauthorizedException) {
               throw (UnauthorizedException) e.getCause();

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

@@ -0,0 +1,37 @@
+/*
+ * Copyright © 2026 CDAP 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.gateway.handlers;
+
+import java.util.Arrays;
+import java.util.stream.Collectors;
+
+/**
+ * Policy to control skipping of duplicate application deployments.
+ */
+public enum AppDeployStrategy {
+  SKIP_ON_NO_CHANGE,
+  ALWAYS_DEPLOY;
+
+  /**
+   * Returns a comma-separated string of all allowed policy values.
+   */
+  public static String getAllowedValues() {
+    return Arrays.stream(values())
+        .map(Enum::name)
+        .collect(Collectors.joining(", "));
+  }
+}

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

@@ -156,17 +156,19 @@ public class AppLifecycleHttpHandler extends AbstractAppLifecycleHttpHandler {
   @AuditPolicy(AuditDetail.REQUEST_BODY)
   public BodyConsumer create(HttpRequest request, HttpResponder responder,
       @PathParam("namespace-id") final String namespaceId,
-      @PathParam("app-id") final String appId)
+      @PathParam("app-id") final String appId,
+      @QueryParam("deployStrategy") @DefaultValue("ALWAYS_DEPLOY") String deployStrategy)
       throws BadRequestException, NamespaceNotFoundException, AccessException {
     String versionId = ApplicationId.DEFAULT_VERSION;
     // If LCM flow is enabled - we generate specific versions of the app.
     if (Feature.LIFECYCLE_MANAGEMENT_EDIT.isEnabled(featureFlagsProvider)) {
       versionId = RunIds.generate().getId();
     }
     ApplicationId applicationId = validateApplicationVersionId(namespaceId, appId, versionId);
+    AppDeployStrategy strategy = parseDeployStrategy(deployStrategy);
 
     try {
-      return deployAppFromArtifact(applicationId);
+      return deployAppFromArtifact(applicationId, strategy);
     } catch (Exception ex) {
       responder.sendString(HttpResponseStatus.INTERNAL_SERVER_ERROR,
           "Deploy failed: " + ex.getMessage());
@@ -215,7 +217,8 @@ public BodyConsumer createAppVersion(HttpRequest request, HttpResponder responde
 
     // If LCM flow is enabled - Ignore the version provided by the user. Treating it the same as deploy without version
     if (Feature.LIFECYCLE_MANAGEMENT_EDIT.isEnabled(featureFlagsProvider)) {
-      return create(request, responder, namespaceId, appId);
+      return create(request, responder, namespaceId, appId,
+          String.valueOf(AppDeployStrategy.ALWAYS_DEPLOY));
     }
 
     ApplicationId applicationId = validateApplicationVersionId(namespaceId, appId, versionId);
@@ -755,6 +758,11 @@ private List<ApplicationId> decodeAndValidateBatchApplicationRecord(NamespaceId
   // the other behavior requires a BodyConsumer and only have one method per path is allowed,
   // so we have to use a BodyConsumer
   private BodyConsumer deployAppFromArtifact(final ApplicationId appId) throws IOException {
+    return deployAppFromArtifact(appId, AppDeployStrategy.ALWAYS_DEPLOY);
+  }
+
+  private BodyConsumer deployAppFromArtifact(final ApplicationId appId,
+      final AppDeployStrategy appDeployStrategy) throws IOException {
     // Perform auth checks outside BodyConsumer as only the first http request containing auth header
     // to populate SecurityRequestContext while http chunk doesn't. BodyConsumer runs in the thread
     // that processes the last http chunk.
@@ -763,7 +771,7 @@ private BodyConsumer deployAppFromArtifact(final ApplicationId appId) throws IOE
         appId.getParent(),
         applicationLifecycleService.decodeUserId(authenticationContext));
     // createTempFile() needs a prefix of at least 3 characters
-    return deployAppFromArtifact(appId, false);
+    return deployAppFromArtifact(appId, false, appDeployStrategy);
   }
 
   private BodyConsumer deployApplication(final HttpResponder responder,
@@ -878,4 +886,14 @@ private ApplicationId validateApplicationVersionId(@Nullable String namespace,
       throws BadRequestException, NamespaceNotFoundException, AccessException {
     return validateApplicationVersionId(validateNamespace(namespace), appId, versionId);
   }
+
+  private static AppDeployStrategy parseDeployStrategy(String strategy) throws BadRequestException {
+    try {
+      return AppDeployStrategy.valueOf(strategy.toUpperCase());
+    } catch (IllegalArgumentException e) {
+      throw new BadRequestException(String.format(
+          "Invalid value '%s' for query parameter 'deployStrategy'. Allowed values are: %s",
+          strategy, AppDeployStrategy.getAllowedValues()));
+    }
+  }
 }

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

@@ -261,7 +261,7 @@ public BodyConsumer create(HttpRequest request, HttpResponder responder,
     }
     ApplicationId applicationId = validateApplicationVersionId(validateNamespace(namespaceId), appId, versionId);
 
-    return deployAppFromArtifact(applicationId, skipMarkingLatest);
+    return deployAppFromArtifact(applicationId, skipMarkingLatest, AppDeployStrategy.ALWAYS_DEPLOY);
   }
 
   /**

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/deploy/pipeline/ApplicationWithPrograms.java

@@ -27,9 +27,10 @@
 public class ApplicationWithPrograms extends ApplicationDeployable {
 
   private final List<ProgramDescriptor> programDescriptors;
+  private final boolean deploySkipped;
 
   public ApplicationWithPrograms(ApplicationDeployable applicationDeployable,
-      Iterable<? extends ProgramDescriptor> programDescriptors) {
+      Iterable<? extends ProgramDescriptor> programDescriptors, boolean deploySkipped) {
     super(applicationDeployable.getArtifactId(), applicationDeployable.getArtifactLocation(),
         applicationDeployable.getApplicationId(), applicationDeployable.getSpecification(),
         applicationDeployable.getExistingAppSpec(),
@@ -40,6 +41,19 @@ public ApplicationWithPrograms(ApplicationDeployable applicationDeployable,
         applicationDeployable.getSourceControlMeta(), applicationDeployable.isUpgrade(),
         applicationDeployable.isSkipMarkingLatest());
     this.programDescriptors = ImmutableList.copyOf(programDescriptors);
+    this.deploySkipped = deploySkipped;
+  }
+
+  public ApplicationWithPrograms(ApplicationDeployable applicationDeployable,
+      Iterable<? extends ProgramDescriptor> programDescriptors) {
+    this(applicationDeployable, programDescriptors, false);
+  }
+
+  /**
+   * Returns true if the deployment was skipped because it was a duplicate request.
+   */
+  public boolean isDeploySkipped() {
+    return deploySkipped;
   }
 
   /**

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/preview/DefaultPreviewManager.java

@@ -191,7 +191,9 @@ public class DefaultPreviewManager extends AbstractIdleService implements Previe
   protected void startUp() throws Exception {
     previewInjector = createPreviewInjector();
     StoreDefinition.createAllTables(previewInjector.getInstance(StructuredTableAdmin.class));
-    metricsCollectionService.startAsync();
+    if (metricsCollectionService.state() == State.NEW) {
+      metricsCollectionService.startAsync();
+    }
     logAppender = previewInjector.getInstance(LogAppender.class);
     logAppender.start();
     LoggingContextAccessor.setLoggingContext(
@@ -202,7 +204,9 @@ protected void startUp() throws Exception {
     logSubscriberService.startAsync().awaitRunning();
     dataSubscriberService = previewInjector.getInstance(PreviewDataSubscriberService.class);
     dataSubscriberService.startAsync().awaitRunning();
-    previewDataCleanupService.startAsync().awaitRunning();
+    if (previewDataCleanupService.state() == State.NEW) {
+      previewDataCleanupService.startAsync().awaitRunning();
+    }
   }
 
   @Override

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/runtime/batch/MapReduceClassLoader.java

@@ -148,8 +148,10 @@ public MapReduceTaskContextProvider getTaskContextProvider() {
     synchronized (this) {
       taskContextProvider = Optional.ofNullable(taskContextProvider)
           .orElseGet(taskContextProviderSupplier::get);
+      if (taskContextProvider.state() == Service.State.NEW) {
+        taskContextProvider.startAsync().awaitRunning();
+      }
     }
-    taskContextProvider.startAsync().awaitRunning();
     return taskContextProvider;
   }
 

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/runtime/distributed/remote/RemoteExecutionTwillController.java

@@ -66,6 +66,7 @@ class RemoteExecutionTwillController implements TwillController {
   private final RemoteProcessController remoteProcessController;
   private final RemoteExecutionService executionService;
   private final long pollCompletedMillis;
+  private final long stopDelayMillis;
   private final boolean terminateWithController;
   private volatile boolean terminateOnServiceStop;
 
@@ -77,6 +78,7 @@ class RemoteExecutionTwillController implements TwillController {
     this.programRunId = programRunId;
     this.runId = RunIds.fromString(programRunId.getRun());
     this.pollCompletedMillis = cConf.getLong(Constants.RuntimeMonitor.POLL_TIME_MS);
+    this.stopDelayMillis = cConf.getLong(Constants.RuntimeMonitor.REMOTE_STOP_DELAY_SECS, 10) * 1000;
 
     // On start up task succeeded, complete the started stage to unblock the onRunning()
     // On start up task failure, mark this controller as terminated with exception
@@ -121,15 +123,18 @@ public void complete() {
     try {
       RuntimeJobStatus status;
       RetryStrategy retryStrategy = RetryStrategies.timeLimit(
-          5, TimeUnit.SECONDS, RetryStrategies.exponentialDelay(500, 2000, TimeUnit.MILLISECONDS));
+          stopDelayMillis, TimeUnit.MILLISECONDS, RetryStrategies.exponentialDelay(500, 2000,
+              TimeUnit.MILLISECONDS));
 
-      // Make sure the remote execution is completed
-      // Give 5 seconds for the remote process to shutdown. After 5 seconds, issues a kill.
+      // Wait for the remote process (e.g. Dataproc job) to complete.
+      // We give 5 sec for the remote process (e.g. DP's master process) to shutdown and another
+      // 5 sec to account for DP's backend propagation delays (CDAP-21219). If we kill/cancel too
+      // early while the job is finishing, it can transition the Dataproc job to an ERROR state .
       long startTime = System.currentTimeMillis();
       while ((status = Retries.callWithRetries(
           remoteProcessController::getStatus, retryStrategy, Exception.class::isInstance))
           == RuntimeJobStatus.RUNNING) {
-        if (System.currentTimeMillis() - startTime >= 5000) {
+        if (System.currentTimeMillis() - startTime >= stopDelayMillis) {
           throw new IllegalStateException(
               "Remote process for " + programRunId + " is still running");
         }

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/runtime/schedule/store/DatasetBasedTimeScheduleStore.java

@@ -32,6 +32,7 @@
 import io.cdap.cdap.spi.data.table.field.Range;
 import io.cdap.cdap.spi.data.transaction.TransactionRunner;
 import io.cdap.cdap.spi.data.transaction.TransactionRunners;
+import io.cdap.cdap.spi.data.transaction.TxRunnable;
 import io.cdap.cdap.store.StoreDefinition;
 import java.io.Externalizable;
 import java.io.IOException;
@@ -164,7 +165,7 @@ static StructuredTable getTimeScheduleStructuredTable(StructuredTableContext con
 
   private void executeDelete(final TriggerKey triggerKey) {
     try {
-      TransactionRunners.run(transactionRunner, context -> {
+      TransactionRunners.run(transactionRunner, (TxRunnable) context -> {
         delete(getTimeScheduleStructuredTable(context), TRIGGER_KEY, triggerKey.getName());
       });
     } catch (Throwable th) {
@@ -174,7 +175,7 @@ private void executeDelete(final TriggerKey triggerKey) {
 
   private void executeDelete(final JobKey jobKey) {
     try {
-      TransactionRunners.run(transactionRunner, context -> {
+      TransactionRunners.run(transactionRunner, (TxRunnable) context -> {
         delete(getTimeScheduleStructuredTable(context), JOB_KEY, jobKey.getName());
       });
     } catch (Throwable t) {
@@ -186,7 +187,7 @@ private void persistChangeOfState(final TriggerKey triggerKey,
       final Trigger.TriggerState newTriggerState) {
     try {
       Preconditions.checkNotNull(triggerKey);
-      TransactionRunners.run(transactionRunner, context -> {
+      TransactionRunners.run(transactionRunner, (TxRunnable) context -> {
         StructuredTable table = getTimeScheduleStructuredTable(context);
         TriggerStatusV2 storedTriggerStatus = readTrigger(table, triggerKey);
         if (storedTriggerStatus != null) {
@@ -210,7 +211,7 @@ private void persistJobAndTrigger(final JobDetail newJob, final OperableTrigger
         triggerState = super.getTriggerState(newTrigger.getKey());
       }
       final Trigger.TriggerState finalTriggerState = triggerState;
-      TransactionRunners.run(transactionRunner, context -> {
+      TransactionRunners.run(transactionRunner, (TxRunnable) context -> {
         StructuredTable table = getTimeScheduleStructuredTable(context);
         if (newJob != null) {
           persistJob(table, newJob);
@@ -285,7 +286,7 @@ private void readSchedulesFromPersistentStore() throws Exception {
     final List<JobDetail> jobs = Lists.newArrayList();
     final List<TriggerStatusV2> triggers = Lists.newArrayList();
 
-    TransactionRunners.run(transactionRunner, context -> {
+    TransactionRunners.run(transactionRunner, (TxRunnable) context -> {
       StructuredTable table = getTimeScheduleStructuredTable(context);
       try (CloseableIterator<StructuredRow> iterator =
           table.scan(Range.singleton(getScanPrefix(JOB_KEY)), Integer.MAX_VALUE)) {