AbstractJobProducer implements JobProducer { @Override public boolean isReadyToAccept(Job job) throws ServiceRegistryException, UndispatchableJobException { if (!jobType.equals(job.getJobType())) { logger.debug("Invalid job type submitted: {}", job.getJobType()); return false; } NodeLoad maxload; try { maxload = getServiceRegistry().getMaxLoadOnNode(getServiceRegistry().getRegistryHostname()); } catch (NotFoundException e) { throw new ServiceRegistryException(e); } float currentLoad = getServiceRegistry().getOwnLoad(); logger.debug("{} Current load on this host: {}, job's load: {}, job's status: {}, max load: {}", Thread.currentThread().getId(), currentLoad, job.getJobLoad(), job.getStatus().name(), maxload.getMaxLoad()); currentLoad += job.getJobLoad(); if (job.getJobLoad() > maxload.getMaxLoad() && acceptJobLoadsExeedingMaxLoad) { logger.warn( "{} Accepting job {} of type {} with load {} even though load of {} is above this node's limit of {}.", Thread.currentThread().getId(), job.getId(), job.getJobType(), df.format(job.getJobLoad()), df.format(currentLoad), df.format(maxload.getMaxLoad())); logger.warn("This is a configuration issue that you should resolve in a production system!"); return true; } else if (currentLoad > maxload.getMaxLoad()) { logger.debug( "{} Declining job {} of type {} with load {} because load of {} would exceed this node's limit of {}.", Thread.currentThread().getId(), job.getId(), job.getJobType(), df.format(job.getJobLoad()), df.format(currentLoad), df.format(maxload.getMaxLoad())); return false; } else { logger.debug("{} Accepting job {} of type {} with load {} because load of {} is within this node's limit of {}.", Thread.currentThread().getId(), job.getId(), job.getJobType(), df.format(job.getJobLoad()), df.format(currentLoad), df.format(maxload.getMaxLoad())); return true; } } AbstractJobProducer(String jobType); void activate(ComponentContext cc); @Override String getJobType(); @Override long countJobs(Status status); @Override void acceptJob(final Job job); @Override boolean isReadyToAcceptJobs(String operation); @Override boolean isReadyToAccept(Job job); Incidents incident(); static final boolean DEFAULT_ACCEPT_JOB_LOADS_EXCEEDING; static final String ACCEPT_JOB_LOADS_EXCEEDING_PROPERTY; }
@Test public void testIsReadyToAccept() throws Exception { expect(serviceRegistry.getRegistryHostname()).andReturn("test").anyTimes(); expect(serviceRegistry.getMaxLoadOnNode("test")).andReturn(new NodeLoad("test", 0.0f, 4.0f)).anyTimes(); expect(serviceRegistry.getOwnLoad()).andReturn(1.0f); expect(serviceRegistry.getOwnLoad()).andReturn(5.0f); expect(serviceRegistry.getOwnLoad()).andReturn(10.0f); replayAll(); jobProducer.setAcceptOversizeJobs(false); Job job = new JobImpl(3); job.setJobType("test"); job.setStatus(Status.DISPATCHING); job.setProcessingHost("same"); job.setJobLoad(1.0f); assertTrue("Job load + current load < max load, so accept job", jobProducer.isReadyToAccept(job)); job.setJobLoad(1.0f); assertFalse("Job load + current load >= max load, so reject job", jobProducer.isReadyToAccept(job)); job.setJobLoad(10.0f); assertFalse("Job load >= max load, so reject job", jobProducer.isReadyToAccept(job)); }
@Test public void testIsReadyToAcceptOversize() throws Exception { expect(serviceRegistry.getRegistryHostname()).andReturn("test").anyTimes(); expect(serviceRegistry.getMaxLoadOnNode("test")).andReturn(new NodeLoad("test", 0.0f, 4.0f)).anyTimes(); expect(serviceRegistry.getOwnLoad()).andReturn(1.0f); expect(serviceRegistry.getOwnLoad()).andReturn(5.0f); expect(serviceRegistry.getOwnLoad()).andReturn(10.0f); replayAll(); jobProducer.setAcceptOversizeJobs(true); Job job = new JobImpl(3); job.setJobType("test"); job.setStatus(Status.DISPATCHING); job.setProcessingHost("same"); job.setJobLoad(1.0f); assertTrue("Job load + current load < max load, so accept job", jobProducer.isReadyToAccept(job)); job.setJobLoad(1.0f); assertFalse("Job load + current load >= max load, so accept job", jobProducer.isReadyToAccept(job)); job.setJobLoad(10.0f); assertTrue("Job load >= max load, but accepting oversize jobs is true so accept job", jobProducer.isReadyToAccept(job)); }