The implementation here will first partition the job by pages, and then by dates.
@return
501856
@Override
  public List<? extends ProducerJob> partitionJobs() {
    UrlTrieNode root = _jobNode.getRight();
    if (isOperatorEquals() || root.getSize() == 1) {
      //Either at an Equals-Node or a Leaf-Node, both of which actually has actual size 1.
      return super.partitionJobs();
    } else {
      if (_groupSize <= 1) {
        throw new RuntimeException("This is impossible. When group size is 1, the operator must be equals");
      }

      UrlTrie trie = new UrlTrie(getPage(), root);
      int gs = Math.min(root.getSize(), _groupSize);

      UrlTriePrefixGrouper grouper = new UrlTriePrefixGrouper(trie, (int) Math.ceil(gs / 2.0));

      List<TrieBasedProducerJob> jobs = new ArrayList<>();
      while (grouper.hasNext()) {
        jobs.add(new TrieBasedProducerJob(_startDate, _endDate, grouper.next(), grouper.getGroupSize()));
      }
      return jobs;
    }
  }
535013
private Completable initFromBeginningToNow() {
    return initWithCallback(new Action1<PartitionAndSeqno>() {
      @Override
      public void call(PartitionAndSeqno partitionAndSeqno) {
        short partition = partitionAndSeqno.partition();
        long seqno = partitionAndSeqno.seqno();
        PartitionState partitionState = sessionState().get(partition);
        partitionState.setEndSeqno(seqno);
        sessionState().set(partition, partitionState);
      }
    });
  }
236727
private void createPartitions(int numPartitions) {
		this.partitions.clear();
		
		ListMemorySegmentSource memSource = new ListMemorySegmentSource(this.availableMemory);
		
		for (int i = 0; i < numPartitions; i++) {
			this.partitions.add(new InMemoryPartition<T>(this.buildSideSerializer, i, memSource, this.segmentSize, pageSizeInBits));
		}
		this.compactionMemory = new InMemoryPartition<T>(this.buildSideSerializer, -1, memSource, this.segmentSize, pageSizeInBits);
	}
99855
private void splitDays(ProjectCalendar calendar, LinkedList<TimephasedWork> list)
   {
      LinkedList<TimephasedWork> result = new LinkedList<TimephasedWork>();
      boolean remainderInserted = false;
      
      for (TimephasedWork assignment : list)
      {
         if (remainderInserted)
         {
            assignment.setStart(DateHelper.addDays(assignment.getStart(), 1));
            remainderInserted = false;
         }

         while (assignment != null)
         {
            Date startDay = DateHelper.getDayStartDate(assignment.getStart());
            Date finishDay = DateHelper.getDayStartDate(assignment.getFinish());

            // special case - when the finishday time is midnight, it's really the previous day...
            if (assignment.getFinish().getTime() == finishDay.getTime())
            {
               finishDay = DateHelper.addDays(finishDay, -1);
            }

            if (startDay.getTime() == finishDay.getTime())
            {
               Duration totalWork = assignment.getTotalAmount();
               Duration assignmentWork = getAssignmentWork(calendar, assignment);
               if ((totalWork.getDuration() - assignmentWork.getDuration()) > EQUALITY_DELTA)
               {
                  assignment.setTotalAmount(assignmentWork);
                  result.add(assignment);
                  Duration remainingWork = Duration.getInstance(totalWork.getDuration() - assignmentWork.getDuration(), TimeUnit.MINUTES);
                  
                  Date remainderStart = DateHelper.addDays(finishDay, 1);
                  Date remainderFinish = DateHelper.addDays(remainderStart, 1);
                  
                  TimephasedWork remainder = new TimephasedWork();
                  remainder.setStart(remainderStart);
                  remainder.setFinish(remainderFinish);
                  remainder.setTotalAmount(remainingWork);
                  result.add(remainder);

                  remainderInserted = true;
               }
               else
               {
                  result.add(assignment);
               }
               break;
            }

            TimephasedWork[] split = splitFirstDay(calendar, assignment);
            if (split[0] != null)
            {
               result.add(split[0]);
            }

            if (assignment.equals(split[1]))
            {
               break;
            }

            assignment = split[1];
         }
      }
     
      list.clear();
      list.addAll(result);
   }
