Merge pull request #15368 from cdapio/operation-interface

CDAP-20809 Add LongRunningOperation interface

cdap-app-fabric/src/main/java/io/cdap/cdap/gateway/handlers/OperationRunHttpHandler.java → cdap-app-fabric/src/main/java/io/cdap/cdap/gateway/handlers/OperationHttpHandler.java

@@ -20,7 +20,7 @@
 import com.google.inject.Inject;
 import io.cdap.cdap.common.conf.Constants;
 import io.cdap.cdap.gateway.handlers.util.AbstractAppFabricHttpHandler;
-import io.cdap.cdap.proto.operationrun.OperationRun;
+import io.cdap.cdap.proto.operation.OperationRun;
 import io.cdap.http.HttpHandler;
 import io.cdap.http.HttpResponder;
 import io.netty.handler.codec.http.FullHttpRequest;
@@ -38,12 +38,12 @@
  * The {@link HttpHandler} for handling REST calls to namespace endpoints.
  */
 @Path(Constants.Gateway.API_VERSION_3 + "/namespaces/{namespace-id}/operations")
-public class OperationRunHttpHandler extends AbstractAppFabricHttpHandler {
+public class OperationHttpHandler extends AbstractAppFabricHttpHandler {
 
   private static final Gson GSON = new Gson();
 
   @Inject
-  OperationRunHttpHandler() {
+  OperationHttpHandler() {
   }
 
   /**

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

@@ -0,0 +1,40 @@
+/*
+ * 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.app.sourcecontrol;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import io.cdap.cdap.internal.operation.LongRunningOperation;
+import io.cdap.cdap.internal.operation.LongRunningOperationContext;
+import io.cdap.cdap.proto.operation.OperationError;
+
+/**
+ * Defines operation for doing SCM Pull for connected repositories.
+ * TODO(samik) implement the pull-op
+ **/
+public class PullAppsOperation implements LongRunningOperation {
+
+  private final PullAppsRequest request;
+
+  public PullAppsOperation(PullAppsRequest request) {
+    this.request = request;
+  }
+
+  @Override
+  public ListenableFuture<OperationError> run(LongRunningOperationContext context) {
+    return null;
+  }
+}

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

@@ -0,0 +1,59 @@
+/*
+ * 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.app.sourcecontrol;
+
+import com.google.common.base.Objects;
+import java.util.Set;
+
+/**
+ * Request type for {@link PullAppsOperation}.
+ */
+public class PullAppsRequest {
+
+  private final Set<String> apps;
+
+  /**
+   * Default Constructor.
+   *
+   * @param apps Set of apps to pull.
+   */
+  public PullAppsRequest(Set<String> apps) {
+    this.apps = apps;
+  }
+
+  public Set<String> getApps() {
+    return apps;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+
+    PullAppsRequest that = (PullAppsRequest) o;
+    return Objects.equal(this.getApps(), that.getApps());
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hashCode(getApps());
+  }
+}

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operation/AbstractLongRunningOperationContext.java

@@ -0,0 +1,37 @@
+/*
+ * 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.operation;
+
+import io.cdap.cdap.proto.id.OperationRunId;
+import io.cdap.cdap.proto.operation.OperationType;
+
+/**
+ * Abstract implementation of {@link LongRunningOperationContext} providing shared functionalities.
+ */
+public abstract class AbstractLongRunningOperationContext implements LongRunningOperationContext {
+
+  private final OperationRunId runId;
+  private final OperationType type;
+
+  /**
+   * Default constructor.
+   */
+  protected AbstractLongRunningOperationContext(OperationRunId runid, OperationType operationType) {
+    this.runId = runid;
+    this.type = operationType;
+  }
+}

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operation/LongRunningOperation.java

@@ -0,0 +1,36 @@
+/*
+ * 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.operation;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import io.cdap.cdap.proto.operation.OperationError;
+
+/**
+ * LongRunningOperation represents a long-running asynchronous operation.
+ */
+public interface LongRunningOperation {
+
+  /**
+   * Run the operation with the given request.
+   *
+   * @param request the operation input
+   * @param updateMetadata func to update the metadata of the operation. This would be passed by
+   *     the runner.
+   * @return {@link ListenableFuture} containing the {@link OperationError} for the run
+   */
+  ListenableFuture<OperationError> run(LongRunningOperationContext context);
+}

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operation/LongRunningOperationContext.java

@@ -0,0 +1,53 @@
+/*
+ * 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.operation;
+
+import io.cdap.cdap.proto.id.OperationRunId;
+import io.cdap.cdap.proto.operation.OperationResource;
+import io.cdap.cdap.proto.operation.OperationType;
+import java.util.Set;
+
+/**
+ * Provides the context for the current operation run.
+ */
+public interface LongRunningOperationContext {
+
+  /**
+   * Get the {@link OperationRunId} for the current run.
+   *
+   * @return the current runid
+   */
+  OperationRunId getRunId();
+
+  /**
+   * Get the {@link OperationType} to be used by the runner for loading the right operation class.
+   *
+   * @return the type of the current operation
+   */
+  OperationType getType();
+
+  /**
+   * Used by the {@link LongRunningOperation} to update the resources operated on in the
+   * {@link io.cdap.cdap.proto.operation.OperationMeta} for the run. The input is set as we want all
+   * resources to be unique
+   *
+   * @param resources A set of resources to be updated.
+   *
+   */
+  // TODO Add exceptions based on implementations.
+  void updateOperationResources(Set<OperationResource> resources);
+}

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operations/OperationLifecycleManager.java → cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operation/OperationLifecycleManager.java

@@ -14,10 +14,9 @@
  * the License.
  */
 
-package io.cdap.cdap.internal.operations;
+package io.cdap.cdap.internal.operation;
 
 import com.google.inject.Inject;
-import io.cdap.cdap.internal.app.store.OperationRunDetail;
 import io.cdap.cdap.spi.data.StructuredTableContext;
 import io.cdap.cdap.spi.data.transaction.TransactionRunner;
 import io.cdap.cdap.spi.data.transaction.TransactionRunners;
@@ -46,7 +45,7 @@ public class OperationLifecycleManager {
    *     the caller to identify if there is any further runs left to scan.
    */
   public boolean scanOperations(ScanOperationRunsRequest request, int txBatchSize,
-      Consumer<OperationRunDetail<?>> consumer) throws OperationRunNotFoundException, IOException {
+      Consumer<OperationRunDetail> consumer) throws OperationRunNotFoundException, IOException {
     String lastKey = request.getScanAfter();
     int currentLimit = request.getLimit();
 

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operations/OperationRunAlreadyExistsException.java → cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operation/OperationRunAlreadyExistsException.java

@@ -14,10 +14,10 @@
  * the License.
  */
 
-package io.cdap.cdap.internal.operations;
+package io.cdap.cdap.internal.operation;
 
 import io.cdap.cdap.common.AlreadyExistsException;
-import io.cdap.cdap.proto.operationrun.OperationRunStatus;
+import io.cdap.cdap.proto.operation.OperationRunStatus;
 
 /**
  * Thrown when an operation run already exists.