MemoryLeak: do not keep finished jobs state in memory

 - SchedulerStateImpl : use a maximum size for finished jobs. As test is using the getState method a lot, it is necessary to keep some retro-compatibility. The default number of jobs kept in SchedulerStateImpl is 1000, but can be configured through a JVM property.
 - SchedulerFrontendState : delete job from jobsMap when the job is finished.

scheduler/scheduler-api/src/main/java/org/ow2/proactive/scheduler/common/Scheduler.java

@@ -1147,6 +1147,9 @@ public interface Scheduler extends SchedulerUsage, ThirdPartyCredentials {
      * If a PermissionException is thrown, try using {@link #getState(boolean)}
      * method with argument {@code true}.
      *
+     * NOTE: the total list of finished jobs is limited by the Java property 'scheduler.state.max.finished.jobs'
+     * configured on the server. Default is 1000 finished jobs.
+     *
      * @return the list of every jobs in the Scheduler
      * @throws NotConnectedException
      *             if you are not authenticated.
@@ -1161,6 +1164,10 @@ public interface Scheduler extends SchedulerUsage, ThirdPartyCredentials {
      * the given argument is true, only job that you own will be returned,
      * otherwise every jobs will be returned depending on your right.
      *
+     * NOTE: the total list of finished jobs is limited by the Java property 'scheduler.state.max.finished.jobs'
+     * configured on the server. Default is 1000 finished jobs.
+     * From this list, only jobs belonging to the current user will be returned.
+     *
      * @param myJobsOnly
      *            true to get only my jobs, false to get any.
      * @return the list of every jobs in the Scheduler

scheduler/scheduler-client/src/main/java/org/ow2/proactive/scheduler/core/SchedulerStateImpl.java

@@ -64,6 +64,14 @@ public final class SchedulerStateImpl<T extends JobState> implements SchedulerSt
     /** Finished jobs */
     private Map<JobId, T> finishedJobs = Collections.synchronizedMap(new LinkedHashMap<JobId, T>());
 
+    /**
+     * Used to control the maximum size of finished jobs. This is mostly used for test compatibility,
+     * As tests use a lot the getState() request. Normal usage of the scheduler does not rely on getState()
+     */
+    private static Integer maxFinishedJobsSize;
+
+    public final static int DEFAULT_MAX_FINISHED_JOBS_SIZE = 1000;
+
     /** Scheduler status */
     private SchedulerStatus status = SchedulerStatus.STARTED;
 
@@ -89,6 +97,7 @@ public final class SchedulerStateImpl<T extends JobState> implements SchedulerSt
      * ProActive Empty constructor.
      */
     public SchedulerStateImpl() {
+        initMaxFinishedJobs();
     }
 
     /**
@@ -364,12 +373,22 @@ public final class SchedulerStateImpl<T extends JobState> implements SchedulerSt
 
     public synchronized void pendingToFinished(T js) {
         pendingJobs.remove(js);
+        putInFinished(js);
+    }
+
+    private void putInFinished(T js) {
+        if (finishedJobs.size() >= maxFinishedJobsSize) {
+            Iterator<Map.Entry<JobId, T>> iterator = finishedJobs.entrySet().iterator();
+            Map.Entry<JobId, T> entryToRemove = iterator.next();
+            iterator.remove();
+            jobs.remove(entryToRemove.getKey());
+        }
         finishedJobs.put(js.getId(), js);
     }
 
     public synchronized void runningToFinished(T js) {
         runningJobs.remove(js);
-        finishedJobs.put(js.getId(), js);
+        putInFinished(js);
     }
 
     public synchronized void removeFinished(T js) {
@@ -382,4 +401,18 @@ public final class SchedulerStateImpl<T extends JobState> implements SchedulerSt
         jobs.remove(jobId);
     }
 
+    private static void initMaxFinishedJobs() {
+        if (maxFinishedJobsSize == null) {
+            maxFinishedJobsSize = DEFAULT_MAX_FINISHED_JOBS_SIZE;
+            String configuredMaxJobs = System.getProperty("scheduler.state.max.finished.jobs");
+            if (configuredMaxJobs != null) {
+                try {
+                    maxFinishedJobsSize = Integer.parseInt(configuredMaxJobs);
+                } catch (NumberFormatException e) {
+                    logger.warn("Invalid value of scheduler.state.max.finished.jobs property : " + configuredMaxJobs);
+                }
+            }
+        }
+    }
+
 }

scheduler/scheduler-server/src/main/java/org/ow2/proactive/scheduler/core/SchedulerFrontendState.java

@@ -1299,12 +1299,14 @@ class SchedulerFrontendState implements SchedulerStateUpdate {
                             schedulerState.pendingToFinished(js);
                             // set this job finished, user can get its result
                             jobs.remove(notification.getData().getJobId()).setFinished(true);
+                            jobsMap.remove(notification.getData().getJobId());
                             withAttachment = true;
                             break;
                         case JOB_RUNNING_TO_FINISHED:
                             schedulerState.runningToFinished(js);
                             // set this job finished, user can get its result
                             jobs.remove(notification.getData().getJobId()).setFinished(true);
+                            jobsMap.remove(notification.getData().getJobId());
                             withAttachment = true;
                             break;
                         default: