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

PiperOrigin-RevId: 899225510
Change-Id: I7be55cf6b2b38e0913b9e1470b24545e86909502

Author: gae-java-bot

api_dev/pom.xml

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

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

@@ -16,12 +16,6 @@
 
 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;
@@ -34,24 +28,19 @@
 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;
 
 /**
@@ -100,7 +89,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.pauseTriggers(triggerGroupEquals(getQueueName()));
+          scheduler.pauseTriggerGroup(getQueueName());
         } catch (SchedulerException e) {
           throw new ApiProxy.ApplicationException(ErrorCode.INTERNAL_ERROR_VALUE, e.getMessage());
         }
@@ -123,7 +112,7 @@ private synchronized String scheduleTask(TaskQueueAddRequest.Builder addRequest)
       taskName = genTaskName();
     }
     try {
-      if (scheduler.checkExists(jobKey(taskName, getQueueName()))) {
+      if (scheduler.getJobDetail(taskName, getQueueName()) != null) {
         throw new ApiProxy.ApplicationException(ErrorCode.TASK_ALREADY_EXISTS_VALUE);
       }
     } catch (SchedulerException e) {
@@ -132,32 +121,19 @@ private synchronized String scheduleTask(TaskQueueAddRequest.Builder addRequest)
 
     TaskQueueRetryParameters retryParams = getRetryParameters(addRequest);
     long etaMillis = addRequest.getEtaUsec() / 1000L;
-
-    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();
-
+    SimpleTrigger trigger = new SimpleTrigger(taskName, getQueueName());
+    trigger.setStartTime(new Date(etaMillis));
+    JobDetail jd = newUrlFetchJobDetail(taskName, getQueueName(), addRequest, retryParams);
     try {
-      scheduler.scheduleJob(job, trigger);
+      scheduler.scheduleJob(jd, trigger);
     } catch (SchedulerException e) {
       throw new ApiProxy.ApplicationException(ErrorCode.INTERNAL_ERROR_VALUE, e.getMessage());
     }
     return taskName;
   }
 
   // broken out to support testing
-  JobDataMap newUrlFetchJobDataMap(
+  JobDetail newUrlFetchJobDetail(
       String taskName,
       String queueName,
       TaskQueueAddRequest.Builder addRequest,
@@ -167,20 +143,18 @@ JobDataMap newUrlFetchJobDataMap(
         String host = header.getValue().toStringUtf8();
         if (host.startsWith("localhost:")) {
           return new UrlFetchJobDetail(
-                  taskName,
-                  queueName,
-                  addRequest,
-                  "http://" + host,
-                  callback,
-                  queueXmlEntry,
-                  retryParams)
-              .getJobDataMap();
+              taskName,
+              queueName,
+              addRequest,
+              "http://" + host,
+              callback,
+              queueXmlEntry,
+              retryParams);
         }
       }
     }
     return new UrlFetchJobDetail(
-            taskName, queueName, addRequest, baseUrl, callback, queueXmlEntry, retryParams)
-        .getJobDataMap();
+        taskName, queueName, addRequest, baseUrl, callback, queueXmlEntry, retryParams);
   }
 
   @Override
@@ -203,11 +177,8 @@ TaskQueueAddResponse add(TaskQueueAddRequest.Builder addRequest) {
   }
 
   List<String> getSortedJobNames() throws SchedulerException {
-    Set<JobKey> jobKeys = scheduler.getJobKeys(jobGroupEquals(getQueueName()));
-    List<String> jobNameList = new ArrayList<>();
-    for (JobKey jobKey : jobKeys) {
-      jobNameList.add(jobKey.getName());
-    }
+    String[] jobNames = scheduler.getJobNames(getQueueName());
+    List<String> jobNameList = Arrays.asList(jobNames);
     Collections.sort(jobNameList);
     return jobNameList;
   }
@@ -220,27 +191,23 @@ QueueStateInfo getStateInfo() {
       // Get the names of all jobs belonging to this queue (group).
       for (String jobName : getSortedJobNames()) {
         // Now get job details
-        JobDetail jobDetail = scheduler.getJobDetail(jobKey(jobName, getQueueName()));
-        if (jobDetail == null) {
+        UrlFetchJobDetail jd = (UrlFetchJobDetail) scheduler.getJobDetail(jobName, getQueueName());
+        if (jd == null) {
           // oops, gone, must have already run
           continue;
         }
-
-        UrlFetchJobDetail jd = new UrlFetchJobDetail(jobDetail.getJobDataMap());
-
-        List<? extends Trigger> triggers =
-            scheduler.getTriggersOfJob(jobKey(jobName, getQueueName()));
-        if (triggers.isEmpty()) {
+        Trigger[] triggers = scheduler.getTriggersOfJob(jobName, getQueueName());
+        if (triggers.length == 0) {
           // must have run in between the time we fetched the job detail and the time we fetched the
           // trigger
           continue;
         }
-        if (triggers.size() != 1) {
+        if (triggers.length != 1) {
           throw new IllegalStateException(
               "Multiple triggers for task " + jobName + " in queue " + getQueueName());
         }
-        long execTime = triggers.get(0).getStartTime().toInstant().toEpochMilli();
-        taskInfoList.add(new TaskStateInfo(jd.getTaskName(), execTime, jd.getAddRequest(), clock));
+        long execTime = triggers[0].getStartTime().getTime();
+        taskInfoList.add(new TaskStateInfo(jd.getName(), execTime, jd.getAddRequest(), clock));
       }
     } catch (SchedulerException e) {
       throw new ApiProxy.ApplicationException(ErrorCode.INTERNAL_ERROR_VALUE);
@@ -264,7 +231,7 @@ QueueStateInfo getStateInfo() {
   @Override
   boolean deleteTask(String taskName) {
     try {
-      return scheduler.deleteJob(jobKey(taskName, getQueueName()));
+      return scheduler.deleteJob(taskName, getQueueName());
     } catch (SchedulerException e) {
       throw new ApiProxy.ApplicationException(ErrorCode.INTERNAL_ERROR_VALUE);
     }
@@ -274,18 +241,20 @@ boolean deleteTask(String taskName) {
   @Override
   void flush() {
     try {
-      Set<JobKey> jobKeys = scheduler.getJobKeys(jobGroupEquals(getQueueName()));
-      scheduler.deleteJobs(new ArrayList<>(jobKeys));
+      for (String name : scheduler.getJobNames(getQueueName())) {
+        scheduler.deleteJob(name, getQueueName());
+      }
     } catch (SchedulerException e) {
       throw new ApiProxy.ApplicationException(ErrorCode.INTERNAL_ERROR_VALUE);
     }
   }
 
-  private JobExecutionContext getExecutionContext(JobDetail jobDetail, SimpleTrigger trigger) {
+  private JobExecutionContext getExecutionContext(UrlFetchJobDetail jobDetail) {
+    Trigger trigger = new SimpleTrigger(jobDetail.getTaskName(), jobDetail.getQueueName());
+    trigger.setJobDataMap(jobDetail.getJobDataMap());
     TriggerFiredBundle bundle =
-        new TriggerFiredBundle(
-            jobDetail, (OperableTrigger) trigger, null, false, null, null, null, null);
-    return new JobExecutionContextImpl(scheduler, bundle, null);
+        new TriggerFiredBundle(jobDetail, trigger, null, false, null, null, null, null);
+    return new JobExecutionContext(scheduler, bundle, null);
   }
 
   /**
@@ -300,19 +269,11 @@ boolean runTask(String taskName) {
     Job job;
     JobExecutionContext context;
     try {
-      JobDetail jd = scheduler.getJobDetail(jobKey(taskName, getQueueName()));
+      UrlFetchJobDetail jd = (UrlFetchJobDetail) scheduler.getJobDetail(taskName, getQueueName());
       if (jd == null) {
         return false;
       }
-      // 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);
+      context = getExecutionContext(jd);
       job = (Job) jd.getJobClass().newInstance();
     } catch (SchedulerException e) {
       return false;

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

@@ -63,7 +63,6 @@
 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;
@@ -691,13 +690,7 @@ static Scheduler startScheduler(boolean disableAutoTaskExecution) {
     // TODO: Investigate config options for the scheduler like
     // threadpool size.
     try {
-      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();
+      Scheduler scheduler = StdSchedulerFactory.getDefaultScheduler();
       // 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,29 +16,23 @@
 
 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,
@@ -91,9 +85,8 @@ public void execute(JobExecutionContext context) throws JobExecutionException {
       throw new JobExecutionException(
           "Interrupted while waiting for server to initialize.", e, false);
     }
-
-    UrlFetchJobDetail jd = new UrlFetchJobDetail(context.getJobDetail().getJobDataMap());
-
+    Trigger trigger = context.getTrigger();
+    UrlFetchJobDetail jd = (UrlFetchJobDetail) context.getJobDetail();
     URLFetchRequest fetchReq =
         newFetchRequest(
             jd.getTaskName(),
@@ -111,11 +104,10 @@ 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(), context.getTrigger(), jd, firstTryMs, status);
+      reschedule(context.getScheduler(), trigger, jd, firstTryMs, status);
     } else {
       try {
-        Trigger trigger = context.getTrigger();
-        context.getScheduler().unscheduleJob(trigger.getKey());
+        context.getScheduler().unscheduleJob(trigger.getName(), trigger.getGroup());
       } catch (SchedulerException e) {
         logger.atSevere().withCause(e).log("Unsubscription of task %s failed.", jd.getAddRequest());
       }
@@ -149,29 +141,16 @@ private void reschedule(
       long firstTryMs,
       int previousResponse) {
     // Builds a new job.
-    UrlFetchJobDetail newJobData = jd.retry(firstTryMs, previousResponse);
-    JobDataMap newJobDataMap = newJobData.getJobDataMap();
+    UrlFetchJobDetail newJobDetail = jd.retry(firstTryMs, previousResponse);
 
     // Build the new trigger from the old trigger
-    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();
-
+    SimpleTrigger newTrigger = new SimpleTrigger(trigger.getName(), trigger.getGroup());
+    newTrigger.setStartTime(new Date(clock.getCurrentTime() + newJobDetail.getRetryDelayMs()));
     try {
-      // Quartz 2.x doesn't allow 2 jobs with the same name so we need to first
+      // Quartz doesn't allow 2 jobs with the same name so we need to first
       // unschedule the currently executing job before we reschedule
-      scheduler.unscheduleJob(triggerKey);
-      scheduler.scheduleJob(newJob, newTrigger);
+      scheduler.unscheduleJob(trigger.getName(), trigger.getGroup());
+      scheduler.scheduleJob(newJobDetail, newTrigger);
     } catch (SchedulerException e) {
       logger.atSevere().withCause(e).log("Reschedule of task %s failed.", jd.getAddRequest());
     }