S196 sharded jobs queue

java/appengine-pipeline-0.3+worklytics.10-pom.xml → java/appengine-pipeline-0.3+worklytics.11-pom.xml

@@ -10,7 +10,7 @@
   <!-- just rename the version and copy over -->
   <!-- follow semver, with suggested guidance for versioning fork of OSS -->
   <!-- see https://gofore.com/en/best-practices-for-forking-a-git-repo/ -->
-  <version>0.3+worklytics.10</version>
+  <version>0.3+worklytics.11</version>
   <packaging>jar</packaging>
   <licenses>
     <license>

java/pom.xml

@@ -8,7 +8,7 @@
   <url>https://github.com/Worklytics/appengine-pipelines/</url>
   <!-- follow semver, with suggested guidance for versioning fork of OSS -->
   <!-- see https://gofore.com/en/best-practices-for-forking-a-git-repo/ -->
-  <version>0.3+worklytics.10</version>
+  <version>0.3+worklytics.11</version>
   <packaging>jar</packaging>
   <licenses>
     <license>

java/src/main/java/com/google/appengine/tools/mapreduce/MapJob.java

@@ -17,6 +17,7 @@
 import com.google.appengine.tools.pipeline.JobSetting;
 import com.google.appengine.tools.pipeline.PromisedValue;
 import com.google.appengine.tools.pipeline.Value;
+import com.google.appengine.tools.pipeline.impl.PipelineManager;
 import com.google.cloud.datastore.DatastoreOptions;
 import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