622506
@Override
	public CompletableFuture<JobSubmissionResult> submitJob(@Nonnull JobGraph jobGraph) {
		// we have to enable queued scheduling because slot will be allocated lazily
		jobGraph.setAllowQueuedScheduling(true);

		CompletableFuture<java.nio.file.Path> jobGraphFileFuture = CompletableFuture.supplyAsync(() -> {
			try {
				final java.nio.file.Path jobGraphFile = Files.createTempFile("flink-jobgraph", ".bin");
				try (ObjectOutputStream objectOut = new ObjectOutputStream(Files.newOutputStream(jobGraphFile))) {
					objectOut.writeObject(jobGraph);
				}
				return jobGraphFile;
			} catch (IOException e) {
				throw new CompletionException(new FlinkException("Failed to serialize JobGraph.", e));
			}
		}, executorService);

		CompletableFuture<Tuple2<JobSubmitRequestBody, Collection<FileUpload>>> requestFuture = jobGraphFileFuture.thenApply(jobGraphFile -> {
			List<String> jarFileNames = new ArrayList<>(8);
			List<JobSubmitRequestBody.DistributedCacheFile> artifactFileNames = new ArrayList<>(8);
			Collection<FileUpload> filesToUpload = new ArrayList<>(8);

			filesToUpload.add(new FileUpload(jobGraphFile, RestConstants.CONTENT_TYPE_BINARY));

			for (Path jar : jobGraph.getUserJars()) {
				jarFileNames.add(jar.getName());
				filesToUpload.add(new FileUpload(Paths.get(jar.toUri()), RestConstants.CONTENT_TYPE_JAR));
			}

			for (Map.Entry<String, DistributedCache.DistributedCacheEntry> artifacts : jobGraph.getUserArtifacts().entrySet()) {
				artifactFileNames.add(new JobSubmitRequestBody.DistributedCacheFile(artifacts.getKey(), new Path(artifacts.getValue().filePath).getName()));
				filesToUpload.add(new FileUpload(Paths.get(artifacts.getValue().filePath), RestConstants.CONTENT_TYPE_BINARY));
			}

			final JobSubmitRequestBody requestBody = new JobSubmitRequestBody(
				jobGraphFile.getFileName().toString(),
				jarFileNames,
				artifactFileNames);

			return Tuple2.of(requestBody, Collections.unmodifiableCollection(filesToUpload));
		});

		final CompletableFuture<JobSubmitResponseBody> submissionFuture = requestFuture.thenCompose(
			requestAndFileUploads -> sendRetriableRequest(
				JobSubmitHeaders.getInstance(),
				EmptyMessageParameters.getInstance(),
				requestAndFileUploads.f0,
				requestAndFileUploads.f1,
				isConnectionProblemOrServiceUnavailable())
		);

		submissionFuture
			.thenCombine(jobGraphFileFuture, (ignored, jobGraphFile) -> jobGraphFile)
			.thenAccept(jobGraphFile -> {
			try {
				Files.delete(jobGraphFile);
			} catch (IOException e) {
				log.warn("Could not delete temporary file {}.", jobGraphFile, e);
			}
		});

		return submissionFuture
			.thenApply(
				(JobSubmitResponseBody jobSubmitResponseBody) -> new JobSubmissionResult(jobGraph.getJobID()))
			.exceptionally(
				(Throwable throwable) -> {
					throw new CompletionException(new JobSubmissionException(jobGraph.getJobID(), "Failed to submit JobGraph.", ExceptionUtils.stripCompletionException(throwable)));
				});
	}
