Upgrade App Engine dev task queue emulation to Quartz 2.4.1 which is the supported version going forward.

PiperOrigin-RevId: 899078838
Change-Id: Id81b6a5e26d3b587b715692edbb288063247ac39

Author: ludoch

api_dev/pom.xml

@@ -112,7 +112,7 @@
       <artifactId>jsoup</artifactId>
     </dependency>
     <dependency>
-      <groupId>quartz</groupId>
+      <groupId>org.quartz-scheduler</groupId>
       <artifactId>quartz</artifactId>
     </dependency>
     <dependency>
@@ -244,3 +244,4 @@
     </plugins>
   </build>
 </project>
+

api_dev/src/main/java/com/google/appengine/api/taskqueue/dev/DevPushQueue.java

@@ -16,6 +16,12 @@
 
 package com.google.appengine.api.taskqueue.dev;
 
+import static org.quartz.JobBuilder.newJob;
+import static org.quartz.JobKey.jobKey;
+import static org.quartz.TriggerBuilder.newTrigger;
+import static org.quartz.impl.matchers.GroupMatcher.jobGroupEquals;
+import static org.quartz.impl.matchers.GroupMatcher.triggerGroupEquals;
+
 import com.google.appengine.api.taskqueue.dev.QueueStateInfo.TaskStateInfo;
 import com.google.appengine.api.taskqueue_bytes.TaskQueuePb.TaskQueueAddRequest;
 import com.google.appengine.api.taskqueue_bytes.TaskQueuePb.TaskQueueAddRequest.Header;
@@ -28,19 +34,24 @@
 import com.google.apphosting.utils.config.QueueXml;
 import com.google.common.flogger.GoogleLogger;
 import com.google.protobuf.ByteString;
+import java.time.Instant;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collections;
 import java.util.Date;
 import java.util.List;
+import java.util.Set;
 import org.quartz.Job;
+import org.quartz.JobDataMap;
 import org.quartz.JobDetail;
 import org.quartz.JobExecutionContext;
 import org.quartz.JobExecutionException;
+import org.quartz.JobKey;
 import org.quartz.Scheduler;
 import org.quartz.SchedulerException;
 import org.quartz.SimpleTrigger;
 import org.quartz.Trigger;