@@ -69,10 +70,10 @@ public Value<MapReduceResult<R>> run() {
     ShardedJobAbstractSettings settings = this.settings;
     if (settings.getWorkerQueueName() == null) {
       String queue = getOnQueue();
-      if (queue == null) {
+      if (queue == null || DEFAULT_QUEUE_NAME.equals(queue)) {
+        queue = PipelineManager.ConfigProperty.INCREMENTAL_TASK_DEFAULT_QUEUE.getValue().orElse(DEFAULT_QUEUE_NAME);
         log.warning("workerQueueName is null and current queue is not available in the pipeline"
-            + " job, using 'default'");
-        queue = DEFAULT_QUEUE_NAME;
+          + " job, using " + queue);
       }
       if (settings instanceof MapSettings) {
         settings = ((MapSettings) settings).withWorkerQueueName(queue);

java/src/main/java/com/google/appengine/tools/mapreduce/MapReduceJob.java

@@ -13,6 +13,7 @@
 import com.google.appengine.tools.mapreduce.inputs.GoogleCloudStorageLineInput;
 import com.google.appengine.tools.mapreduce.outputs.GoogleCloudStorageFileOutput;
 import com.google.appengine.tools.pipeline.*;
+import com.google.appengine.tools.pipeline.impl.PipelineManager;
 import com.google.appengine.tools.pipeline.impl.backend.AppEngineEnvironment;
 import com.google.cloud.datastore.Datastore;
 import com.google.cloud.datastore.DatastoreOptions;
@@ -34,6 +35,8 @@
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
+import static com.google.appengine.tools.pipeline.impl.PipelineManager.DEFAULT_QUEUE_NAME;
+
 
 /**
  * A Pipeline job that runs a MapReduce.
@@ -568,9 +571,9 @@ public Value<MapReduceResult<R>> run() {
     if (settings.getWorkerQueueName() == null) {
       String queue = getOnQueue();
       if (queue == null) {
+        queue = PipelineManager.ConfigProperty.INCREMENTAL_TASK_DEFAULT_QUEUE.getValue().orElse(DEFAULT_QUEUE_NAME);
         log.warning("workerQueueName is null and current queue is not available in the pipeline"
-            + " job, using 'default'");
-        queue = "default";
+          + " job, using " + queue);
       }
       settings = settings.toBuilder().workerQueueName(queue).build();
     }

java/src/main/java/com/google/appengine/tools/pipeline/impl/PipelineManager.java

@@ -53,7 +53,9 @@
 import com.google.common.collect.Iterables;
 import com.google.common.util.concurrent.Uninterruptibles;
 import lombok.AllArgsConstructor;
+import lombok.Getter;
 import lombok.NonNull;
+import lombok.RequiredArgsConstructor;
 import lombok.extern.java.Log;
 
 import javax.inject.Inject;
@@ -81,6 +83,14 @@ public class PipelineManager implements PipelineRunner, PipelineOrchestrator {
 
   public static final String DEFAULT_QUEUE_NAME = "default";
 
+  @RequiredArgsConstructor
+  @Getter
+  public enum ConfigProperty implements com.google.appengine.tools.pipeline.util.ConfigProperty {
+    INCREMENTAL_TASK_DEFAULT_QUEUE("com.google.appengine.tools.pipeline.INCREMENTAL_TASK_DEFAULT_QUEUE");
+
+    final String propertyName;
+  }
+
   final Provider<PipelineService> pipelineServiceProvider;
   final ShardedJobRunner shardedJobRunner;
   final PipelineBackEnd backEnd;
@@ -420,8 +430,8 @@ public void stopJob(@NonNull JobRunId jobHandle) throws NoSuchObjectException {
 
   @Override
   public <I, O, R> JobRunId start(MapSpecification<I, O, R> specification, MapSettings settings) {
-    if (settings.getWorkerQueueName() == null) {
-      settings = settings.toBuilder().workerQueueName(DEFAULT_QUEUE_NAME).build();
+    if (settings.getWorkerQueueName() == null || DEFAULT_QUEUE_NAME.equals(settings.getWorkerQueueName())) {
+      settings = settings.toBuilder().workerQueueName(shardedJobDefaultQueueName()).build();
     }
     return startNewPipeline(settings.toJobSettings(), new MapJob<>(specification, settings));
   }
@@ -430,12 +440,14 @@ public <I, O, R> JobRunId start(MapSpecification<I, O, R> specification, MapSett
   public <I, K, V, O, R> JobRunId start(@NonNull MapReduceSpecification<I, K, V, O, R> specification,
                                         @NonNull MapReduceSettings settings) {
     if (settings.getWorkerQueueName() == null) {
-      settings =  settings.toBuilder().workerQueueName(DEFAULT_QUEUE_NAME).build();
+      settings =  settings.toBuilder().workerQueueName(shardedJobDefaultQueueName()).build();
     }
     return startNewPipeline(settings.toJobSettings(), new MapReduceJob<>(specification, settings));
   }
 
-
+  private String shardedJobDefaultQueueName() {
+    return ConfigProperty.INCREMENTAL_TASK_DEFAULT_QUEUE.getValue().orElse(DEFAULT_QUEUE_NAME);
+  }
 
   public void cancelJob(@NonNull JobRunId jobHandle) throws NoSuchObjectException {
     Key key = JobRecord.keyFromPipelineHandle(jobHandle);

java/src/main/java/com/google/appengine/tools/pipeline/impl/backend/AppEngineServicesServiceImpl.java

@@ -48,6 +48,11 @@ enum ConfigProperty implements com.google.appengine.tools.pipeline.util.ConfigPr
      */
     DEFAULT_PIPELINES_SERVICE,
     ;
+
+    @Override
+    public String getPropertyName() {
+      return this.name();
+    }
   }
 
 

java/src/main/java/com/google/appengine/tools/pipeline/impl/backend/CloudTasksTaskQueue.java

@@ -41,6 +41,11 @@ public class CloudTasksTaskQueue implements PipelineTaskQueue {
   enum ConfigProperty implements com.google.appengine.tools.pipeline.util.ConfigProperty {
     CLOUDTASKS_QUEUE_LOCATION,
     ;
+
+    @Override
+    public String getPropertyName() {
+      return name();
+    }
   }
 
   @NonNull

java/src/main/java/com/google/appengine/tools/pipeline/impl/servlets/PipelineServlet.java

@@ -20,12 +20,16 @@
 import com.google.appengine.tools.pipeline.di.JobRunServiceComponentContainer;
 import com.google.appengine.tools.pipeline.util.Pair;
 import com.google.common.annotations.VisibleForTesting;
-import lombok.Setter;
 
 import javax.servlet.ServletException;
 import javax.servlet.http.HttpServlet;
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
+
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import lombok.Setter;
+
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.Optional;
@@ -39,7 +43,13 @@
  */
 public class PipelineServlet extends HttpServlet {
 
-  public static final String BASE_URL_PROPERTY = "com.google.appengine.tools.pipeline.BASE_URL";
+  @RequiredArgsConstructor
+  @Getter
+  enum ConfigProperty implements com.google.appengine.tools.pipeline.util.ConfigProperty {
+    BASE_URL_PROPERTY("com.google.appengine.tools.pipeline.BASE_URL");
+
+    final String propertyName;
+  }
 
   @Setter(onMethod_ = @VisibleForTesting)
   JobRunServiceComponent component;
@@ -81,7 +91,7 @@ public void doGet(HttpServletRequest req, HttpServletResponse resp)
    * This must match the URL in web.xml
    */
   public static String baseUrl() {
-    String baseURL =  System.getProperty(BASE_URL_PROPERTY, "/_ah/pipeline/");
+    String baseURL = ConfigProperty.BASE_URL_PROPERTY.getValue().orElse( "/_ah/pipeline/");
     if (!baseURL.endsWith("/")) {
       baseURL += "/";
     }

java/src/main/java/com/google/appengine/tools/pipeline/util/ConfigProperty.java

@@ -15,13 +15,10 @@ enum ConfigProperty implements ConfigProperty {
  */
 public interface ConfigProperty {
 
-  String name();
+  String getPropertyName();
 
   default Optional<String> getValue() {
-    return Optional.ofNullable(System.getProperty(name(), System.getenv(name())));
+    return Optional.ofNullable(System.getProperty(getPropertyName(), System.getenv(getPropertyName())));
   }
 
-  default Optional<Boolean> getBoolean() {
-    return getValue().map(Boolean::parseBoolean);
-  }
 }