@SuppressWarnings("deprecation") private static JobStats generateFakeJobStats(final int numMaps, final int numReds, final boolean isSuccessful, final boolean lost) throws IOException { // A fake job Job fakeJob = new Job() { @Override public int getNumReduceTasks() { return numReds; }; @Override public boolean isSuccessful() throws IOException { if (lost) { throw new IOException("Test failure!"); } return isSuccessful; }; }; return new JobStats(numMaps, numReds, fakeJob); }
/** * Test {@link Statistics.JobStats}. */ @Test @SuppressWarnings("deprecation") public void testJobStats() throws Exception { Job job = new Job() {}; JobStats stats = new JobStats(1, 2, job); assertEquals("Incorrect num-maps", 1, stats.getNoOfMaps()); assertEquals("Incorrect num-reds", 2, stats.getNoOfReds()); assertTrue("Incorrect job", job == stats.getJob()); assertNull("Unexpected job status", stats.getJobStatus()); // add a new status JobStatus status = new JobStatus(); stats.updateJobStatus(status); assertNotNull("Missing job status", stats.getJobStatus()); assertTrue("Incorrect job status", status == stats.getJobStatus()); }
private static void testJobStats(JobStats stats, int numMaps, int numReds, JobStatus jStatus, Job job) { assertEquals("Incorrect num map tasks", numMaps, stats.getNoOfMaps()); assertEquals("Incorrect num reduce tasks", numReds, stats.getNoOfReds()); if (job != null) { assertNotNull("Missing job", job); } // check running job assertTrue("Incorrect job", job == stats.getJob()); if (jStatus != null) { assertNotNull("Missing job status", jStatus); } // check job stats assertTrue("Incorrect job status", jStatus == stats.getJobStatus()); }
/** * Create a JobMonitor that sleeps for the specified duration after * polling a still-running job. * @param pollDelay Delay after polling a running job * @param unit Time unit for pollDelaySec (rounded to milliseconds) * @param statistics StatCollector , listener to job completion. */ public JobMonitor(int pollDelay, TimeUnit unit, Statistics statistics, int numPollingThreads) { executor = Executors.newCachedThreadPool(); this.numPollingThreads = numPollingThreads; runningJobs = new LinkedBlockingQueue<JobStats>(); mJobs = new LinkedList<JobStats>(); this.pollDelayMillis = TimeUnit.MILLISECONDS.convert(pollDelay, unit); this.statistics = statistics; }
/** * Add a submission failed job's status, such that it can be communicated * back to serial. * TODO: Cleaner solution for this problem * @param job */ public void submissionFailed(JobStats job) { String jobID = job.getJob().getConfiguration().get(Gridmix.ORIGINAL_JOB_ID); LOG.info("Job submission failed notification for job " + jobID); synchronized (statistics) { this.statistics.add(job); } }
private void processJobState(JobStats stats) { Job job = stats.getJob(); try { if (job.isSuccessful()) { ++totalSuccessfulJobs; } else { ++totalFailedJobs; } } catch (Exception e) { // this behavior is consistent with job-monitor which marks the job as // complete (lost) if the status polling bails out ++totalLostJobs; } }
private void process(JobStats stats) { // process the job run state processJobState(stats); // process the tasks information processJobTasks(stats); }
@Override public void update(JobStats item) { // process only if the simulation has started if (simulationStartTime > 0) { process(item); totalSimulationTime = System.currentTimeMillis() - getSimulationStartTime(); } }
/** * SERIAL. Once you get notification from StatsCollector about the job * completion ,simply notify the waiting thread. * * @param item */ @Override public void update(Statistics.JobStats item) { //simply notify in case of serial submissions. We are just bothered //if submitted job is completed or not. lock.lock(); try { jobCompleted.signalAll(); } finally { lock.unlock(); } }