+import org.quartz.impl.JobExecutionContextImpl;
+import org.quartz.spi.OperableTrigger;
 import org.quartz.spi.TriggerFiredBundle;
 
 /**
@@ -89,7 +100,7 @@ Mode getMode() {
           // to the contrary, pausing a job group only pauses jobs that already
           // exist.  We need to make sure all future jobs are paused, and that
           // works if we pause the trigger group.
-          scheduler.pauseTriggerGroup(getQueueName());
+          scheduler.pauseTriggers(triggerGroupEquals(getQueueName()));
         } catch (SchedulerException e) {
           throw new ApiProxy.ApplicationException(ErrorCode.INTERNAL_ERROR_VALUE, e.getMessage());
         }
@@ -112,7 +123,7 @@ private synchronized String scheduleTask(TaskQueueAddRequest.Builder addRequest)
       taskName = genTaskName();
     }
     try {
-      if (scheduler.getJobDetail(taskName, getQueueName()) != null) {
+      if (scheduler.checkExists(jobKey(taskName, getQueueName()))) {
         throw new ApiProxy.ApplicationException(ErrorCode.TASK_ALREADY_EXISTS_VALUE);
       }
     } catch (SchedulerException e) {
@@ -121,19 +132,32 @@ private synchronized String scheduleTask(TaskQueueAddRequest.Builder addRequest)
 
     TaskQueueRetryParameters retryParams = getRetryParameters(addRequest);
     long etaMillis = addRequest.getEtaUsec() / 1000L;
-    SimpleTrigger trigger = new SimpleTrigger(taskName, getQueueName());
-    trigger.setStartTime(new Date(etaMillis));
-    JobDetail jd = newUrlFetchJobDetail(taskName, getQueueName(), addRequest, retryParams);
+
+    JobDataMap jobDataMap =
+        newUrlFetchJobDataMap(taskName, getQueueName(), addRequest, retryParams);
+
+    JobDetail job =
+        newJob(UrlFetchJob.class)
+            .withIdentity(taskName, getQueueName())
+            .usingJobData(jobDataMap)
+            .build();
+
+    Trigger trigger =
+        newTrigger()
+            .withIdentity(taskName, getQueueName())
+            .startAt(Date.from(Instant.ofEpochMilli(etaMillis)))
+            .build();
+
     try {
-      scheduler.scheduleJob(jd, trigger);
+      scheduler.scheduleJob(job, trigger);
     } catch (SchedulerException e) {
       throw new ApiProxy.ApplicationException(ErrorCode.INTERNAL_ERROR_VALUE, e.getMessage());
     }
     return taskName;
   }
 
   // broken out to support testing
-  JobDetail newUrlFetchJobDetail(
+  JobDataMap newUrlFetchJobDataMap(
       String taskName,
       String queueName,
       TaskQueueAddRequest.Builder addRequest,
@@ -143,18 +167,20 @@ JobDetail newUrlFetchJobDetail(
         String host = header.getValue().toStringUtf8();
         if (host.startsWith("localhost:")) {
           return new UrlFetchJobDetail(
-              taskName,
-              queueName,
-              addRequest,
-              "http://" + host,
-              callback,
-              queueXmlEntry,
-              retryParams);
+                  taskName,
+                  queueName,
+                  addRequest,
+                  "http://" + host,
+                  callback,
+                  queueXmlEntry,
+                  retryParams)
+              .getJobDataMap();
         }
       }
     }
     return new UrlFetchJobDetail(
-        taskName, queueName, addRequest, baseUrl, callback, queueXmlEntry, retryParams);
+            taskName, queueName, addRequest, baseUrl, callback, queueXmlEntry, retryParams)
+        .getJobDataMap();
   }
 
   @Override
@@ -177,8 +203,11 @@ TaskQueueAddResponse add(TaskQueueAddRequest.Builder addRequest) {
   }
 
   List<String> getSortedJobNames() throws SchedulerException {
-    String[] jobNames = scheduler.getJobNames(getQueueName());
-    List<String> jobNameList = Arrays.asList(jobNames);
+    Set<JobKey> jobKeys = scheduler.getJobKeys(jobGroupEquals(getQueueName()));
+    List<String> jobNameList = new ArrayList<>();
+    for (JobKey jobKey : jobKeys) {
+      jobNameList.add(jobKey.getName());
+    }
     Collections.sort(jobNameList);
     return jobNameList;
   }
@@ -191,23 +220,27 @@ QueueStateInfo getStateInfo() {
       // Get the names of all jobs belonging to this queue (group).
       for (String jobName : getSortedJobNames()) {
         // Now get job details
-        UrlFetchJobDetail jd = (UrlFetchJobDetail) scheduler.getJobDetail(jobName, getQueueName());
-        if (jd == null) {
+        JobDetail jobDetail = scheduler.getJobDetail(jobKey(jobName, getQueueName()));
+        if (jobDetail == null) {
           // oops, gone, must have already run
           continue;
         }
-        Trigger[] triggers = scheduler.getTriggersOfJob(jobName, getQueueName());
-        if (triggers.length == 0) {
+
+        UrlFetchJobDetail jd = new UrlFetchJobDetail(jobDetail.getJobDataMap());
+
+        List<? extends Trigger> triggers =
+            scheduler.getTriggersOfJob(jobKey(jobName, getQueueName()));
+        if (triggers.isEmpty()) {
           // must have run in between the time we fetched the job detail and the time we fetched the
           // trigger
           continue;
         }
-        if (triggers.length != 1) {
+        if (triggers.size() != 1) {
           throw new IllegalStateException(
               "Multiple triggers for task " + jobName + " in queue " + getQueueName());
         }
-        long execTime = triggers[0].getStartTime().getTime();
-        taskInfoList.add(new TaskStateInfo(jd.getName(), execTime, jd.getAddRequest(), clock));
+        long execTime = triggers.get(0).getStartTime().toInstant().toEpochMilli();
+        taskInfoList.add(new TaskStateInfo(jd.getTaskName(), execTime, jd.getAddRequest(), clock));
       }
     } catch (SchedulerException e) {
       throw new ApiProxy.ApplicationException(ErrorCode.INTERNAL_ERROR_VALUE);
@@ -231,7 +264,7 @@ QueueStateInfo getStateInfo() {
   @Override
   boolean deleteTask(String taskName) {
     try {
-      return scheduler.deleteJob(taskName, getQueueName());
+      return scheduler.deleteJob(jobKey(taskName, getQueueName()));
     } catch (SchedulerException e) {
       throw new ApiProxy.ApplicationException(ErrorCode.INTERNAL_ERROR_VALUE);
     }
@@ -241,20 +274,18 @@ boolean deleteTask(String taskName) {
   @Override
   void flush() {
     try {
-      for (String name : scheduler.getJobNames(getQueueName())) {
-        scheduler.deleteJob(name, getQueueName());
-      }
+      Set<JobKey> jobKeys = scheduler.getJobKeys(jobGroupEquals(getQueueName()));
+      scheduler.deleteJobs(new ArrayList<>(jobKeys));
     } catch (SchedulerException e) {
       throw new ApiProxy.ApplicationException(ErrorCode.INTERNAL_ERROR_VALUE);
     }
   }
 
-  private JobExecutionContext getExecutionContext(UrlFetchJobDetail jobDetail) {
-    Trigger trigger = new SimpleTrigger(jobDetail.getTaskName(), jobDetail.getQueueName());
-    trigger.setJobDataMap(jobDetail.getJobDataMap());
+  private JobExecutionContext getExecutionContext(JobDetail jobDetail, SimpleTrigger trigger) {
     TriggerFiredBundle bundle =
-        new TriggerFiredBundle(jobDetail, trigger, null, false, null, null, null, null);
-    return new JobExecutionContext(scheduler, bundle, null);
+        new TriggerFiredBundle(
+            jobDetail, (OperableTrigger) trigger, null, false, null, null, null, null);
+    return new JobExecutionContextImpl(scheduler, bundle, null);
   }
 
   /**
@@ -269,11 +300,19 @@ boolean runTask(String taskName) {
     Job job;
     JobExecutionContext context;
     try {
-      UrlFetchJobDetail jd = (UrlFetchJobDetail) scheduler.getJobDetail(taskName, getQueueName());
+      JobDetail jd = scheduler.getJobDetail(jobKey(taskName, getQueueName()));
       if (jd == null) {
         return false;
       }
-      context = getExecutionContext(jd);
+      // Reconstruct trigger for execution context - just needs to hold data map
+      SimpleTrigger trigger =
+          (SimpleTrigger)
+              newTrigger()
+                  .withIdentity(taskName, getQueueName())
+                  .usingJobData(jd.getJobDataMap())
+                  .build();
+
+      context = getExecutionContext(jd, trigger);
       job = (Job) jd.getJobClass().newInstance();
     } catch (SchedulerException e) {
       return false;

api_dev/src/main/java/com/google/appengine/api/taskqueue/dev/LocalTaskQueue.java

@@ -63,6 +63,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Properties;
 import java.util.Random;
 import java.util.TreeMap;
 import org.jspecify.annotations.Nullable;
@@ -690,7 +691,13 @@ static Scheduler startScheduler(boolean disableAutoTaskExecution) {
     // TODO: Investigate config options for the scheduler like
     // threadpool size.
     try {
-      Scheduler scheduler = StdSchedulerFactory.getDefaultScheduler();
+      StdSchedulerFactory factory = new StdSchedulerFactory();
+      Properties props = new Properties();
+      props.setProperty("org.quartz.scheduler.instanceName", "AppEngineLocalTaskQueue");
+      props.setProperty("org.quartz.threadPool.class", "org.quartz.simpl.SimpleThreadPool");
+      props.setProperty("org.quartz.threadPool.threadCount", "10");
+      factory.initialize(props);
+      Scheduler scheduler = factory.getScheduler();
       // When a scheduler is first created it is in standby mode, which means
       // it will accept and schedule tasks but won't ever run them.
       if (!disableAutoTaskExecution) {

api_dev/src/main/java/com/google/appengine/api/taskqueue/dev/UrlFetchJob.java

@@ -16,23 +16,29 @@
 
 package com.google.appengine.api.taskqueue.dev;
 
+import static org.quartz.JobBuilder.newJob;
+import static org.quartz.TriggerBuilder.newTrigger;
+
 import com.google.appengine.api.taskqueue_bytes.TaskQueuePb.TaskQueueAddRequest;
 import com.google.appengine.api.taskqueue_bytes.TaskQueuePb.TaskQueueRetryParameters;
 import com.google.appengine.api.urlfetch.URLFetchServicePb.URLFetchRequest;
-import com.google.appengine.api.urlfetch.dev.LocalURLFetchService;
 import com.google.appengine.tools.development.Clock;
 import com.google.appengine.tools.development.LocalServerEnvironment;
 import com.google.apphosting.utils.config.QueueXml;
 import com.google.common.flogger.GoogleLogger;
 import java.text.DecimalFormat;
+import java.time.Instant;
 import java.util.Date;
 import org.quartz.Job;
+import org.quartz.JobDataMap;
+import org.quartz.JobDetail;
 import org.quartz.JobExecutionContext;
 import org.quartz.JobExecutionException;
 import org.quartz.Scheduler;
 import org.quartz.SchedulerException;
 import org.quartz.SimpleTrigger;
 import org.quartz.Trigger;
+import org.quartz.TriggerKey;
 
 /**
  * Quartz {@link Job} implementation that hits a url. The url to hit, the http method to invoke,
@@ -85,8 +91,9 @@ public void execute(JobExecutionContext context) throws JobExecutionException {
       throw new JobExecutionException(
           "Interrupted while waiting for server to initialize.", e, false);
     }
-    Trigger trigger = context.getTrigger();
-    UrlFetchJobDetail jd = (UrlFetchJobDetail) context.getJobDetail();
+
+    UrlFetchJobDetail jd = new UrlFetchJobDetail(context.getJobDetail().getJobDataMap());
+
     URLFetchRequest fetchReq =
         newFetchRequest(
             jd.getTaskName(),
@@ -104,10 +111,11 @@ public void execute(JobExecutionContext context) throws JobExecutionException {
     if ((status < 200 || status > 299) && canRetry(jd, firstTryMs)) {
       logger.atInfo().log(
           "Web hook at %s returned status code %d.  Rescheduling...", fetchReq.getUrl(), status);
-      reschedule(context.getScheduler(), trigger, jd, firstTryMs, status);
+      reschedule(context.getScheduler(), context.getTrigger(), jd, firstTryMs, status);
     } else {
       try {
-        context.getScheduler().unscheduleJob(trigger.getName(), trigger.getGroup());
+        Trigger trigger = context.getTrigger();
+        context.getScheduler().unscheduleJob(trigger.getKey());
       } catch (SchedulerException e) {
         logger.atSevere().withCause(e).log("Unsubscription of task %s failed.", jd.getAddRequest());
       }
@@ -141,16 +149,29 @@ private void reschedule(
       long firstTryMs,
       int previousResponse) {
     // Builds a new job.
-    UrlFetchJobDetail newJobDetail = jd.retry(firstTryMs, previousResponse);
+    UrlFetchJobDetail newJobData = jd.retry(firstTryMs, previousResponse);
+    JobDataMap newJobDataMap = newJobData.getJobDataMap();
 
     // Build the new trigger from the old trigger
-    SimpleTrigger newTrigger = new SimpleTrigger(trigger.getName(), trigger.getGroup());
-    newTrigger.setStartTime(new Date(clock.getCurrentTime() + newJobDetail.getRetryDelayMs()));
+    TriggerKey triggerKey = trigger.getKey();
+    Instant newStartTime =
+        Instant.ofEpochMilli(clock.getCurrentTime() + newJobData.getRetryDelayMs());
+
+    SimpleTrigger newTrigger =
+        (SimpleTrigger)
+            newTrigger().withIdentity(triggerKey).startAt(Date.from(newStartTime)).build();
+
+    JobDetail newJob =
+        newJob(UrlFetchJob.class)
+            .withIdentity(jd.getTaskName(), jd.getQueueName())
+            .usingJobData(newJobDataMap)
+            .build();
+
     try {
-      // Quartz doesn't allow 2 jobs with the same name so we need to first
+      // Quartz 2.x doesn't allow 2 jobs with the same name so we need to first
       // unschedule the currently executing job before we reschedule
-      scheduler.unscheduleJob(trigger.getName(), trigger.getGroup());
-      scheduler.scheduleJob(newJobDetail, newTrigger);
+      scheduler.unscheduleJob(triggerKey);
+      scheduler.scheduleJob(newJob, newTrigger);
     } catch (SchedulerException e) {
       logger.atSevere().withCause(e).log("Reschedule of task %s failed.", jd.getAddRequest());
     }