Merge pull request #28 from Worklytics/s174-new-datastore-lib

S174 : new datastore lib

Author: eschultink

.github/workflows/test-java.yml

@@ -14,7 +14,7 @@ jobs:
           java-version: 17
           distribution: 'adopt'
       - name: Install pipelines-jar
-        run: mvn install:install-file -Dfile=java/lib/pipeline.jar -DgroupId=com.google.appengine.tools -DartifactId=appengine-pipeline -Dversion=0.3+worklytics.3 -Dpackaging=jar
+        run: mvn install:install-file -Dfile=java/lib/pipeline.jar -DgroupId=com.google.appengine.tools -DartifactId=appengine-pipeline -Dversion=0.3+worklytics.4 -Dpackaging=jar
       # init maven settings such maven can consume packages from GitHub, not just maven central
       - uses: s4u/[email protected]
         with:

java/example/src/com/google/appengine/demos/mapreduce/randomcollisions/CollisionFindingServlet.java

@@ -27,9 +27,9 @@
 import java.util.ArrayList;
 import java.util.logging.Logger;
 
-import javax.servlet.http.HttpServlet;
-import javax.servlet.http.HttpServletRequest;
-import javax.servlet.http.HttpServletResponse;
+import jakarta.servlet.http.HttpServlet;
+import jakarta.servlet.http.HttpServletRequest;
+import jakarta.servlet.http.HttpServletResponse;
 
 /**
  * This is an example Map Reduce that demos parallel computation.

java/example/src/com/google/appengine/demos/mapreduce/randomcollisions/UsingPipelineServlet.java

@@ -24,9 +24,9 @@
 import java.util.ArrayList;
 import java.util.logging.Logger;
 
-import javax.servlet.http.HttpServlet;
-import javax.servlet.http.HttpServletRequest;
-import javax.servlet.http.HttpServletResponse;
+import jakarta.servlet.http.HttpServlet;
+import jakarta.servlet.http.HttpServletRequest;
+import jakarta.servlet.http.HttpServletResponse;
 
 /**
  * This is an alternative to {@link CollisionFindingServlet} that uses Pipelines to start the

java/lib/appengine-pipeline-0.3+worklytics.3.jar

@@ -1 +1 @@
-appengine-pipeline-0.3+worklytics.3.jar
+appengine-pipeline-0.3+worklytics.4.jar

java/lib/appengine-pipeline-0.3+worklytics.4.jar

@@ -33,34 +33,16 @@
     <appengine.target.version>[2.0.4, 2.0.23)</appengine.target.version>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
-    <github.global.server>github</github.global.server>
     <jackson.version>[2.7,3.0)</jackson.version> <!-- 2.7+ is needed for JDK8 data type serialization -->
-    <google.http.client.version>1.44.1</google.http.client.version>
+    <google.http.client.version>[1.43, 2.0)</google.http.client.version>
   </properties>
-  <repositories>
-    <repository>
-      <id>github</id>
-      <name>Apache Maven Packages by Worklytics</name>
-      <!-- /packages portion is magic, not documented directly anywhere; docs suggest to put a specific repoId, which
-           seems weird as we want to make ALL packages published under Worklytics organization available to maven -->
-      <url>https://maven.pkg.github.com/worklytics/packages</url>
-    </repository>
-  </repositories>
-
-  <distributionManagement>
-    <repository>
-      <id>github</id>
-      <name>Mapreduce for Google App Engine by Worklytics</name>
-      <url>https://maven.pkg.github.com/Worklytics/appengine-mapreduce</url>
-    </repository>
-  </distributionManagement>
 
   <dependencyManagement>
     <dependencies>
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>libraries-bom</artifactId>
-        <version>5.1.0</version>
+        <version>26.27.0</version>
         <type>pom</type>
         <scope>import</scope>
       </dependency>
@@ -86,6 +68,16 @@
           <target>17</target>
         </configuration>
       </plugin>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-surefire-plugin</artifactId>
+        <configuration>
+          <environmentVariables>
+            <!-- allows DatastoreOptions.getDefaultInstance() to work in tests -->
+            <GOOGLE_CLOUD_PROJECT>test-project</GOOGLE_CLOUD_PROJECT>
+          </environmentVariables>
+        </configuration>
+      </plugin>
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-jar-plugin</artifactId>
@@ -154,10 +146,14 @@
         </exclusion>
       </exclusions>
     </dependency>
+    <dependency>
+      <groupId>com.google.cloud</groupId>
+      <artifactId>google-cloud-datastore</artifactId>
+    </dependency>
     <dependency>
       <groupId>com.google.appengine.tools</groupId>
       <artifactId>appengine-pipeline</artifactId>
-      <version>0.3+worklytics.3</version>
+      <version>0.3+worklytics.4</version>
     </dependency>
     <dependency>
       <groupId>com.googlecode.charts4j</groupId>
@@ -172,7 +168,7 @@
     <dependency>
       <groupId>com.google.protobuf</groupId>
       <artifactId>protobuf-java</artifactId>
-      <version>3.16.3</version>
+      <version>3.24.4</version>
     </dependency>
     <dependency>
       <groupId>com.google.guava</groupId>
@@ -192,7 +188,7 @@
     <dependency>
       <groupId>com.google.api-client</groupId>
       <artifactId>google-api-client</artifactId>
-      <version>[1.25,2.0)</version>
+      <version>[2.0,3.0)</version>
       <exclusions>
         <exclusion>
           <groupId>com.google.guava</groupId>
@@ -203,7 +199,7 @@
     <dependency>
       <groupId>com.google.api-client</groupId>
       <artifactId>google-api-client-appengine</artifactId>
-      <version>[1.25,2.0)</version>
+      <version>[2.0,3.0)</version>
       <exclusions>
         <exclusion>
           <groupId>com.google.guava</groupId>
@@ -248,6 +244,28 @@
       <artifactId>guava-retrying</artifactId>
       <version>2.0.0</version>
     </dependency>
+    <dependency>
+      <groupId>jakarta.servlet</groupId>
+      <artifactId>jakarta.servlet-api</artifactId>
+      <version>6.0.0</version>
+      <scope>provided</scope> <!-- if deploying to GAE standard with legacy apis enabled -->
+    </dependency>
+
+    <!-- Dagger - Dependency Inject FW - https://github.com/google/dagger -->
+    <!-- TBC: if micronaut, that provides compile-time DI too ... -->
+    <!-- also, is this very extensible? believe this will in effect require users to provide Dagger 2 module FQN to @Injectable to leverage this -->
+    <dependency>
+      <groupId>com.google.dagger</groupId>
+      <artifactId>dagger</artifactId>
+      <version>2.51.1</version>
+      <scope>compile</scope>
+    </dependency>
+    <dependency>
+      <groupId>com.google.dagger</groupId>
+      <artifactId>dagger-compiler</artifactId>
+      <version>2.51.1</version>
+      <scope>compile</scope>
+    </dependency>
 
     <!-- Test Dependencies -->
     <dependency>
@@ -284,6 +302,11 @@
       <artifactId>junit-jupiter-engine</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.junit.jupiter</groupId>
+      <artifactId>junit-jupiter-params</artifactId>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>org.easymock</groupId>
       <artifactId>easymock</artifactId>

java/lib/pipeline.jar

@@ -10,6 +10,7 @@
 import com.google.appengine.tools.mapreduce.impl.pipeline.ExamineStatusAndReturnResult;
 import com.google.appengine.tools.mapreduce.impl.pipeline.ResultAndStatus;
 import com.google.appengine.tools.mapreduce.impl.pipeline.ShardedJob;
+import com.google.appengine.tools.mapreduce.impl.shardedjob.ShardedJobService;
 import com.google.appengine.tools.mapreduce.impl.shardedjob.ShardedJobServiceFactory;
 import com.google.appengine.tools.mapreduce.impl.shardedjob.ShardedJobSettings;
 import com.google.appengine.tools.pipeline.FutureValue;
@@ -19,10 +20,13 @@
 import com.google.appengine.tools.pipeline.PipelineServiceFactory;
 import com.google.appengine.tools.pipeline.PromisedValue;
 import com.google.appengine.tools.pipeline.Value;
+import com.google.cloud.datastore.Datastore;
+import com.google.cloud.datastore.DatastoreOptions;
 import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 
+import javax.inject.Inject;
 import java.io.IOException;
 import java.util.List;
 import java.util.concurrent.CancellationException;
@@ -50,6 +54,10 @@ public MapJob(MapSpecification<I, O, R> specification, MapSettings settings) {
     this.settings = settings;
   }
 
+  @Inject
+  transient ShardedJobService shardedJobService;
+
+
   public static final String DEFAULT_QUEUE_NAME = "default";
 
   /**
@@ -76,7 +84,12 @@ public Value<MapReduceResult<R>> run() {
             + " job, using 'default'");
         queue = DEFAULT_QUEUE_NAME;
       }
-      settings = new MapReduceSettings.Builder().setWorkerQueueName(queue).build();
+      settings = new MapReduceSettings.Builder()
+        .setProjectId(settings.getProjectId())
+        .setDatabaseId(settings.getDatabaseId())
+        .setNamespace(settings.getNamespace())
+        .setWorkerQueueName(queue)
+        .build();
     }
     String jobId = getJobKey().getName();
     Context context = new BaseContext(jobId);
@@ -104,7 +117,7 @@ public Value<MapReduceResult<R>> run() {
     WorkerController<I, O, R, MapOnlyMapperContext<O>> workerController = new WorkerController<>(
         jobId, new CountersImpl(), output, resultAndStatus.getHandle());
     ShardedJob<?> shardedJob =
-        new ShardedJob<>(jobId, mapTasks.build(), workerController, shardedJobSettings);
+        new ShardedJob<>(settings.getDatastoreOptions(), jobId, mapTasks.build(), workerController, shardedJobSettings);
     FutureValue<Void> shardedJobResult = futureCall(shardedJob, settings.toJobSettings());
     JobSetting[] jobSetting = settings.toJobSettings(waitFor(shardedJobResult),
             statusConsoleUrl(shardedJobSettings.getMapReduceStatusUrl()), maxAttempts(1));
@@ -116,7 +129,9 @@ public Value<MapReduceResult<R>> run() {
    */
   public Value<MapReduceResult<R>> handleException(CancellationException ex) {
     String mrJobId = getJobKey().getName();
-    ShardedJobServiceFactory.getShardedJobService().abortJob(mrJobId);
+    Datastore datastore = DatastoreOptions.getDefaultInstance().toBuilder()
+        .setNamespace(settings.getNamespace()).build().getService();
+    shardedJobService.abortJob(datastore, mrJobId);
     return null;
   }