1003726
public SequencalJobExecuteResults runAll() {
        SequencalJobExecuteResults results = new SequencalJobExecuteResults();
        for (Job job : jobs) {
            job.setJarByClass(SequencalJobChain.class);
            try {
                boolean isSuccessful = job.waitForCompletion(true);
                results.add(new SequencalJobExecuteResult(isSuccessful, job.getJobName()));
                if (!isSuccessful) {
                    break;
                }
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
        return results;
    }
384621
private Map<String, List<ResourceGrant>> scheduleTasks() {
    fullyScheduled = false;
    long nodeWait = configManager.getLocalityWait(type, LocalityLevel.NODE);
    long rackWait = configManager.getLocalityWait(type, LocalityLevel.RACK);
    int tasksToSchedule = configManager.getGrantsPerIteration();
    Map<String, List<ResourceGrant>> sessionIdToGranted =
        new HashMap<String, List<ResourceGrant>>();
    for (int i = 0; i < tasksToSchedule; i++) {
      ScheduledPair scheduled = scheduleOneTask(nodeWait, rackWait);
      if (scheduled == null) {
        // Cannot find matched request-node anymore. We are done.
        fullyScheduled = true;
        break;
      }
      List<ResourceGrant> granted =
        sessionIdToGranted.get(scheduled.sessionId.toString());
      if (granted == null) {
        granted = new LinkedList<ResourceGrant>();
        sessionIdToGranted.put(scheduled.sessionId.toString(), granted);
      }
      granted.add(scheduled.grant);
    }
    return sessionIdToGranted;
  }
615539
public static List<PartitionDefinition> assignPartitions(List<PartitionDefinition> partitions, int currentTask, int totalTasks) {
        int esPartitions = partitions.size();
        if (totalTasks >= esPartitions) {
            return (currentTask >= esPartitions ? Collections.<PartitionDefinition>emptyList() : Collections.singletonList(partitions.get(currentTask)));
        } else {
            int partitionsPerTask = esPartitions / totalTasks;
            int remainder = esPartitions % totalTasks;

            int partitionsPerCurrentTask = partitionsPerTask;

            // spread the reminder against the tasks
            if (currentTask < remainder) {
                partitionsPerCurrentTask++;
            }

            // find the offset inside the collection
            int offset = partitionsPerTask * currentTask;
            if (currentTask != 0) {
                offset += (remainder > currentTask ? 1 : remainder);
            }

            // common case
            if (partitionsPerCurrentTask == 1) {
                return Collections.singletonList(partitions.get(offset));
            }

            List<PartitionDefinition> pa = new ArrayList<PartitionDefinition>(partitionsPerCurrentTask);
            for (int index = offset; index < offset + partitionsPerCurrentTask; index++) {
                pa.add(partitions.get(index));
            }
            return pa;
        }
    }
984268
public Iterator<AbstractJobOutputVertex> getOutputVertices() {

		final Collection<AbstractJobOutputVertex> coll = this.outputVertices.values();

		return coll.iterator();
	}
434058
public Iterator<AbstractJobInputVertex> getInputVertices() {

		final Collection<AbstractJobInputVertex> coll = this.inputVertices.values();

		return coll.iterator();
	}
1373825
private TimephasedCost[] splitFirstDay(ProjectCalendar calendar, TimephasedCost assignment)
   {
      TimephasedCost[] result = new TimephasedCost[2];

      //
      // Retrieve data used to calculate the pro-rata work split
      //
      Date assignmentStart = assignment.getStart();
      Date assignmentFinish = assignment.getFinish();
      Duration calendarWork = calendar.getWork(assignmentStart, assignmentFinish, TimeUnit.MINUTES);

      if (calendarWork.getDuration() != 0)
      {
         //
         // Split the first day
         //
         Date splitFinish;
         double splitCost;
         if (calendar.isWorkingDate(assignmentStart))
         {
            Date splitStart = assignmentStart;
            Date splitFinishTime = calendar.getFinishTime(splitStart);
            splitFinish = DateHelper.setTime(splitStart, splitFinishTime);
            Duration calendarSplitWork = calendar.getWork(splitStart, splitFinish, TimeUnit.MINUTES);
            splitCost = (assignment.getTotalAmount().doubleValue() * calendarSplitWork.getDuration()) / calendarWork.getDuration();

            TimephasedCost split = new TimephasedCost();
            split.setStart(splitStart);
            split.setFinish(splitFinish);
            split.setTotalAmount(Double.valueOf(splitCost));

            result[0] = split;
         }
         else
         {
            splitFinish = assignmentStart;
            splitCost = 0;
         }

         //
         // Split the remainder
         //
         Date splitStart = calendar.getNextWorkStart(splitFinish);
         splitFinish = assignmentFinish;
         TimephasedCost split;
         if (splitStart.getTime() > splitFinish.getTime())
         {
            split = null;
         }
         else
         {
            splitCost = assignment.getTotalAmount().doubleValue() - splitCost;

            split = new TimephasedCost();
            split.setStart(splitStart);
            split.setFinish(splitFinish);
            split.setTotalAmount(Double.valueOf(splitCost));
            split.setAmountPerDay(assignment.getAmountPerDay());
         }

         result[1] = split;
      }
      return result;
   }
1224903
private Collection<String> getPages(String startDate, String endDate, List<Dimension> dimensions,
      ApiDimensionFilter countryFilter, Queue<Pair<String, FilterOperator>> toProcess, int rowLimit) {
    String country = GoogleWebmasterFilter.countryFilterToString(countryFilter);

    ConcurrentLinkedDeque<String> allPages = new ConcurrentLinkedDeque<>();
    int r = 0;
    while (r <= GET_PAGES_RETRIES) {
      ++r;
      log.info(String.format("Get pages at round %d with size %d.", r, toProcess.size()));
      ConcurrentLinkedDeque<Pair<String, FilterOperator>> nextRound = new ConcurrentLinkedDeque<>();
      ExecutorService es = Executors.newFixedThreadPool(10,
          ExecutorsUtils.newDaemonThreadFactory(Optional.of(log), Optional.of(this.getClass().getSimpleName())));

      while (!toProcess.isEmpty()) {
        submitJob(toProcess.poll(), countryFilter, startDate, endDate, dimensions, es, allPages, nextRound, rowLimit);
      }
      //wait for jobs to finish and start next round if necessary.
      try {
        es.shutdown();
        boolean terminated = es.awaitTermination(5, TimeUnit.MINUTES);
        if (!terminated) {
          es.shutdownNow();
          log.warn("Timed out while getting all pages for country-{} at round {}. Next round now has size {}.", country,
              r, nextRound.size());
        }
      } catch (InterruptedException e) {
        throw new RuntimeException(e);
      }

      if (nextRound.isEmpty()) {
        break;
      }
      toProcess = nextRound;
      coolDown(r, PAGES_GET_COOLDOWN_TIME);
    }
    if (r == GET_PAGES_RETRIES + 1) {
      throw new RuntimeException(
          String.format("Getting all pages reaches the maximum number of retires %d. Date range: %s ~ %s. Country: %s.",
              GET_PAGES_RETRIES, startDate, endDate, country));
    }
    return allPages;
  }
373060
protected MemoryBlock allocatePage(long required) {
    MemoryBlock page = taskMemoryManager.allocatePage(Math.max(pageSize, required), this);
    if (page == null || page.size() < required) {
      throwOom(page, required);
    }
    used += page.size();
    return page;
  }
633248
public List getSortedEntries() {

        Map grouped = new HashMap();

        for (int i = 0; i < indexentry.size(); i++) {
            Entry e = (Entry) indexentry.get(i);
            String key = e.getKey();

            Entry master = (Entry) grouped.get(key);
            if (master != null) {
                master.addPageNumberAndTag(e.getPageNumber(), e.getTag());
            } else {
                e.addPageNumberAndTag(e.getPageNumber(), e.getTag());
                grouped.put(key, e);
            }
        }

        // copy to a list and sort it
        List sorted = new ArrayList(grouped.values());
        Collections.sort(sorted, comparator);
        return sorted;
    }
1045400
@Override
    public org.apache.hadoop.mapred.InputSplit[] getSplits(JobConf job, int numSplits) throws IOException {

        Settings settings = HadoopSettingsManager.loadFrom(job);
        Collection<PartitionDefinition> partitions = RestService.findPartitions(settings, log);
        EsInputSplit[] splits = new EsInputSplit[partitions.size()];

        int index = 0;
        for (PartitionDefinition part : partitions) {
            splits[index++] = new EsInputSplit(part);
        }
        log.info(String.format("Created [%d] splits", splits.length));
        return splits;
    }
732088
protected boolean jobTooOld(final JobInfo jobInfo, final JobDefinition jobDefinition) {
        final Optional<OffsetDateTime> stopped = jobInfo.getStopped();
        if (stopped.isPresent() && jobDefinition.maxAge().isPresent()) {
            final OffsetDateTime deadlineToRerun = stopped.get().plus(jobDefinition.maxAge().get());
            return deadlineToRerun.isBefore(now());
        }

        return false;
    }
137553
public JobDetail build() {

    JobDetailImpl job = new JobDetailImpl();

    job.setJobClass(jobClass);
    job.setDescription(description);
    if (key == null) {
      key = UUID.randomUUID().toString();
    }
    job.setName(key);
    job.setIsConcurrencyAllowed(isConcurrencyAllowed);

    if (!jobDataMap.isEmpty()) {
      job.setJobDataMap(jobDataMap);
    }

    return job;
  }