@Test @Transactional public void testFindJob() throws IOException { String sql = new String(Files.readAllBytes(Paths.get("src/test/resources/dbunit/job_picker_dataset.sql"))); template.update(sql); String sql1 = new String(Files.readAllBytes(Paths.get("src/test/resources/dbunit/object_lock_dataset.sql"))); template.update(sql1); assertEquals(2, jp.findJob("C03", "1").size()); assertEquals(1, template.queryForList("select * from METASTOR_OBJECT_LOCKS where CLUSTER_ID='C03'").size()); }
List<JobDefinition> findJob( String clusterID, String workerID ) { List<JobDefinition> jobs = new ArrayList<JobDefinition>(); try { deleteNotProcessingNotifications(); deleteExpiredLocks(); List<JobDefinition> result ; logger.info("Get Stats: " + analyzeStats); if ( analyzeStats ) { logger.info( "Running for stats" ); result = template.query( FIND_UNLOCKED_STATS_JOB_QUERY, new Object[]{maxRetry, jobRetryIntervalInSecs, clusterID}, new JobDefinition.ObjectDefinitionMapper() ); } else { result = template.query( FIND_UNLOCKED_JOB_QUERY, new Object[]{maxRetry, jobRetryIntervalInSecs, clusterID}, new JobDefinition.ObjectDefinitionMapper() ); } ObjectDefinition lockedJd = null; String threadId = workerID; for ( JobDefinition jd : result ) { if ( lockedJd == null ) { if ( lockObj( jd, clusterID, threadId ) ) { lockedJd = jd.getObjectDefinition(); jobs.add( jd ); } } else { if ( jd.objectDefinition.equals( lockedJd ) ) { jobs.add( jd ); } else { break; } } } } catch ( Exception ex ) { logger.severe( "Exception occurred when looking for jobs" ); ex.printStackTrace(); } return jobs; }
JobPicker { List<JobDefinition> findJob( String clusterID, String workerID ) { List<JobDefinition> jobs = new ArrayList<JobDefinition>(); try { deleteNotProcessingNotifications(); deleteExpiredLocks(); List<JobDefinition> result ; logger.info("Get Stats: " + analyzeStats); if ( analyzeStats ) { logger.info( "Running for stats" ); result = template.query( FIND_UNLOCKED_STATS_JOB_QUERY, new Object[]{maxRetry, jobRetryIntervalInSecs, clusterID}, new JobDefinition.ObjectDefinitionMapper() ); } else { result = template.query( FIND_UNLOCKED_JOB_QUERY, new Object[]{maxRetry, jobRetryIntervalInSecs, clusterID}, new JobDefinition.ObjectDefinitionMapper() ); } ObjectDefinition lockedJd = null; String threadId = workerID; for ( JobDefinition jd : result ) { if ( lockedJd == null ) { if ( lockObj( jd, clusterID, threadId ) ) { lockedJd = jd.getObjectDefinition(); jobs.add( jd ); } } else { if ( jd.objectDefinition.equals( lockedJd ) ) { jobs.add( jd ); } else { break; } } } } catch ( Exception ex ) { logger.severe( "Exception occurred when looking for jobs" ); ex.printStackTrace(); } return jobs; } }
JobPicker { List<JobDefinition> findJob( String clusterID, String workerID ) { List<JobDefinition> jobs = new ArrayList<JobDefinition>(); try { deleteNotProcessingNotifications(); deleteExpiredLocks(); List<JobDefinition> result ; logger.info("Get Stats: " + analyzeStats); if ( analyzeStats ) { logger.info( "Running for stats" ); result = template.query( FIND_UNLOCKED_STATS_JOB_QUERY, new Object[]{maxRetry, jobRetryIntervalInSecs, clusterID}, new JobDefinition.ObjectDefinitionMapper() ); } else { result = template.query( FIND_UNLOCKED_JOB_QUERY, new Object[]{maxRetry, jobRetryIntervalInSecs, clusterID}, new JobDefinition.ObjectDefinitionMapper() ); } ObjectDefinition lockedJd = null; String threadId = workerID; for ( JobDefinition jd : result ) { if ( lockedJd == null ) { if ( lockObj( jd, clusterID, threadId ) ) { lockedJd = jd.getObjectDefinition(); jobs.add( jd ); } } else { if ( jd.objectDefinition.equals( lockedJd ) ) { jobs.add( jd ); } else { break; } } } } catch ( Exception ex ) { logger.severe( "Exception occurred when looking for jobs" ); ex.printStackTrace(); } return jobs; }  }
JobPicker { List<JobDefinition> findJob( String clusterID, String workerID ) { List<JobDefinition> jobs = new ArrayList<JobDefinition>(); try { deleteNotProcessingNotifications(); deleteExpiredLocks(); List<JobDefinition> result ; logger.info("Get Stats: " + analyzeStats); if ( analyzeStats ) { logger.info( "Running for stats" ); result = template.query( FIND_UNLOCKED_STATS_JOB_QUERY, new Object[]{maxRetry, jobRetryIntervalInSecs, clusterID}, new JobDefinition.ObjectDefinitionMapper() ); } else { result = template.query( FIND_UNLOCKED_JOB_QUERY, new Object[]{maxRetry, jobRetryIntervalInSecs, clusterID}, new JobDefinition.ObjectDefinitionMapper() ); } ObjectDefinition lockedJd = null; String threadId = workerID; for ( JobDefinition jd : result ) { if ( lockedJd == null ) { if ( lockObj( jd, clusterID, threadId ) ) { lockedJd = jd.getObjectDefinition(); jobs.add( jd ); } } else { if ( jd.objectDefinition.equals( lockedJd ) ) { jobs.add( jd ); } else { break; } } } } catch ( Exception ex ) { logger.severe( "Exception occurred when looking for jobs" ); ex.printStackTrace(); } return jobs; }  boolean lockObj( JobDefinition jd, String clusterID, String threadID ); void unlockWorker( String clusterID, String threadID ); boolean extendLock( JobDefinition jd, String clusterID, String workerID ); }
JobPicker { List<JobDefinition> findJob( String clusterID, String workerID ) { List<JobDefinition> jobs = new ArrayList<JobDefinition>(); try { deleteNotProcessingNotifications(); deleteExpiredLocks(); List<JobDefinition> result ; logger.info("Get Stats: " + analyzeStats); if ( analyzeStats ) { logger.info( "Running for stats" ); result = template.query( FIND_UNLOCKED_STATS_JOB_QUERY, new Object[]{maxRetry, jobRetryIntervalInSecs, clusterID}, new JobDefinition.ObjectDefinitionMapper() ); } else { result = template.query( FIND_UNLOCKED_JOB_QUERY, new Object[]{maxRetry, jobRetryIntervalInSecs, clusterID}, new JobDefinition.ObjectDefinitionMapper() ); } ObjectDefinition lockedJd = null; String threadId = workerID; for ( JobDefinition jd : result ) { if ( lockedJd == null ) { if ( lockObj( jd, clusterID, threadId ) ) { lockedJd = jd.getObjectDefinition(); jobs.add( jd ); } } else { if ( jd.objectDefinition.equals( lockedJd ) ) { jobs.add( jd ); } else { break; } } } } catch ( Exception ex ) { logger.severe( "Exception occurred when looking for jobs" ); ex.printStackTrace(); } return jobs; }  boolean lockObj( JobDefinition jd, String clusterID, String threadID ); void unlockWorker( String clusterID, String threadID ); boolean extendLock( JobDefinition jd, String clusterID, String workerID ); static final String FIND_UNLOCKED_JOB_QUERY; static final String FIND_UNLOCKED_STATS_JOB_QUERY; }