ServiceRegistryJpaImpl implements ServiceRegistry, ManagedService { @Override public Job createJob(String type, String operation) throws ServiceRegistryException { return createJob(this.hostName, type, operation, null, null, true, getCurrentJob(), DEFAULT_JOB_LOAD); }  @Activate void activate(ComponentContext cc); @Override float getOwnLoad(); @Override String getRegistryHostname(); @Deactivate void deactivate(); @Override Job createJob(String type, String operation); @Override Job createJob(String type, String operation, List<String> arguments); @Override Job createJob(String type, String operation, List<String> arguments, Float jobLoad); @Override Job createJob(String type, String operation, List<String> arguments, String payload, boolean dispatchable); @Override Job createJob(String type, String operation, List<String> arguments, String payload, boolean dispatchable,
          Float jobLoad); Job createJob(String type, String operation, List<String> arguments, String payload, boolean dispatchable,
          Job parentJob); @Override Job createJob(String type, String operation, List<String> arguments, String payload, boolean dispatchable,
          Job parentJob, Float jobLoad); Job createJob(String host, String serviceType, String operation, List<String> arguments, String payload,
          boolean dispatchable, Job parentJob); Job createJob(String host, String serviceType, String operation, List<String> arguments, String payload,
          boolean dispatchable, Job parentJob, float jobLoad); @Override void removeJobs(List<Long> jobIds); @Override void removeParentlessJobs(int lifetime); @Override @SuppressWarnings("rawtypes") void updated(Dictionary properties); @Modified void modified(Map<String, Object> config); @Override Job getJob(long id); @Override Job getCurrentJob(); @Override void setCurrentJob(Job job); @Override Job updateJob(Job job); @Override void registerHost(String host, String address, String nodeName, long memory, int cores, float maxLoad); @Override void unregisterHost(String host); @Override void enableHost(String host); @Override void disableHost(String host); @Override ServiceRegistration registerService(String serviceType, String baseUrl, String path); @Override ServiceRegistration registerService(String serviceType, String baseUrl, String path, boolean jobProducer); @Override void unRegisterService(String serviceType, String baseUrl); @Override void setMaintenanceStatus(String baseUrl, boolean maintenance); @Override List<ServiceRegistration> getServiceRegistrations(); @Override Incidents incident(); @Override List<HostRegistration> getHostRegistrations(); @Override HostRegistration getHostRegistration(String hostname); @Override List<Job> getChildJobs(long id); @Override List<Job> getJobs(String type, Status status); @Override List<String> getJobPayloads(String operation); @Override List<String> getJobPayloads(String operation, int limit, int offset); @Override int getJobCount(final String operation); @Override List<Job> getActiveJobs(); List<JpaJob> getJobsByStatus(EntityManager em, Status... statuses); @Override long count(String serviceType, Status status); @Override long countByHost(String serviceType, String host, Status status); @Override long countByOperation(String serviceType, String operation, Status status); @Override long count(String serviceType, String host, String operation, Status status); @Override List<ServiceStatistics> getServiceStatistics(); @Override List<ServiceRegistration> getServiceRegistrationsByLoad(String serviceType); @Override SystemLoad getCurrentHostLoads(); @SuppressWarnings("unchecked") @Override List<ServiceRegistration> getServiceRegistrationsByType(String serviceType); @SuppressWarnings("unchecked") @Override List<ServiceRegistration> getServiceRegistrationsByHost(String host); @Override ServiceRegistration getServiceRegistration(String serviceType, String host); @Reference(name = "security-service") void setSecurityService(SecurityService securityService); @Reference(name = "user-directory") void setUserDirectoryService(UserDirectoryService userDirectoryService); @Reference(name = "orgDirectory") void setOrganizationDirectoryService(OrganizationDirectoryService organizationDirectory); @Reference(name = "incidentService", cardinality = ReferenceCardinality.OPTIONAL, policy = ReferencePolicy.DYNAMIC, unbind = "unsetIncidentService") void setIncidentService(IncidentService incidentService); void unsetIncidentService(IncidentService incidentService); @Override void sanitize(String serviceType, String host); @Override SystemLoad getMaxLoads(); @Override NodeLoad getMaxLoadOnNode(String host); static final String PERSISTENCE_UNIT; static final String START_OPERATION; static final String START_WORKFLOW; static final String RESUME; static final String TYPE_WORKFLOW; }
@Test public void testHostAddedToPriorityList() throws Exception { Job testJob = serviceRegistryJpaImpl.createJob(TEST_HOST, TEST_SERVICE, TEST_OPERATION, null, null, true, null); JobBarrier barrier = new JobBarrier(null, serviceRegistryJpaImpl, testJob); launchDispatcherOnce(false); try { barrier.waitForJobs(JOB_BARRIER_TIMEOUT); Assert.fail("Did not receive a timeout exception"); } catch (Exception e) { Assert.assertEquals(1, serviceRegistryJpaImpl.dispatchPriorityList.size()); } }
@Test public void testHostsBeingRemovedFromPriorityList() throws Exception { serviceRegistryJpaImpl.dispatchPriorityList.put(0L, TEST_HOST); Job testJob = serviceRegistryJpaImpl.createJob(TEST_HOST, TEST_SERVICE_2, TEST_OPERATION, null, null, true, null); JobBarrier barrier = new JobBarrier(null, serviceRegistryJpaImpl, testJob); launchDispatcherOnce(false); try { barrier.waitForJobs(JOB_BARRIER_TIMEOUT); Assert.fail("Did not receive a timeout exception"); } catch (Exception e) { Assert.assertEquals(0, serviceRegistryJpaImpl.dispatchPriorityList.size()); } finally { logger.debug("end testHostsBeingRemovedFromPriorityList"); } }
@Test public void testJobDispatchingFairness() throws Exception { Job j = serviceRegistryJpaImpl.createJob(TEST_HOST, TEST_SERVICE_FAIRNESS, TEST_OPERATION, null, null, true, null, 1.0f); assertHostloads(j, 0.0f, 0.0f, 1.0f); j = serviceRegistryJpaImpl.createJob(TEST_HOST, TEST_SERVICE_FAIRNESS, TEST_OPERATION, null, null, true, null, 1.0f); assertHostloads(j, 0.0f, 1.0f, 1.0f); j = serviceRegistryJpaImpl.createJob(TEST_HOST, TEST_SERVICE_FAIRNESS, TEST_OPERATION, null, null, true, null, 1.0f); assertHostloads(j, 1.0f, 1.0f, 1.0f); j = serviceRegistryJpaImpl.createJob(TEST_HOST, TEST_SERVICE_FAIRNESS, TEST_OPERATION, null, null, true, null, 1.0f); assertHostloads(j,1.0f, 1.0f, 2.0f); j = serviceRegistryJpaImpl.createJob(TEST_HOST, TEST_SERVICE_FAIRNESS, TEST_OPERATION, null, null, true, null, 1.0f); assertHostloads(j, 1.0f, 1.0f, 3.0f); j = serviceRegistryJpaImpl.createJob(TEST_HOST, TEST_SERVICE_FAIRNESS, TEST_OPERATION, null, null, true, null, 1.0f); assertHostloads(j, 1.0f, 2.0f, 3.0f); j = serviceRegistryJpaImpl.createJob(TEST_HOST, TEST_SERVICE_FAIRNESS, TEST_OPERATION, null, null, true, null, 1.0f); assertHostloads(j,1.0f, 2.0f, 4.0f); }