srm: fix restore so jobs for other schedulers are not restored

The dCache may have multiple SRM instances and jobs are associated
with a specific scheduler.

Currently on startup, jobs may be restored that are associated with
other SRM instances.  This may lead to the job being processed
multiple times, leading to inconsistent results.

This patch alters the behaviour so only jobs for the named scheduler
will be started.  It also adds a check that jobs that have been
assigned a scheduler are submitted only to that scheduler.

Target: master
Patch: http://rb.dcache.org/r/6826/
Acked-by: Gerd Behrmann

modules/srm-server/src/main/java/org/dcache/srm/scheduler/Scheduler.java

@@ -230,7 +230,7 @@ public void stop()
 
 
     public void schedule(Job job)
-            throws IllegalStateException,
+            throws IllegalStateException, IllegalArgumentException,
                    IllegalStateTransition
     {
         checkState(running, "scheduler is not running");
@@ -254,6 +254,8 @@ public void schedule(Job job)
             case ASYNCWAIT:
             case RETRYWAIT:
             case RUNNINGWITHOUTTHREAD:
+                checkArgument(id.equals(job.getSchedulerId()),
+                        "job assigned to wrong scheduler");
                 LOGGER.trace("putting job in a priority thread queue, job#{}", job.getId());
                 job.setState(State.PRIORITYTQUEUED, "queued for execution");
                 if (!priorityQueue(job)) {

modules/srm-server/src/main/java/org/dcache/srm/scheduler/SchedulerContainer.java

@@ -66,6 +66,7 @@ private Scheduler getScheduler(Class<? extends Job> type)
     }
 
     private Scheduler getScheduler(Scheduler suggestion, Class<? extends Job> type)
+            throws UnsupportedOperationException
     {
         if (suggestion == null || !suggestion.getType().isAssignableFrom(type)) {
             for (Entry<Class<? extends Job>, Scheduler> entry : schedulers.entrySet()) {
@@ -112,7 +113,9 @@ public void restoreJobsOnSrmStart(Iterable<? extends Job> activeJobs)
 
         for (Job job : activeJobs) {
             scheduler = getScheduler(scheduler, job.getSchedulerType());
-            job.onSrmRestart(scheduler);
+            if (scheduler.getId().equals(job.getSchedulerId())) {
+                job.onSrmRestart(scheduler);
+            } // else another SRM instance is handling this job
         }
     }
 }