TransactionSchedulerImpl implements TransactionScheduler, PriorityScheduler, ProfileListener { public void runTask(KernelRunnable task, Identity owner) throws Exception { if (isShutdown) { throw new IllegalStateException("Scheduler is shutdown"); } if (ContextResolver.isCurrentTransaction()) { task.run(); } else { ScheduledTaskImpl scheduledTask = new ScheduledTaskImpl.Builder( task, owner, defaultPriority).build(); waitForTask(scheduledTask); } } TransactionSchedulerImpl(Properties properties,
                             TransactionCoordinator transactionCoordinator,
                             ProfileCollectorHandle profileCollectorHandle,
                             AccessCoordinatorHandle accessCoordinator); TaskReservation reserveTask(KernelRunnable task, Identity owner); TaskReservation reserveTask(KernelRunnable task, Identity owner,
                                       long startTime); void scheduleTask(KernelRunnable task, Identity owner); void scheduleTask(KernelRunnable task, Identity owner,
                             long startTime); RecurringTaskHandle scheduleRecurringTask(KernelRunnable task,
                                                     Identity owner,
                                                     long startTime,
                                                     long period); TaskQueue createTaskQueue(); void runTask(KernelRunnable task, Identity owner); TaskReservation reserveTask(KernelRunnable task, Identity owner,
                                       Priority priority); void scheduleTask(KernelRunnable task, Identity owner,
                             Priority priority); void propertyChange(PropertyChangeEvent event); void report(ProfileReport profileReport); void shutdown(); static final String SCHEDULER_QUEUE_PROPERTY; static final String DEFAULT_SCHEDULER_QUEUE; static final String SCHEDULER_RETRY_PROPERTY; static final String DEFAULT_SCHEDULER_RETRY; static final String CONSUMER_THREADS_PROPERTY; static final String DEFAULT_CONSUMER_THREADS; }
@Test (expected=NullPointerException.class) public void runTaskNullTask() throws Exception { txnScheduler.runTask(null, taskOwner); }
@Test (expected=NullPointerException.class) public void runTaskNullOwner() throws Exception { txnScheduler.runTask(new TestAbstractKernelRunnable() { public void run() {} }, null); }
@Test public void runTaskSingleTask() throws Exception { RunCountTestRunner runner = new RunCountTestRunner(1); txnScheduler.runTask(runner, taskOwner); assertEquals(0, runner.getRunCount()); }
@Test public void runTaskMultipleTasks() throws Exception { RunCountTestRunner runner = new RunCountTestRunner(4); txnScheduler.runTask(runner, taskOwner); assertEquals(3, runner.getRunCount()); txnScheduler.runTask(runner, taskOwner); assertEquals(2, runner.getRunCount()); txnScheduler.runTask(runner, taskOwner); assertEquals(1, runner.getRunCount()); txnScheduler.runTask(runner, taskOwner); assertEquals(0, runner.getRunCount()); }
@Test public void runTaskTransactional() throws Exception { final TransactionProxy proxy = serverNode.getProxy(); txnScheduler.runTask(new TestAbstractKernelRunnable() { public void run() throws Exception { proxy.getCurrentTransaction(); proxy.getCurrentOwner(); } }, taskOwner); }
@Test public void runTransactionInTransaction() throws Exception { final TransactionProxy proxy = serverNode.getProxy(); KernelRunnable task = new TestAbstractKernelRunnable() { public void run() throws Exception { final Transaction t = proxy.getCurrentTransaction(); txnScheduler.runTask(new TestAbstractKernelRunnable() { public void run() throws Exception { Transaction t2 = proxy.getCurrentTransaction(); assertTrue(t.equals(t2)); } }, taskOwner); } }; txnScheduler.runTask(task, taskOwner); }
@Test public void runTransactionInterrupted() throws Exception { final AtomicInteger i = new AtomicInteger(0); final KernelRunnable r = new TestAbstractKernelRunnable() { public void run() throws Exception { if (i.getAndIncrement() == 0) throw new InterruptedException("test"); } }; try { txnScheduler.runTask(r, taskOwner); fail("Expected Interrupted Exception"); } catch (InterruptedException ie) {} assertEquals(i.get(), 1); }
@Test public void runTaskWithRetry() throws Exception { RetryTestRunner runner = new RetryTestRunner(); txnScheduler.runTask(runner, taskOwner); assertTrue(runner.isFinished()); }
@Test (expected=RuntimeException.class) public void runNonRetriedTask() throws Exception { txnScheduler.runTask(new TestAbstractKernelRunnable() { public void run() throws Exception { throw new RuntimeException("intentionally thrown"); } }, taskOwner); }
@Test (expected=RuntimeException.class) public void runNonRetriedTaskExplicitAbort() throws Exception { final TransactionProxy proxy = serverNode.getProxy(); txnScheduler.runTask(new TestAbstractKernelRunnable() { public void run() throws Exception { RuntimeException re = new RuntimeException("intentional"); proxy.getCurrentTransaction().abort(re); throw re; } }, taskOwner); }
@Test (expected=Error.class) public void testRunTransactionThrowsError() throws Exception { txnScheduler.runTask(new TestAbstractKernelRunnable() { public void run() throws Exception { throw new Error("intentionally thrown"); } }, taskOwner); }
@Test public void dropFailedTask() throws Exception { final Exception result = new Exception("task failed"); replaceRetryPolicy(createRetryPolicy(SchedulerRetryAction.DROP)); final AtomicInteger i = new AtomicInteger(0); final KernelRunnable r = new TestAbstractKernelRunnable() { public void run() throws Exception { if (i.getAndIncrement() == 0) throw result; } }; try { txnScheduler.runTask(r, taskOwner); fail("expected Exception"); } catch(Exception e) { assertEquals(result, e); } finally { assertEquals(i.get(), 1); } }
@Test public void retryFailedTask() throws Exception { final Exception result = new Exception("task failed"); replaceRetryPolicy(createRetryPolicy(SchedulerRetryAction.RETRY_NOW)); final AtomicInteger i = new AtomicInteger(0); final KernelRunnable r = new TestAbstractKernelRunnable() { public void run() throws Exception { if (i.getAndIncrement() == 0) throw result; } }; txnScheduler.runTask(r, taskOwner); assertEquals(i.get(), 2); }
@Test public void handoffFailedTask() throws Exception { final Exception result = new Exception("task failed"); replaceRetryPolicy(createRetryPolicy(SchedulerRetryAction.RETRY_LATER)); final AtomicInteger i = new AtomicInteger(0); final KernelRunnable r = new TestAbstractKernelRunnable() { public void run() throws Exception { if (i.getAndIncrement() == 0) throw result; } }; txnScheduler.runTask(r, taskOwner); assertEquals(i.get(), 2); }