Merge pull request cdapio#15450 from cdapio/skip-pulling-for-same-hash

Skip pulling a pipeline if the current hash is same

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/sourcecontrol/PullAppsOperation.java

@@ -26,6 +26,7 @@
 import io.cdap.cdap.internal.operation.LongRunningOperation;
 import io.cdap.cdap.internal.operation.LongRunningOperationContext;
 import io.cdap.cdap.internal.operation.OperationException;
+import io.cdap.cdap.proto.ApplicationDetail;
 import io.cdap.cdap.proto.id.ApplicationId;
 import io.cdap.cdap.proto.id.ApplicationReference;
 import io.cdap.cdap.proto.operation.OperationResource;
@@ -41,6 +42,8 @@
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Defines operation for doing SCM Pull for connected repositories.
@@ -54,6 +57,8 @@ public class PullAppsOperation implements LongRunningOperation {
 
   private final Set<ApplicationId> deployed;
 
+  private static final Logger LOG = LoggerFactory.getLogger(PullAppsOperation.class);
+
   /**
    * Only request is passed using AssistedInject. See {@link PullAppsOperationFactory}
    *
@@ -89,8 +94,25 @@ public ListenableFuture<Set<OperationResource>> run(LongRunningOperationContext
 
       // pull and deploy applications one at a time
       scmOpRunner.multiPull(pullReq, response -> {
-        appTobeDeployed.set(new ApplicationReference(context.getRunId().getNamespace(),
-            response.getApplicationName()));
+        ApplicationReference appRef = new ApplicationReference(context.getRunId().getNamespace(),
+            response.getApplicationName());
+        appTobeDeployed.set(appRef);
+
+        try {
+          ApplicationDetail currentDetail = applicationManager.get(appRef);
+          if (currentDetail.getSourceControlMeta() != null && currentDetail.getSourceControlMeta()
+              .getFileHash().equals(response.getApplicationFileHash())) {
+            LOG.trace("Application {} already have same commit, skipping",
+                response.getApplicationName());
+            return;
+          }
+        } catch (NotFoundException e) {
+          // if application does not exist, we create it
+          LOG.trace("Application {} does not exist, creating it", response.getApplicationName());
+        } catch (Exception e) {
+          throw new SourceControlException(e);
+        }
+
         try {
           ApplicationId deployedVersion = applicationManager.deployApp(
               appTobeDeployed.get(), response

cdap-app-fabric/src/test/java/io/cdap/cdap/internal/app/sourcecontrol/PullAppsOperationTest.java

@@ -22,15 +22,18 @@
 import com.google.inject.AbstractModule;
 import com.google.inject.Injector;
 import io.cdap.cdap.api.artifact.ArtifactSummary;
+import io.cdap.cdap.common.NotFoundException;
 import io.cdap.cdap.common.conf.CConfiguration;
 import io.cdap.cdap.internal.AppFabricTestHelper;
 import io.cdap.cdap.internal.operation.LongRunningOperationContext;
 import io.cdap.cdap.internal.operation.OperationException;
+import io.cdap.cdap.proto.ApplicationDetail;
 import io.cdap.cdap.proto.artifact.AppRequest;
 import io.cdap.cdap.proto.id.ApplicationId;
 import io.cdap.cdap.proto.id.ApplicationReference;
 import io.cdap.cdap.proto.id.OperationRunId;
 import io.cdap.cdap.proto.operation.OperationResource;
+import io.cdap.cdap.proto.sourcecontrol.SourceControlMeta;
 import io.cdap.cdap.sourcecontrol.ApplicationManager;
 import io.cdap.cdap.sourcecontrol.RepositoryManager;
 import io.cdap.cdap.sourcecontrol.RepositoryManagerFactory;
@@ -39,6 +42,7 @@
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashSet;
 import java.util.Set;
@@ -55,6 +59,15 @@ public class PullAppsOperationTest {
   private static final AppRequest<?> TEST_APP_REQUEST = new AppRequest<>(
       new ArtifactSummary("name", "version"));
 
+  private static final ApplicationDetail testAppDetails = new ApplicationDetail(
+      "testApp", "v1", "description1", null, null, "conf1", new ArrayList<>(),
+      new ArrayList<>(), new ArrayList<>(), null, null);
+
+  private static final ApplicationDetail testAppDetailsWithGitMeta = new ApplicationDetail(
+      "testApp", "v1", "description1", null, new SourceControlMeta("testHash"), "conf1",
+      new ArrayList<>(),
+      new ArrayList<>(), new ArrayList<>(), null, null);
+
   private static final Gson GSON = new Gson();
 
   private final PullAppsRequest req = new PullAppsRequest(ImmutableSet.of("1", "2", "3", "4"),
@@ -112,6 +125,43 @@ protected void configure() {
   public void testRunSuccess() throws Exception {
     ApplicationManager mockManager = Mockito.mock(ApplicationManager.class);
     PullAppsOperation operation = new PullAppsOperation(this.req, opRunner, mockManager);
+    Mockito.when(mockManager.get(Mockito.any())).thenReturn(testAppDetails);
+
+    Mockito.doAnswer(i -> {
+          ApplicationReference appref = (ApplicationReference) i.getArguments()[0];
+          return appref.app(appref.getApplication());
+        }
+    ).when(mockManager).deployApp(Mockito.any(), Mockito.any());
+
+    gotResources = operation.run(context).get();
+
+    verifyCreatedResources(ImmutableSet.of("1", "2", "3", "4"));
+  }
+
+  @Test
+  public void testRunSuccessNoPull() throws Exception {
+    ApplicationManager mockManager = Mockito.mock(ApplicationManager.class);
+    PullAppsOperation operation = new PullAppsOperation(this.req, opRunner, mockManager);
+
+    Mockito.when(mockManager.get(Mockito.any())).thenReturn(testAppDetailsWithGitMeta);
+
+    Mockito.doAnswer(i -> {
+          ApplicationReference appref = (ApplicationReference) i.getArguments()[0];
+          return appref.app(appref.getApplication());
+        }
+    ).when(mockManager).deployApp(Mockito.any(), Mockito.any());
+
+    gotResources = operation.run(context).get();
+
+    verifyCreatedResources(ImmutableSet.of());
+  }
+
+  @Test
+  public void testRunSuccessNewPipelines() throws Exception {
+    ApplicationManager mockManager = Mockito.mock(ApplicationManager.class);
+    PullAppsOperation operation = new PullAppsOperation(this.req, opRunner, mockManager);
+
+    Mockito.when(mockManager.get(Mockito.any())).thenThrow(new NotFoundException(""));
 
     Mockito.doAnswer(i -> {
           ApplicationReference appref = (ApplicationReference) i.getArguments()[0];
@@ -128,6 +178,7 @@ public void testRunSuccess() throws Exception {
   public void testRunFailedAtFirstApp() throws Exception {
     ApplicationManager mockManager = Mockito.mock(ApplicationManager.class);
     PullAppsOperation operation = new PullAppsOperation(this.req, opRunner, mockManager);
+    Mockito.when(mockManager.get(Mockito.any())).thenReturn(testAppDetails);
 
     Mockito.doThrow(new SourceControlException("")).when(mockManager)
         .deployApp(Mockito.any(), Mockito.any());
@@ -139,6 +190,7 @@ public void testRunFailedAtFirstApp() throws Exception {
   public void testRunFailedAtSecondApp() throws Exception {
     ApplicationManager mockManager = Mockito.mock(ApplicationManager.class);
     PullAppsOperation operation = new PullAppsOperation(this.req, opRunner, mockManager);
+    Mockito.when(mockManager.get(Mockito.any())).thenReturn(testAppDetails);
 
     Mockito.doAnswer(
             i -> {
@@ -161,6 +213,7 @@ public void testRunFailedAtSecondApp() throws Exception {
   public void testRunFailedWhenMarkingLatest() throws Exception {
     ApplicationManager mockManager = Mockito.mock(ApplicationManager.class);
     PullAppsOperation operation = new PullAppsOperation(this.req, opRunner, mockManager);
+    Mockito.when(mockManager.get(Mockito.any())).thenReturn(testAppDetails);
 
     Mockito.doAnswer(
         i -> {