You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by sn...@apache.org on 2014/01/28 23:21:12 UTC
[02/96] [abbrv] [partial] Change package namespace to
org.apache.usergrid
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/test/java/org/apache/usergrid/CoreITSetupImpl.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/CoreITSetupImpl.java b/stack/core/src/test/java/org/apache/usergrid/CoreITSetupImpl.java
new file mode 100644
index 0000000..eee42af
--- /dev/null
+++ b/stack/core/src/test/java/org/apache/usergrid/CoreITSetupImpl.java
@@ -0,0 +1,145 @@
+package org.apache.usergrid;
+
+
+import java.util.UUID;
+
+import org.apache.usergrid.cassandra.CassandraResource;
+import org.apache.usergrid.mq.QueueManagerFactory;
+import org.apache.usergrid.persistence.EntityManagerFactory;
+import org.apache.usergrid.persistence.IndexBucketLocator;
+import org.apache.usergrid.persistence.cassandra.CassandraService;
+import org.apache.usergrid.utils.JsonUtils;
+import org.junit.runner.Description;
+import org.junit.runners.model.Statement;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class CoreITSetupImpl implements CoreITSetup {
+ private static final Logger LOG = LoggerFactory.getLogger( CoreITSetupImpl.class );
+
+ protected EntityManagerFactory emf;
+ protected QueueManagerFactory qmf;
+ protected IndexBucketLocator indexBucketLocator;
+ protected CassandraService cassandraService;
+ protected CassandraResource cassandraResource;
+ protected boolean enabled = false;
+
+
+ public CoreITSetupImpl( CassandraResource cassandraResource ) {
+ this.cassandraResource = cassandraResource;
+ }
+
+
+ @Override
+ public Statement apply( Statement base, Description description ) {
+ return statement( base, description );
+ }
+
+
+ private Statement statement( final Statement base, final Description description ) {
+ return new Statement() {
+ @Override
+ public void evaluate() throws Throwable {
+ before( description );
+
+ try {
+ base.evaluate();
+ }
+ finally {
+ after( description );
+ }
+ }
+ };
+ }
+
+
+ /**
+ * Sets up the resources for the test here.
+ *
+ * @throws Throwable if setup fails (which will disable {@code after}
+ */
+ protected void before( Description description ) throws Throwable {
+ LOG.info( "Setting up for {}", description.getDisplayName() );
+ initialize();
+ }
+
+
+ private void initialize() {
+ if ( !enabled ) {
+ emf = cassandraResource.getBean( EntityManagerFactory.class );
+ qmf = cassandraResource.getBean( QueueManagerFactory.class );
+ indexBucketLocator = cassandraResource.getBean( IndexBucketLocator.class );
+ cassandraService = cassandraResource.getBean( CassandraService.class );
+ enabled = true;
+ }
+ }
+
+
+ /** Override to tear down your specific external resource. */
+ protected void after( Description description ) {
+ LOG.info( "Tearing down for {}", description.getDisplayName() );
+ }
+
+
+ @Override
+ public EntityManagerFactory getEmf() {
+ if ( emf == null ) {
+ initialize();
+ }
+
+ return emf;
+ }
+
+
+ @Override
+ public QueueManagerFactory getQmf() {
+ if ( qmf == null ) {
+ initialize();
+ }
+
+ return qmf;
+ }
+
+
+ @Override
+ public IndexBucketLocator getIbl() {
+ if ( indexBucketLocator == null ) {
+ initialize();
+ }
+
+ return indexBucketLocator;
+ }
+
+
+ @Override
+ public CassandraService getCassSvc() {
+ if ( cassandraService == null ) {
+ initialize();
+ }
+
+ return cassandraService;
+ }
+
+
+ @Override
+ public UUID createApplication( String organizationName, String applicationName ) throws Exception {
+ if ( USE_DEFAULT_APPLICATION ) {
+ return CassandraService.DEFAULT_APPLICATION_ID;
+ }
+
+ if ( emf == null ) {
+ emf = cassandraResource.getBean( EntityManagerFactory.class );
+ }
+
+ return emf.createApplication( organizationName, applicationName );
+ }
+
+
+ @Override
+ public void dump( String name, Object obj ) {
+ if ( obj != null && LOG.isInfoEnabled() ) {
+ LOG.info( name + ":\n" + JsonUtils.mapToFormattedJsonString( obj ) );
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/test/java/org/apache/usergrid/CoreITSuite.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/CoreITSuite.java b/stack/core/src/test/java/org/apache/usergrid/CoreITSuite.java
new file mode 100644
index 0000000..f08916d
--- /dev/null
+++ b/stack/core/src/test/java/org/apache/usergrid/CoreITSuite.java
@@ -0,0 +1,34 @@
+package org.apache.usergrid;
+
+
+import org.apache.usergrid.cassandra.CassandraResource;
+import org.apache.usergrid.cassandra.Concurrent;
+import org.apache.usergrid.locking.cassandra.HectorLockManagerIT;
+import org.apache.usergrid.mq.MessagesIT;
+import org.apache.usergrid.persistence.CollectionIT;
+import org.apache.usergrid.persistence.CounterIT;
+import org.apache.usergrid.persistence.EntityConnectionsIT;
+import org.apache.usergrid.persistence.EntityDictionaryIT;
+import org.apache.usergrid.persistence.EntityManagerIT;
+import org.apache.usergrid.persistence.GeoIT;
+import org.apache.usergrid.persistence.IndexIT;
+import org.apache.usergrid.persistence.PathQueryIT;
+import org.apache.usergrid.persistence.PermissionsIT;
+import org.apache.usergrid.persistence.cassandra.EntityManagerFactoryImplIT;
+import org.apache.usergrid.system.UsergridSystemMonitorIT;
+import org.junit.ClassRule;
+import org.junit.runner.RunWith;
+import org.junit.runners.Suite;
+
+
+@RunWith(Suite.class)
+@Suite.SuiteClasses({
+ HectorLockManagerIT.class, UsergridSystemMonitorIT.class, CollectionIT.class, CounterIT.class,
+ EntityConnectionsIT.class, EntityDictionaryIT.class, EntityManagerIT.class, GeoIT.class, IndexIT.class,
+ MessagesIT.class, PermissionsIT.class, PathQueryIT.class, EntityManagerFactoryImplIT.class
+})
+@Concurrent()
+public class CoreITSuite {
+ @ClassRule
+ public static CassandraResource cassandraResource = CassandraResource.newWithAvailablePorts( "coreManager" );
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/test/java/org/apache/usergrid/CoreTestSuite.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/CoreTestSuite.java b/stack/core/src/test/java/org/apache/usergrid/CoreTestSuite.java
new file mode 100644
index 0000000..3b45c0a
--- /dev/null
+++ b/stack/core/src/test/java/org/apache/usergrid/CoreTestSuite.java
@@ -0,0 +1,47 @@
+/*
+ * Copyright 2012 Apigee Corporation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid;
+
+
+import org.apache.usergrid.cassandra.Concurrent;
+import org.apache.usergrid.locking.zookeeper.ZookeeperLockManagerTest;
+import org.apache.usergrid.mq.QueuePathsTest;
+import org.apache.usergrid.persistence.EntityTest;
+import org.apache.usergrid.persistence.QueryTest;
+import org.apache.usergrid.persistence.QueryUtilsTest;
+import org.apache.usergrid.persistence.SchemaTest;
+import org.apache.usergrid.persistence.UtilsTest;
+import org.apache.usergrid.persistence.cassandra.QueryProcessorTest;
+import org.apache.usergrid.persistence.cassandra.SimpleIndexBucketLocatorImplTest;
+import org.apache.usergrid.persistence.query.ir.result.IntersectionIteratorTest;
+import org.apache.usergrid.persistence.query.ir.result.SubtractionIteratorTest;
+import org.apache.usergrid.persistence.query.ir.result.UnionIteratorTest;
+import org.apache.usergrid.persistence.query.tree.GrammarTreeTest;
+import org.apache.usergrid.persistence.query.tree.LongLiteralTest;
+import org.apache.usergrid.persistence.query.tree.StringLiteralTest;
+import org.junit.runner.RunWith;
+import org.junit.runners.Suite;
+
+
+@RunWith(Suite.class)
+@Suite.SuiteClasses({
+ ZookeeperLockManagerTest.class, QueuePathsTest.class, QueryProcessorTest.class,
+ SimpleIndexBucketLocatorImplTest.class, EntityTest.class, QueryTest.class, QueryUtilsTest.class,
+ SchemaTest.class, UtilsTest.class, IntersectionIteratorTest.class, SubtractionIteratorTest.class,
+ UnionIteratorTest.class, GrammarTreeTest.class, LongLiteralTest.class, StringLiteralTest.class
+})
+@Concurrent()
+public class CoreTestSuite {}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/test/java/org/apache/usergrid/batch/AppArgsTest.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/batch/AppArgsTest.java b/stack/core/src/test/java/org/apache/usergrid/batch/AppArgsTest.java
new file mode 100644
index 0000000..73b193a
--- /dev/null
+++ b/stack/core/src/test/java/org/apache/usergrid/batch/AppArgsTest.java
@@ -0,0 +1,53 @@
+package org.apache.usergrid.batch;
+
+
+import org.apache.usergrid.batch.AppArgs;
+import org.apache.usergrid.cassandra.Concurrent;
+import org.junit.Test;
+
+import com.google.common.base.CharMatcher;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+
+/** @author zznate */
+@Concurrent()
+public class AppArgsTest {
+
+ @Test
+ public void verifyDefaults() {
+ AppArgs aa = AppArgs.parseArgs( new String[] { "" } );
+ assertEquals( "127.0.0.1", aa.getHost() );
+ assertEquals( 9160, aa.getPort() );
+ }
+
+
+ @Test
+ public void verifyArgs() {
+ AppArgs aa =
+ AppArgs.parseArgs( new String[] { "-host", "127.0.0.2", "-appContext", "classpath:/appContext.xml" } );
+ assertEquals( "127.0.0.2", aa.getHost() );
+ assertNotNull( aa.getAppContext() );
+ }
+
+
+ @Test
+ public void verifyContextSwitch() {
+ AppArgs appArgs = AppArgs.parseArgs( new String[] { "-appContext", "classpath:/appContext.xml" } );
+ assertEquals( "/appContext.xml", getIndex( appArgs.getAppContext() ) );
+ appArgs = AppArgs.parseArgs( new String[] { "-appContext", "/appContext.xml" } );
+ assertEquals( "/appContext.xml", getIndex( appArgs.getAppContext() ) );
+ }
+
+
+ private String getIndex( String path ) {
+ int index = CharMatcher.is( ':' ).indexIn( path );
+ if ( index > 0 ) {
+ return path.substring( ++index );
+ }
+ else {
+ return path;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/test/java/org/apache/usergrid/batch/BulkJobExecutionUnitTest.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/batch/BulkJobExecutionUnitTest.java b/stack/core/src/test/java/org/apache/usergrid/batch/BulkJobExecutionUnitTest.java
new file mode 100644
index 0000000..501bc53
--- /dev/null
+++ b/stack/core/src/test/java/org/apache/usergrid/batch/BulkJobExecutionUnitTest.java
@@ -0,0 +1,211 @@
+package org.apache.usergrid.batch;
+
+
+import java.util.UUID;
+
+import org.apache.usergrid.batch.JobExecution;
+import org.apache.usergrid.batch.JobExecutionImpl;
+import org.apache.usergrid.batch.JobExecution.Status;
+import org.apache.usergrid.batch.repository.JobDescriptor;
+import org.apache.usergrid.cassandra.Concurrent;
+import org.apache.usergrid.persistence.entities.JobData;
+import org.apache.usergrid.persistence.entities.JobStat;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+
+/**
+ * @author zznate
+ * @author tnine
+ */
+@Concurrent()
+public class BulkJobExecutionUnitTest {
+
+ @Test
+ public void transitionsOk() {
+ JobData data = new JobData();
+ JobStat stat = new JobStat();
+ JobDescriptor jobDescriptor = new JobDescriptor( "", UUID.randomUUID(), UUID.randomUUID(), data, stat, null );
+ JobExecution bje = new JobExecutionImpl( jobDescriptor );
+ assertEquals( JobExecution.Status.NOT_STARTED, bje.getStatus() );
+ bje.start( 1 );
+ assertEquals( JobExecution.Status.IN_PROGRESS, bje.getStatus() );
+ bje.completed();
+ assertEquals( JobExecution.Status.COMPLETED, bje.getStatus() );
+ }
+
+
+ @Test
+ public void transitionsDead() {
+ JobData data = new JobData();
+ JobStat stat = new JobStat();
+ JobDescriptor jobDescriptor = new JobDescriptor( "", UUID.randomUUID(), UUID.randomUUID(), data, stat, null );
+ JobExecution bje = new JobExecutionImpl( jobDescriptor );
+ assertEquals( JobExecution.Status.NOT_STARTED, bje.getStatus() );
+ bje.start( 1 );
+ assertEquals( JobExecution.Status.IN_PROGRESS, bje.getStatus() );
+ bje.killed();
+ assertEquals( JobExecution.Status.DEAD, bje.getStatus() );
+ }
+
+
+ @Test
+ public void transitionsRetry() {
+ JobData data = new JobData();
+ JobStat stat = new JobStat();
+ JobDescriptor jobDescriptor = new JobDescriptor( "", UUID.randomUUID(), UUID.randomUUID(), data, stat, null );
+ JobExecution bje = new JobExecutionImpl( jobDescriptor );
+ assertEquals( JobExecution.Status.NOT_STARTED, bje.getStatus() );
+ bje.start( JobExecution.FOREVER );
+ assertEquals( JobExecution.Status.IN_PROGRESS, bje.getStatus() );
+ bje.failed();
+ assertEquals( JobExecution.Status.FAILED, bje.getStatus() );
+ }
+
+
+ @Test
+ public void transitionFail() {
+ JobData data = new JobData();
+ JobStat stat = new JobStat();
+ JobDescriptor jobDescriptor = new JobDescriptor( "", UUID.randomUUID(), UUID.randomUUID(), data, stat, null );
+ JobExecution bje = new JobExecutionImpl( jobDescriptor );
+ try {
+ bje.completed();
+ fail( "Should have throw ISE on NOT_STARTED to IN_PROGRESS" );
+ }
+ catch ( IllegalStateException ise ) {
+ }
+
+ try {
+ bje.failed();
+ fail( "Should have thrown ISE on NOT_STARTED to FAILED" );
+ }
+ catch ( IllegalStateException ise ) {
+ }
+ bje.start( 1 );
+
+ bje.completed();
+ try {
+ bje.failed();
+ fail( "Should have failed failed after complete call" );
+ }
+ catch ( IllegalStateException ise ) {
+ }
+ }
+
+
+ @Test
+ public void transitionFailOnDeath() {
+ JobData data = new JobData();
+ JobStat stat = new JobStat();
+ JobDescriptor jobDescriptor = new JobDescriptor( "", UUID.randomUUID(), UUID.randomUUID(), data, stat, null );
+ JobExecution bje = new JobExecutionImpl( jobDescriptor );
+ try {
+ bje.completed();
+ fail( "Should have throw ISE on NOT_STARTED to IN_PROGRESS" );
+ }
+ catch ( IllegalStateException ise ) {
+ }
+
+ try {
+ bje.failed();
+ fail( "Should have thrown ISE on NOT_STARTED to FAILED" );
+ }
+ catch ( IllegalStateException ise ) {
+ }
+ bje.start( 1 );
+
+ bje.killed();
+ try {
+ bje.killed();
+ fail( "Should have failed failed after complete call" );
+ }
+ catch ( IllegalStateException ise ) {
+ }
+ }
+
+
+ @Test
+ public void failureTriggerCount() {
+ JobData data = new JobData();
+ JobStat stat = new JobStat();
+ JobDescriptor jobDescriptor = new JobDescriptor( "", UUID.randomUUID(), UUID.randomUUID(), data, stat, null );
+ JobExecution bje = new JobExecutionImpl( jobDescriptor );
+
+ bje.start( 1 );
+
+ assertEquals( Status.IN_PROGRESS, bje.getStatus() );
+ assertEquals( 1, stat.getRunCount() );
+
+
+ bje.failed();
+
+ assertEquals( Status.FAILED, bje.getStatus() );
+ assertEquals( 1, stat.getRunCount() );
+
+
+ // now fail again, we should trigger a state change
+ bje = new JobExecutionImpl( jobDescriptor );
+ bje.start( 1 );
+
+ assertEquals( Status.DEAD, bje.getStatus() );
+ assertEquals( 2, stat.getRunCount() );
+ }
+
+
+ @Test
+ public void failureTriggerNoTrip() {
+ JobData data = new JobData();
+ JobStat stat = new JobStat();
+ JobDescriptor jobDescriptor = new JobDescriptor( "", UUID.randomUUID(), UUID.randomUUID(), data, stat, null );
+ JobExecution bje = new JobExecutionImpl( jobDescriptor );
+
+ bje.start( JobExecution.FOREVER );
+
+ assertEquals( Status.IN_PROGRESS, bje.getStatus() );
+ assertEquals( 1, stat.getRunCount() );
+
+ bje.failed();
+
+ assertEquals( Status.FAILED, bje.getStatus() );
+ assertEquals( 1, stat.getRunCount() );
+
+ // now fail again, we should trigger a state change
+ bje = new JobExecutionImpl( jobDescriptor );
+ bje.start( JobExecution.FOREVER );
+
+ assertEquals( Status.IN_PROGRESS, bje.getStatus() );
+ assertEquals( 2, stat.getRunCount() );
+
+ bje.failed();
+
+ assertEquals( Status.FAILED, bje.getStatus() );
+ assertEquals( 2, stat.getRunCount() );
+ }
+
+
+ @Test
+ public void doubleInvokeFail() {
+ JobData data = new JobData();
+ JobStat stat = new JobStat();
+ JobDescriptor jobDescriptor = new JobDescriptor( "", UUID.randomUUID(), UUID.randomUUID(), data, stat, null );
+ JobExecution bje = new JobExecutionImpl( jobDescriptor );
+ bje.start( 1 );
+ try {
+ bje.start( 1 );
+ fail( "Should have failed on double start() call" );
+ }
+ catch ( IllegalStateException ise ) {
+ }
+
+ bje.completed();
+ try {
+ bje.completed();
+ fail( "Should have failed on double complete call" );
+ }
+ catch ( IllegalStateException ise ) {
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/test/java/org/apache/usergrid/batch/BulkTestUtils.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/batch/BulkTestUtils.java b/stack/core/src/test/java/org/apache/usergrid/batch/BulkTestUtils.java
new file mode 100644
index 0000000..01ea153
--- /dev/null
+++ b/stack/core/src/test/java/org/apache/usergrid/batch/BulkTestUtils.java
@@ -0,0 +1,41 @@
+package org.apache.usergrid.batch;
+
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.usergrid.batch.Job;
+import org.apache.usergrid.batch.JobExecution;
+import org.apache.usergrid.batch.JobFactory;
+import org.apache.usergrid.batch.repository.JobDescriptor;
+import org.junit.Ignore;
+
+
+/** @author zznate */
+@Ignore
+public class BulkTestUtils {
+
+ public static JobFactory getBulkJobFactory() {
+ return new MyBulkJobFactory();
+ }
+
+
+ private static class MyBulkJobFactory implements JobFactory {
+ /* (non-Javadoc)
+ * @see org.apache.usergrid.batch.JobFactory#jobsFrom(org.apache.usergrid.batch.repository.JobDescriptor)
+ */
+ @Override
+ public List<Job> jobsFrom( JobDescriptor descriptor ) {
+ return Arrays.asList( new Job[] { new MyBulkJob() } );
+ }
+ }
+
+
+ private static class MyBulkJob implements Job {
+ @Override
+ public void execute( JobExecution execution ) throws Exception {
+ // do some stuff
+
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/test/java/org/apache/usergrid/batch/ConcurrentSchedulerITSuite.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/batch/ConcurrentSchedulerITSuite.java b/stack/core/src/test/java/org/apache/usergrid/batch/ConcurrentSchedulerITSuite.java
new file mode 100644
index 0000000..af1ce02
--- /dev/null
+++ b/stack/core/src/test/java/org/apache/usergrid/batch/ConcurrentSchedulerITSuite.java
@@ -0,0 +1,31 @@
+package org.apache.usergrid.batch;
+
+
+import org.apache.usergrid.batch.job.SchedulerRuntime1IT;
+import org.apache.usergrid.batch.job.SchedulerRuntime2IT;
+import org.apache.usergrid.batch.job.SchedulerRuntime3IT;
+import org.apache.usergrid.batch.job.SchedulerRuntime4IT;
+import org.apache.usergrid.batch.job.SchedulerRuntime5IT;
+import org.apache.usergrid.batch.job.SchedulerRuntime6IT;
+import org.apache.usergrid.batch.job.SchedulerRuntime7IT;
+import org.apache.usergrid.batch.job.SchedulerRuntime8IT;
+import org.apache.usergrid.cassandra.CassandraResource;
+import org.apache.usergrid.cassandra.Concurrent;
+import org.apache.usergrid.cassandra.ConcurrentSuite;
+import org.junit.ClassRule;
+import org.junit.runner.RunWith;
+import org.junit.runners.Suite;
+
+
+@RunWith(ConcurrentSuite.class)
+@Suite.SuiteClasses(
+ {
+ SchedulerRuntime1IT.class, SchedulerRuntime2IT.class, SchedulerRuntime3IT.class,
+ SchedulerRuntime4IT.class, SchedulerRuntime5IT.class, SchedulerRuntime6IT.class,
+ SchedulerRuntime7IT.class, SchedulerRuntime8IT.class
+ })
+@Concurrent()
+public class ConcurrentSchedulerITSuite {
+ @ClassRule
+ public static CassandraResource cassandraResource = CassandraResource.newWithAvailablePorts();
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/test/java/org/apache/usergrid/batch/ConcurrentSchedulerTestSuite.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/batch/ConcurrentSchedulerTestSuite.java b/stack/core/src/test/java/org/apache/usergrid/batch/ConcurrentSchedulerTestSuite.java
new file mode 100644
index 0000000..debd291
--- /dev/null
+++ b/stack/core/src/test/java/org/apache/usergrid/batch/ConcurrentSchedulerTestSuite.java
@@ -0,0 +1,19 @@
+package org.apache.usergrid.batch;
+
+
+import org.apache.usergrid.batch.AppArgsTest;
+import org.apache.usergrid.batch.BulkJobExecutionUnitTest;
+import org.apache.usergrid.batch.UsergridJobFactoryTest;
+import org.apache.usergrid.cassandra.Concurrent;
+import org.apache.usergrid.cassandra.ConcurrentSuite;
+import org.junit.runner.RunWith;
+import org.junit.runners.Suite;
+
+
+@RunWith(ConcurrentSuite.class)
+@Suite.SuiteClasses(
+ {
+ AppArgsTest.class, UsergridJobFactoryTest.class, BulkJobExecutionUnitTest.class,
+ })
+@Concurrent()
+public class ConcurrentSchedulerTestSuite {}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/test/java/org/apache/usergrid/batch/SchedulerITSuite.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/batch/SchedulerITSuite.java b/stack/core/src/test/java/org/apache/usergrid/batch/SchedulerITSuite.java
new file mode 100644
index 0000000..804e9bf
--- /dev/null
+++ b/stack/core/src/test/java/org/apache/usergrid/batch/SchedulerITSuite.java
@@ -0,0 +1,34 @@
+package org.apache.usergrid.batch;
+
+
+import org.apache.usergrid.batch.job.SchedulerRuntime1IT;
+import org.apache.usergrid.batch.job.SchedulerRuntime2IT;
+import org.apache.usergrid.batch.job.SchedulerRuntime3IT;
+import org.apache.usergrid.batch.job.SchedulerRuntime4IT;
+import org.apache.usergrid.batch.job.SchedulerRuntime5IT;
+import org.apache.usergrid.batch.job.SchedulerRuntime6IT;
+import org.apache.usergrid.batch.job.SchedulerRuntime7IT;
+import org.apache.usergrid.batch.job.SchedulerRuntime8IT;
+import org.apache.usergrid.cassandra.CassandraResource;
+import org.apache.usergrid.cassandra.Concurrent;
+import org.junit.ClassRule;
+import org.junit.Ignore;
+import org.junit.runner.RunWith;
+import org.junit.runners.Suite;
+
+
+@RunWith(Suite.class)
+@Suite.SuiteClasses(
+ {
+ SchedulerRuntime1IT.class, SchedulerRuntime2IT.class, SchedulerRuntime3IT.class,
+ SchedulerRuntime4IT.class, SchedulerRuntime5IT.class, SchedulerRuntime6IT.class,
+ SchedulerRuntime7IT.class, SchedulerRuntime8IT.class
+ })
+@Concurrent()
+@Ignore("TODO: Todd fix. Does not reliably pass on our build server.")
+// TODO - this suite actually runs correctly now so we can
+// remove this ignore if Todd is OK with it.
+public class SchedulerITSuite {
+ @ClassRule
+ public static CassandraResource cassandraResource = CassandraResource.newWithAvailablePorts();
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/test/java/org/apache/usergrid/batch/SchedulerTestSuite.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/batch/SchedulerTestSuite.java b/stack/core/src/test/java/org/apache/usergrid/batch/SchedulerTestSuite.java
new file mode 100644
index 0000000..2509568
--- /dev/null
+++ b/stack/core/src/test/java/org/apache/usergrid/batch/SchedulerTestSuite.java
@@ -0,0 +1,18 @@
+package org.apache.usergrid.batch;
+
+
+import org.apache.usergrid.batch.AppArgsTest;
+import org.apache.usergrid.batch.BulkJobExecutionUnitTest;
+import org.apache.usergrid.batch.UsergridJobFactoryTest;
+import org.apache.usergrid.cassandra.Concurrent;
+import org.junit.runner.RunWith;
+import org.junit.runners.Suite;
+
+
+@RunWith(Suite.class)
+@Suite.SuiteClasses(
+ {
+ AppArgsTest.class, UsergridJobFactoryTest.class, BulkJobExecutionUnitTest.class,
+ })
+@Concurrent()
+public class SchedulerTestSuite {}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/test/java/org/apache/usergrid/batch/UsergridJobFactoryTest.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/batch/UsergridJobFactoryTest.java b/stack/core/src/test/java/org/apache/usergrid/batch/UsergridJobFactoryTest.java
new file mode 100644
index 0000000..6c44b33
--- /dev/null
+++ b/stack/core/src/test/java/org/apache/usergrid/batch/UsergridJobFactoryTest.java
@@ -0,0 +1,33 @@
+package org.apache.usergrid.batch;
+
+
+import java.util.List;
+import java.util.UUID;
+
+import org.apache.usergrid.batch.Job;
+import org.apache.usergrid.batch.JobNotFoundException;
+import org.apache.usergrid.batch.repository.JobDescriptor;
+import org.apache.usergrid.cassandra.Concurrent;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+
+/** @author zznate */
+@Concurrent()
+public class UsergridJobFactoryTest {
+
+ private static UUID jobId = UUID.randomUUID();
+
+
+ @Test
+ public void verifyBuildup() throws JobNotFoundException {
+ JobDescriptor jobDescriptor = new JobDescriptor( "", jobId, UUID.randomUUID(), null, null, null );
+
+
+ List<Job> bulkJobs = BulkTestUtils.getBulkJobFactory().jobsFrom( jobDescriptor );
+ assertNotNull( bulkJobs );
+ assertEquals( 1, bulkJobs.size() );
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/test/java/org/apache/usergrid/batch/job/AbstractSchedulerRuntimeIT.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/batch/job/AbstractSchedulerRuntimeIT.java b/stack/core/src/test/java/org/apache/usergrid/batch/job/AbstractSchedulerRuntimeIT.java
new file mode 100644
index 0000000..377ed08
--- /dev/null
+++ b/stack/core/src/test/java/org/apache/usergrid/batch/job/AbstractSchedulerRuntimeIT.java
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.batch.job;
+
+
+import java.util.Properties;
+
+import com.google.common.util.concurrent.Service.State;
+
+import org.apache.usergrid.batch.SchedulerITSuite;
+import org.apache.usergrid.batch.service.JobSchedulerService;
+import org.apache.usergrid.batch.service.SchedulerService;
+import org.apache.usergrid.cassandra.CassandraResource;
+import org.junit.Before;
+
+
+/**
+ * Class to test job runtimes
+ */
+public class AbstractSchedulerRuntimeIT {
+ protected static final int DEFAULT_COUNT = 10;
+ protected static final String COUNT_PROP = AbstractSchedulerRuntimeIT.class.getCanonicalName();
+ protected static final String TIMEOUT_PROP = "usergrid.scheduler.job.timeout";
+ protected static final String RUNLOOP_PROP = "usergrid.scheduler.job.interval";
+ protected static final String FAIL_PROP = "usergrid.scheduler.job.maxfail";
+
+ public static CassandraResource cassandraResource = SchedulerITSuite.cassandraResource;
+
+ private TestJobListener listener = new TestJobListener();
+ private int count = DEFAULT_COUNT;
+ protected SchedulerService scheduler;
+ protected Properties props;
+ protected long waitTime = listener.WAIT_MAX_MILLIS;
+
+
+ @Before
+ @SuppressWarnings( "all" )
+ public void setup() {
+ scheduler = cassandraResource.getBean( SchedulerService.class );
+ props = cassandraResource.getBean( "properties", Properties.class );
+
+ if ( System.getProperties().containsKey( COUNT_PROP ) ) {
+ count = Integer.getInteger( System.getProperty( COUNT_PROP ) );
+ }
+
+ // start the scheduler after we're all set up
+ JobSchedulerService jobScheduler = cassandraResource.getBean( JobSchedulerService.class );
+ jobScheduler.setJobListener( listener );
+ if ( jobScheduler.state() != State.RUNNING ) {
+ jobScheduler.startAndWait();
+ }
+ }
+
+
+ protected int getCount() {
+ return count;
+ }
+
+
+ protected TestJobListener getJobListener() {
+ return listener;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/test/java/org/apache/usergrid/batch/job/CountdownLatchJob.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/batch/job/CountdownLatchJob.java b/stack/core/src/test/java/org/apache/usergrid/batch/job/CountdownLatchJob.java
new file mode 100644
index 0000000..61c7560
--- /dev/null
+++ b/stack/core/src/test/java/org/apache/usergrid/batch/job/CountdownLatchJob.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.batch.job;
+
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.usergrid.batch.Job;
+import org.apache.usergrid.batch.JobExecution;
+import org.junit.Ignore;
+import org.springframework.stereotype.Component;
+
+
+/**
+ * A simple job that does nothing but increment an atomic counter
+ *
+ * @author tnine
+ */
+@Component("countdownLatch")
+@Ignore("Not a test")
+public class CountdownLatchJob implements Job {
+
+ private CountDownLatch latch = null;
+
+
+ /**
+ *
+ */
+ public CountdownLatchJob() {
+ }
+
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see org.apache.usergrid.batch.Job#execute(org.apache.usergrid.batch.JobExecution)
+ */
+ @Override
+ public void execute( JobExecution execution ) throws Exception {
+ latch.countDown();
+ }
+
+
+ public void setLatch( int size ) {
+ latch = new CountDownLatch( size );
+ }
+
+
+ public boolean waitForCount( long timeout, TimeUnit unit ) throws InterruptedException {
+ return latch.await( timeout, unit );
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/test/java/org/apache/usergrid/batch/job/DelayExecution.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/batch/job/DelayExecution.java b/stack/core/src/test/java/org/apache/usergrid/batch/job/DelayExecution.java
new file mode 100644
index 0000000..23b4206
--- /dev/null
+++ b/stack/core/src/test/java/org/apache/usergrid/batch/job/DelayExecution.java
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.batch.job;
+
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.usergrid.batch.Job;
+import org.apache.usergrid.batch.JobExecution;
+import org.junit.Ignore;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+
+/**
+ * A simple job that does nothing but increment an atomic counter
+ *
+ * @author tnine
+ */
+@Component("delayExecution")
+@Ignore("Not a test")
+public class DelayExecution implements Job {
+
+ private static final Logger logger = LoggerFactory.getLogger( DelayExecution.class );
+
+ private CountDownLatch latch = null;
+ private long timeout;
+ private int delayCount;
+
+
+ /**
+ *
+ */
+ public DelayExecution() {
+ }
+
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see org.apache.usergrid.batch.Job#execute(org.apache.usergrid.batch.JobExecution)
+ */
+ @Override
+ public void execute( JobExecution execution ) throws Exception {
+
+ logger.info( "Running delay execution" );
+
+ if ( latch.getCount() > 1 ) {
+ execution.delay( timeout );
+ }
+
+ latch.countDown();
+ }
+
+
+ public void setLatch( int calls ) {
+ latch = new CountDownLatch( calls );
+ }
+
+
+ public boolean waitForCount( long timeout, TimeUnit unit ) throws InterruptedException {
+ return latch.await( timeout, unit );
+ }
+
+
+ /** @return the timeout */
+ public long getTimeout() {
+ return timeout;
+ }
+
+
+ /** @param timeout the timeout to set */
+ public void setTimeout( long timeout ) {
+ this.timeout = timeout;
+ }
+
+
+ /** @return the delayCount */
+ public int getDelayCount() {
+ return delayCount;
+ }
+
+
+ /** @param delayCount the delayCount to set */
+ public void setDelayCount( int delayCount ) {
+ this.delayCount = delayCount;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/test/java/org/apache/usergrid/batch/job/DelayHeartbeat.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/batch/job/DelayHeartbeat.java b/stack/core/src/test/java/org/apache/usergrid/batch/job/DelayHeartbeat.java
new file mode 100644
index 0000000..3b9204a
--- /dev/null
+++ b/stack/core/src/test/java/org/apache/usergrid/batch/job/DelayHeartbeat.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.batch.job;
+
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.usergrid.batch.Job;
+import org.apache.usergrid.batch.JobExecution;
+import org.junit.Ignore;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+
+/**
+ * A simple job that does nothing but increment an atomic counter and use heartbeat delay
+ *
+ * @author tnine
+ */
+@Component("delayHeartbeat")
+@Ignore("Not a test")
+public class DelayHeartbeat implements Job {
+
+ private static final Logger logger = LoggerFactory.getLogger( DelayHeartbeat.class );
+
+ private CountDownLatch latch = null;
+ private long timeout;
+ private int delayCount;
+
+
+ /**
+ *
+ */
+ public DelayHeartbeat() {
+ }
+
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see org.apache.usergrid.batch.Job#execute(org.apache.usergrid.batch.JobExecution)
+ */
+ @Override
+ public void execute( JobExecution execution ) throws Exception {
+
+
+ while ( latch.getCount() > 1 ) {
+ logger.info( "Running heartbeat execution" );
+ execution.heartbeat( timeout );
+ latch.countDown();
+ Thread.sleep( timeout - 1 );
+ }
+
+ latch.countDown();
+ }
+
+
+ public void setLatch( int calls ) {
+ latch = new CountDownLatch( calls );
+ }
+
+
+ public boolean waitForCount( long timeout, TimeUnit unit ) throws InterruptedException {
+ return latch.await( timeout, unit );
+ }
+
+
+ /** @return the timeout */
+ public long getTimeout() {
+ return timeout;
+ }
+
+
+ /** @param timeout the timeout to set */
+ public void setTimeout( long timeout ) {
+ this.timeout = timeout;
+ }
+
+
+ /** @return the delayCount */
+ public int getDelayCount() {
+ return delayCount;
+ }
+
+
+ /** @param delayCount the delayCount to set */
+ public void setDelayCount( int delayCount ) {
+ this.delayCount = delayCount;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/test/java/org/apache/usergrid/batch/job/FailureJobExceuction.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/batch/job/FailureJobExceuction.java b/stack/core/src/test/java/org/apache/usergrid/batch/job/FailureJobExceuction.java
new file mode 100644
index 0000000..f13742e
--- /dev/null
+++ b/stack/core/src/test/java/org/apache/usergrid/batch/job/FailureJobExceuction.java
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.batch.job;
+
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.usergrid.batch.Job;
+import org.apache.usergrid.batch.JobExecution;
+import org.junit.Ignore;
+import org.springframework.stereotype.Component;
+
+
+/**
+ * A simple job that does nothing but increment an atomic counter
+ *
+ * @author tnine
+ */
+@Component("failureJobExceuction")
+@Ignore("Not a test")
+public class FailureJobExceuction implements Job {
+
+ private CountDownLatch latch = null;
+
+
+ /**
+ *
+ */
+ public FailureJobExceuction() {
+ }
+
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see org.apache.usergrid.batch.Job#execute(org.apache.usergrid.batch.JobExecution)
+ */
+ @Override
+ public void execute( JobExecution execution ) throws Exception {
+ latch.countDown();
+
+ throw new RuntimeException( "Job Failed" );
+ }
+
+
+ public void setLatch( int calls ) {
+ latch = new CountDownLatch( calls );
+ }
+
+
+ public boolean waitForCount( long timeout, TimeUnit unit ) throws InterruptedException {
+ return latch.await( timeout, unit );
+ }
+
+
+ public long getLatchCount() {
+ return latch.getCount();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/test/java/org/apache/usergrid/batch/job/OnlyOnceExceution.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/batch/job/OnlyOnceExceution.java b/stack/core/src/test/java/org/apache/usergrid/batch/job/OnlyOnceExceution.java
new file mode 100644
index 0000000..7f6305c
--- /dev/null
+++ b/stack/core/src/test/java/org/apache/usergrid/batch/job/OnlyOnceExceution.java
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.batch.job;
+
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.usergrid.batch.JobExecution;
+import org.apache.usergrid.batch.job.OnlyOnceJob;
+import org.junit.Ignore;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+
+/**
+ * A job that will sleep for the amount of time specified. Used to check that our counter is only ever run once.
+ *
+ * @author tnine
+ */
+@Component("onlyOnceExceution")
+@Ignore("Not a test")
+public class OnlyOnceExceution extends OnlyOnceJob {
+
+ private static final Logger logger = LoggerFactory.getLogger( OnlyOnceExceution.class );
+
+ private CountDownLatch latch = null;
+ private CountDownLatch sleptLatch = new CountDownLatch( 1 );
+ private long timeout;
+ private boolean slept = false;
+ private long delay;
+
+
+ /**
+ *
+ */
+ public OnlyOnceExceution() {
+ }
+
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see
+ * org.apache.usergrid.batch.job.OnlyOnceJob#doJob(org.apache.usergrid.batch.JobExecution)
+ */
+ @Override
+ protected void doJob( JobExecution execution ) throws Exception {
+ logger.info( "Running only once execution" );
+
+ latch.countDown();
+
+
+ if ( !slept ) {
+ logger.info( "Sleeping in only once execution" );
+ Thread.sleep( timeout );
+ slept = true;
+ sleptLatch.countDown();
+ }
+ }
+
+
+ /* (non-Javadoc)
+ * @see org.apache.usergrid.batch.job.OnlyOnceJob#getDelay(org.apache.usergrid.batch.JobExecution)
+ */
+ @Override
+ protected long getDelay( JobExecution execution ) throws Exception {
+ return delay;
+ }
+
+
+ public void setDelay( long delay ) {
+ this.delay = delay;
+ }
+
+
+ public void setLatch( int calls ) {
+ latch = new CountDownLatch( calls );
+ }
+
+
+ public boolean waitForCount( long timeout, TimeUnit unit ) throws InterruptedException {
+ return latch.await( timeout, unit );
+ }
+
+
+ public boolean waitForSleep( long timeout, TimeUnit unit ) throws InterruptedException {
+ return sleptLatch.await( timeout, unit );
+ }
+
+
+ /** @return the timeout */
+ public long getTimeout() {
+ return timeout;
+ }
+
+
+ /** @param timeout the timeout to set */
+ public void setTimeout( long timeout ) {
+ this.timeout = timeout;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/test/java/org/apache/usergrid/batch/job/OnlyOnceUnlockOnFailExceution.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/batch/job/OnlyOnceUnlockOnFailExceution.java b/stack/core/src/test/java/org/apache/usergrid/batch/job/OnlyOnceUnlockOnFailExceution.java
new file mode 100644
index 0000000..2df61ef
--- /dev/null
+++ b/stack/core/src/test/java/org/apache/usergrid/batch/job/OnlyOnceUnlockOnFailExceution.java
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.batch.job;
+
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.usergrid.batch.JobExecution;
+import org.apache.usergrid.batch.job.OnlyOnceJob;
+import org.junit.Ignore;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+
+/**
+ * A job that will sleep for the amount of time specified. Used to check that our counter is only ever run once. Checks
+ * the lock is released on fail
+ *
+ * @author tnine
+ */
+@Component("onlyOnceUnlockOnFailExceution")
+@Ignore("Not a test")
+public class OnlyOnceUnlockOnFailExceution extends OnlyOnceJob {
+
+ private static final Logger logger = LoggerFactory.getLogger( OnlyOnceUnlockOnFailExceution.class );
+
+ private CountDownLatch latch = null;
+ private CountDownLatch exception = new CountDownLatch( 1 );
+ private CountDownLatch completed = new CountDownLatch( 1 );
+ private long timeout;
+ private boolean slept = false;
+ private long delay;
+
+
+ /**
+ *
+ */
+ public OnlyOnceUnlockOnFailExceution() {
+ }
+
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see
+ * org.apache.usergrid.batch.job.OnlyOnceJob#doJob(org.apache.usergrid.batch.JobExecution)
+ */
+ @Override
+ protected void doJob( JobExecution execution ) throws Exception {
+ logger.info( "Running only once execution" );
+
+
+ latch.countDown();
+
+ if ( !slept ) {
+ logger.info( "Sleeping in only once execution" );
+ Thread.sleep( timeout );
+ slept = true;
+ exception.countDown();
+ throw new RuntimeException( "I failed to run correctly, I should be retried" );
+ }
+
+ completed.countDown();
+ }
+
+
+ /* (non-Javadoc)
+ * @see org.apache.usergrid.batch.job.OnlyOnceJob#getDelay(org.apache.usergrid.batch.JobExecution)
+ */
+ @Override
+ protected long getDelay( JobExecution execution ) throws Exception {
+ return delay;
+ }
+
+
+ public void setDelay( long delay ) {
+ this.delay = delay;
+ }
+
+
+ public void setLatch( int calls ) {
+ latch = new CountDownLatch( calls );
+ }
+
+
+ public boolean waitForCount( long timeout, TimeUnit unit ) throws InterruptedException {
+ return latch.await( timeout, unit );
+ }
+
+
+ public boolean waitForException( long timeout, TimeUnit unit ) throws InterruptedException {
+ return exception.await( timeout, unit );
+ }
+
+
+ public boolean waitForCompletion( long timeout, TimeUnit unit ) throws InterruptedException {
+ return completed.await( timeout, unit );
+ }
+
+
+ /** @return the timeout */
+ public long getTimeout() {
+ return timeout;
+ }
+
+
+ /** @param timeout the timeout to set */
+ public void setTimeout( long timeout ) {
+ this.timeout = timeout;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime1IT.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime1IT.java b/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime1IT.java
new file mode 100644
index 0000000..67fd2ba
--- /dev/null
+++ b/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime1IT.java
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.batch.job;
+
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.usergrid.cassandra.Concurrent;
+import org.apache.usergrid.persistence.entities.JobData;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Class to test job runtimes
+ */
+@Concurrent
+public class SchedulerRuntime1IT extends AbstractSchedulerRuntimeIT {
+
+ private static final Logger logger = LoggerFactory.getLogger(SchedulerRuntime1IT.class.getName());
+ @Test
+ public void basicScheduling() throws InterruptedException {
+ CountdownLatchJob counterJob = cassandraResource.getBean( CountdownLatchJob.class );
+ // set the counter job latch size
+ counterJob.setLatch( getCount() );
+
+ for ( int i = 0; i < getCount(); i++ ) {
+ scheduler.createJob( "countdownLatch", System.currentTimeMillis(), new JobData() );
+ }
+
+ // previously:
+ // now wait until everything fires or no jobs complete in #waitTime seconds
+ // boolean waited = getJobListener().blockTilDone( getCount(), waitTime);
+
+ // now:
+ // note that the waitForCount only wait for job execution. It does NOT wait for job Completion
+ boolean waited = counterJob.waitForCount(waitTime, TimeUnit.MILLISECONDS);
+ assertTrue( "Failed to run " + getCount() + " number of jobs. Waited " + waitTime + " seconds.", waited );
+
+ // previously:
+ // assertTrue( getJobListener().getSuccessCount() + " successful jobs ran, expected " + getCount(), getCount() == getJobListener().getSuccessCount() );
+
+ // now:
+ // blockTilDone look into the JobListener hook and blocked until jobs are completed.
+ // TODO : need a retry count so it doesn't reblock forever
+ while (!getJobListener().blockTilDone(getCount(), waitTime)) {
+ logger.warn("Jobs not yet finished after waited {}, block again" , waitTime);
+ }
+ assertEquals( "Expected success job: " + getCount()+ ". Actual :" + getJobListener().getSuccessCount() + ". Total count: " + getJobListener().getDoneCount() , getCount() , getJobListener().getSuccessCount() );
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime2IT.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime2IT.java b/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime2IT.java
new file mode 100644
index 0000000..656b329
--- /dev/null
+++ b/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime2IT.java
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.batch.job;
+
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.usergrid.cassandra.Concurrent;
+import org.apache.usergrid.persistence.entities.JobData;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Class to test job runtimes
+ */
+@Concurrent
+public class SchedulerRuntime2IT extends AbstractSchedulerRuntimeIT {
+ private static final Logger logger = LoggerFactory.getLogger(SchedulerRuntime2IT.class.getName());
+ /** Test the scheduler ramps up correctly when there are more jobs to be read after a pause */
+ @Test
+ public void schedulingWithNoJobs() throws InterruptedException {
+ CountdownLatchJob counterJob = cassandraResource.getBean( CountdownLatchJob.class );
+ // set the counter job latch size
+ counterJob.setLatch( getCount() );
+
+ for ( int i = 0; i < getCount(); i++ ) {
+ scheduler.createJob( "countdownLatch", System.currentTimeMillis(), new JobData() );
+ }
+
+ // previously:
+ // now wait until everything fires
+ // boolean waited = getJobListener().blockTilDone( getCount(), 15000L );
+ // assertTrue( "Jobs ran", waited );
+ // assertTrue( getCount() + " successful jobs ran", getCount() == getJobListener().getSuccessCount() );
+
+ // now:
+ // note that the waitForCount only wait for job execution. It does NOT wait for job Completion
+ boolean waited = counterJob.waitForCount(waitTime, TimeUnit.MILLISECONDS);
+ assertTrue( "Failed to run " + getCount() + " number of jobs. Waited " + waitTime + " seconds.", waited );
+
+ // now:
+ // blockTilDone look into the JobListener hook and blocked until jobs are completed.
+ // TODO : need a retry count so it doesn't reblock forever
+ while (!getJobListener().blockTilDone(getCount(), waitTime)) {
+ logger.warn("Jobs not yet finished after waited {}, block again" , waitTime);
+ }
+ assertEquals( "Expected success job: " + getCount()+ ". Actual :" + getJobListener().getSuccessCount() + ". Total count: " + getJobListener().getDoneCount() , getCount() , getJobListener().getSuccessCount() );
+
+ Thread.sleep( 5000L );
+
+ // set the counter job latch size
+ counterJob.setLatch( getCount() );
+
+ for ( int i = 0; i < getCount(); i++ ) {
+ scheduler.createJob( "countdownLatch", System.currentTimeMillis(), new JobData() );
+ }
+
+ // previously:
+ // now wait until everything fires
+ // waited = getJobListener().blockTilDone( 2 * getCount(), 15000L );
+ // waited = counterJob.waitForCount(waitTime, TimeUnit.MILLISECONDS );
+ // assertTrue( "Failed to run " + 2* getCount() + " number of jobs. Success count = " + getJobListener().getSuccessCount() + ". Waited " + waitTime + " seconds.", waited );
+ // assertTrue( 2 * getCount() + " successful jobs ran",
+ // ( 2 * getCount() ) == getJobListener().getSuccessCount() );
+
+ // now:
+ // note that the waitForCount only wait for job execution. It does NOT wait for job Completion
+ waited = counterJob.waitForCount(waitTime, TimeUnit.MILLISECONDS);
+ assertTrue( "Failed to run " + getCount() + " number of jobs. Waited " + waitTime + " seconds.", waited );
+
+ // now:
+ // blockTilDone look into the JobListener hook and blocked until jobs are completed.
+ // TODO : need a retry count so it doesn't reblock forever
+ while (!getJobListener().blockTilDone(getCount(), waitTime)) {
+ logger.warn("Jobs not yet finished after waited {}, block again" , waitTime);
+ }
+ assertEquals( "Expected success job: " +2 * getCount()+ ". Actual :" + getJobListener().getSuccessCount() + ". Total count: " + getJobListener().getDoneCount() , 2 * getCount() , getJobListener().getSuccessCount() );
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime3IT.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime3IT.java b/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime3IT.java
new file mode 100644
index 0000000..8087b21
--- /dev/null
+++ b/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime3IT.java
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.batch.job;
+
+
+import org.apache.usergrid.cassandra.Concurrent;
+import org.apache.usergrid.persistence.entities.JobData;
+import org.apache.usergrid.persistence.entities.JobStat;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+
+/**
+ * Class to test job runtimes
+ */
+@Concurrent
+public class SchedulerRuntime3IT extends AbstractSchedulerRuntimeIT {
+
+
+ /** Test the scheduler ramps up correctly when there are more jobs to be read after a pause */
+ @Test
+ public void failureCausesJobDeath() throws Exception {
+
+ int failCount = Integer.parseInt( props.getProperty( FAIL_PROP ) );
+ long sleepTime = Long.parseLong( props.getProperty( RUNLOOP_PROP ) );
+
+ FailureJobExceuction job = cassandraResource.getBean( "failureJobExceuction", FailureJobExceuction.class );
+
+ int latchValue = failCount + 1;
+
+ job.setLatch( latchValue );
+
+ JobData returned = scheduler.createJob( "failureJobExceuction", System.currentTimeMillis(), new JobData() );
+
+ // sleep until the job should have failed. We sleep 1 extra cycle just to
+ // make sure we're not racing the test
+ boolean waited = getJobListener().blockTilDone( 3, ( failCount + 2 ) * sleepTime + 5000L );
+
+ //we shouldn't trip the latch. It should fail failCount times, and not run again
+ assertTrue( "Jobs ran", waited );
+ assertTrue( failCount + " failures resulted", getJobListener().getFailureCount() == failCount );
+ assertTrue( 1 + " success resulted", getJobListener().getSuccessCount() == 1 );
+
+ //we shouldn't have run the last time, we should have counted down to it
+ assertEquals( 1, job.getLatchCount() );
+
+ JobStat stat = scheduler.getStatsForJob( returned.getJobName(), returned.getUuid() );
+
+ // we should have only marked this as run fail+1 times
+ assertEquals( latchValue, stat.getTotalAttempts() );
+ assertEquals( latchValue, stat.getRunCount() );
+ assertEquals( 0, stat.getDelayCount() );
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime4IT.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime4IT.java b/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime4IT.java
new file mode 100644
index 0000000..8ff75f6
--- /dev/null
+++ b/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime4IT.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.batch.job;
+
+
+import org.apache.usergrid.cassandra.Concurrent;
+import org.apache.usergrid.persistence.entities.JobData;
+import org.apache.usergrid.persistence.entities.JobStat;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+
+/**
+ * Class to test job runtimes
+ */
+@Concurrent
+public class SchedulerRuntime4IT extends AbstractSchedulerRuntimeIT {
+ /**
+ * Test the scheduler ramps up correctly when there are more jobs to be read after a pause when the job specifies
+ * the retry time
+ */
+ @Test
+ public void delayExecution() throws Exception {
+ long sleepTime = Long.parseLong( props.getProperty( TIMEOUT_PROP ) );
+
+ int delayCount = 2;
+
+ long customRetry = sleepTime * 2;
+
+ DelayExecution job = cassandraResource.getBean( "delayExecution", DelayExecution.class );
+
+ job.setTimeout( customRetry );
+ job.setLatch( delayCount + 1 );
+
+ JobData returned = scheduler.createJob( "delayExecution", System.currentTimeMillis(), new JobData() );
+
+ // sleep until the job should have failed. We sleep 1 extra cycle just to
+ // make sure we're not racing the test
+ boolean waited = getJobListener().blockTilDone( 3, 5000L + sleepTime * 2 );
+
+ assertTrue( "Job ran to complete", waited );
+
+ JobStat stat = scheduler.getStatsForJob( returned.getJobName(), returned.getUuid() );
+
+ // we should have only marked this as run once since we delayed furthur execution
+ // we should have only marked this as run once
+ assertEquals( 1, stat.getTotalAttempts() );
+ assertEquals( delayCount + 1, stat.getRunCount() );
+ assertEquals( delayCount, stat.getDelayCount() );
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime5IT.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime5IT.java b/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime5IT.java
new file mode 100644
index 0000000..35fbd01
--- /dev/null
+++ b/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime5IT.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.batch.job;
+
+
+import org.apache.usergrid.cassandra.Concurrent;
+import org.apache.usergrid.persistence.entities.JobData;
+import org.apache.usergrid.persistence.entities.JobStat;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+
+/**
+ * Class to test job runtimes
+ */
+@Concurrent
+public class SchedulerRuntime5IT extends AbstractSchedulerRuntimeIT {
+ /**
+ * Test the scheduler ramps up correctly when there are more jobs to be read after a pause when the job specifies
+ * the retry time
+ */
+ @Test
+ public void delayHeartbeat() throws Exception {
+
+ long sleepTime = Long.parseLong( props.getProperty( TIMEOUT_PROP ) );
+
+ int heartbeatCount = 2;
+
+ long customRetry = sleepTime * 2;
+
+ DelayHeartbeat job = cassandraResource.getBean( "delayHeartbeat", DelayHeartbeat.class );
+
+ job.setTimeout( customRetry );
+ job.setLatch( heartbeatCount + 1 );
+
+ JobData returned = scheduler.createJob( "delayHeartbeat", System.currentTimeMillis(), new JobData() );
+
+ // sleep until the job should have failed. We sleep 1 extra cycle just to
+ // make sure we're not racing the test
+ boolean waited = getJobListener().blockTilDone( 1, customRetry * ( heartbeatCount * 2 ) + 5000L );
+
+ assertTrue( "Job ran to complete", waited );
+
+ JobStat stat = scheduler.getStatsForJob( returned.getJobName(), returned.getUuid() );
+
+ // we should have only marked this as run once since we delayed further execution
+ // we should have only marked this as run once
+ assertEquals( 1, stat.getTotalAttempts() );
+ assertEquals( 1, stat.getRunCount() );
+ assertEquals( 0, stat.getDelayCount() );
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime6IT.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime6IT.java b/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime6IT.java
new file mode 100644
index 0000000..6f05b58
--- /dev/null
+++ b/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime6IT.java
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.batch.job;
+
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.usergrid.cassandra.Concurrent;
+import org.apache.usergrid.persistence.entities.JobData;
+import org.apache.usergrid.persistence.entities.JobStat;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+
+/**
+ * Class to test job runtimes
+ */
+@Concurrent
+public class SchedulerRuntime6IT extends AbstractSchedulerRuntimeIT {
+ /**
+ * Test the scheduler ramps up correctly when there are more jobs to be read after a pause when the job specifies
+ * the retry time
+ */
+ @Test
+ public void onlyOnceTest() throws Exception {
+
+ long sleepTime = Long.parseLong( props.getProperty( TIMEOUT_PROP ) );
+
+ long customRetry = sleepTime + 1000;
+ int numberOfRuns = 1;
+
+ OnlyOnceExceution job = cassandraResource.getBean( "onlyOnceExceution", OnlyOnceExceution.class );
+
+ job.setTimeout( customRetry );
+ job.setLatch( numberOfRuns );
+ job.setDelay( sleepTime );
+
+ JobData returned = scheduler.createJob( "onlyOnceExceution", System.currentTimeMillis(), new JobData() );
+
+ // sleep until the job should have failed. We sleep 1 extra cycle just to
+ // make sure we're not racing the test
+ boolean waited = getJobListener().blockTilDone( 1, customRetry * numberOfRuns * 2 + 5000L );
+
+ assertTrue( "Job ran twice", waited );
+
+
+ //reset our latch immediately for further tests
+ job.setLatch( numberOfRuns );
+
+ JobStat stat = scheduler.getStatsForJob( returned.getJobName(), returned.getUuid() );
+
+ // we should have only marked this as run once since we delayed furthur execution
+ // we should have only marked this as run once
+ assertNotNull( stat );
+ assertEquals( numberOfRuns, stat.getTotalAttempts() );
+ assertEquals( numberOfRuns, stat.getRunCount() );
+ assertEquals( 0, stat.getDelayCount() );
+
+
+ boolean slept = job.waitForSleep( customRetry * numberOfRuns * 2, TimeUnit.MILLISECONDS );
+
+ assertTrue( "Job slept", slept );
+
+
+ //now wait again to see if the job fires one more time, it shouldn't
+ waited = getJobListener().blockTilDone( 2, customRetry * numberOfRuns * 2 );
+
+ assertFalse( "Job ran twice", waited );
+
+ stat = scheduler.getStatsForJob( returned.getJobName(), returned.getUuid() );
+
+ // we should have only marked this as run once since we delayed further execution
+ // we should have only marked this as run once
+ assertEquals( numberOfRuns, stat.getTotalAttempts() );
+ assertEquals( numberOfRuns, stat.getRunCount() );
+ assertEquals( 0, stat.getDelayCount() );
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime7IT.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime7IT.java b/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime7IT.java
new file mode 100644
index 0000000..ea75a2d
--- /dev/null
+++ b/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime7IT.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.batch.job;
+
+
+import org.apache.usergrid.cassandra.Concurrent;
+import org.apache.usergrid.persistence.entities.JobData;
+import org.apache.usergrid.persistence.entities.JobStat;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+
+/**
+ * Class to test job runtimes
+ */
+@Concurrent
+@org.junit.Ignore( "Todd you need to take a look at this since it's not clear to me what was intended in this test." )
+public class SchedulerRuntime7IT extends AbstractSchedulerRuntimeIT {
+
+ /** Test that we're only running once, even when a job exceeds the heartbeat time */
+ @Test
+ public void onlyOnceTestOnException() throws Exception {
+
+ long sleepTime = Long.parseLong( props.getProperty( TIMEOUT_PROP ) );
+
+ long runLoop = Long.parseLong( props.getProperty( RUNLOOP_PROP ) );
+
+ long customRetry = sleepTime * 2;
+ int numberOfRuns = 2;
+
+ OnlyOnceUnlockOnFailExceution job =
+ cassandraResource.getBean( "onlyOnceUnlockOnFailExceution", OnlyOnceUnlockOnFailExceution.class );
+
+ job.setTimeout( customRetry );
+ job.setLatch( numberOfRuns );
+ job.setDelay( sleepTime );
+
+ JobData returned =
+ scheduler.createJob( "onlyOnceUnlockOnFailExceution", System.currentTimeMillis(), new JobData() );
+
+ // sleep until the job should have failed. We sleep 1 extra cycle just to make sure we're not racing the test
+
+ boolean waited = getJobListener().blockTilDone( 1, runLoop * numberOfRuns * 2 + 5000L );
+
+ assertTrue( "Job threw exception", waited );
+ assertTrue( "Job failed", getJobListener().getFailureCount() == 1 );
+ assertTrue( "No Job succeeded", getJobListener().getSuccessCount() == 0 );
+
+ JobStat stat = scheduler.getStatsForJob( returned.getJobName(), returned.getUuid() );
+
+ // we should have only marked this as run once since we delayed furthur execution
+ // we should have only marked this as run once
+ assertEquals( numberOfRuns, stat.getTotalAttempts() );
+ assertEquals( numberOfRuns, stat.getRunCount() );
+ assertEquals( 0, stat.getDelayCount() );
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime8IT.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime8IT.java b/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime8IT.java
new file mode 100644
index 0000000..3a220da
--- /dev/null
+++ b/stack/core/src/test/java/org/apache/usergrid/batch/job/SchedulerRuntime8IT.java
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.batch.job;
+
+
+import java.util.UUID;
+
+import org.apache.usergrid.cassandra.Concurrent;
+import org.apache.usergrid.persistence.Query;
+import org.apache.usergrid.persistence.Results;
+import org.apache.usergrid.persistence.entities.JobData;
+import org.apache.usergrid.utils.UUIDUtils;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+
+
+/**
+ * Class to test job runtimes
+ */
+@Concurrent
+public class SchedulerRuntime8IT extends AbstractSchedulerRuntimeIT {
+ /**
+ * Test the scheduler ramps up correctly when there are more jobs to be read after a pause when the job specifies
+ * the retry time
+ */
+ @Test
+ public void queryAndDeleteJobs() throws Exception {
+
+ CountdownLatchJob job = cassandraResource.getBean( "countdownLatch", CountdownLatchJob.class );
+
+ job.setLatch( 1 );
+
+ // fire the job 30 seconds from now
+ long fireTime = System.currentTimeMillis() + 30000;
+
+ UUID notificationId = UUIDUtils.newTimeUUID();
+
+ JobData test = new JobData();
+ test.setProperty( "stringprop", "test" );
+ test.setProperty( "notificationId", notificationId );
+
+ JobData saved = scheduler.createJob( "countdownLatch", fireTime, test );
+
+ // now query and make sure it equals the saved value
+
+ Query query = new Query();
+ query.addEqualityFilter( "notificationId", notificationId );
+
+ Results r = scheduler.queryJobData( query );
+
+ assertEquals( 1, r.size() );
+
+ assertEquals( saved.getUuid(), r.getEntity().getUuid() );
+
+ // query by uuid
+ query = new Query();
+ query.addEqualityFilter( "stringprop", "test" );
+
+ r = scheduler.queryJobData( query );
+
+ assertEquals( 1, r.size() );
+
+ assertEquals( saved.getUuid(), r.getEntity().getUuid() );
+
+ // now delete the job
+
+ scheduler.deleteJob( saved.getUuid() );
+
+ // sleep until the job should have failed. We sleep 1 extra cycle just to
+ // make sure we're not racing the test
+
+ long waitTime = Math.max( 0, fireTime - System.currentTimeMillis() + 1000 );
+
+ boolean waited = getJobListener().blockTilDone( 1, waitTime );
+
+ assertFalse( "Job ran ", waited );
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/test/java/org/apache/usergrid/batch/job/TestJobListener.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/batch/job/TestJobListener.java b/stack/core/src/test/java/org/apache/usergrid/batch/job/TestJobListener.java
new file mode 100644
index 0000000..8e1144b
--- /dev/null
+++ b/stack/core/src/test/java/org/apache/usergrid/batch/job/TestJobListener.java
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.usergrid.batch.job;
+
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.usergrid.batch.JobExecution;
+import org.apache.usergrid.batch.service.JobListener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Implementation of the JobListener for tests.
+ */
+public class TestJobListener implements JobListener {
+ // public static final long WAIT_MAX_MILLIS = 250;
+ public static final long WAIT_MAX_MILLIS = 60000L; // max wait 1 minutes
+ private static final Logger LOG = LoggerFactory.getLogger( TestJobListener.class );
+ private AtomicInteger submittedCounter = new AtomicInteger();
+ private AtomicInteger failureCounter = new AtomicInteger();
+ private AtomicInteger successCounter = new AtomicInteger();
+ private final Object lock = new Object();
+
+
+ public int getSubmittedCount() {
+ return submittedCounter.get();
+ }
+
+
+ public int getFailureCount() {
+ return failureCounter.get();
+ }
+
+
+ public int getSuccessCount() {
+ return successCounter.get();
+ }
+
+
+ public int getDoneCount() {
+ return successCounter.get() + failureCounter.get();
+ }
+
+
+ public void onSubmit( JobExecution execution ) {
+ LOG.debug( "Job execution {} submitted with count {}.", execution,
+ submittedCounter.incrementAndGet() );
+ }
+
+
+ public void onSuccess( JobExecution execution ) {
+ LOG.debug( "Job execution {} succeeded with count {}.", execution,
+ successCounter.incrementAndGet() );
+ }
+
+
+ public void onFailure( JobExecution execution ) {
+ LOG.debug( "Job execution {} failed with count {}.", execution,
+ failureCounter.incrementAndGet() );
+ }
+
+
+
+ /**
+ * block until submitted jobs are all accounted for.
+ *
+ * @param jobCount total submitted job
+ * @param idleTime idleTime in millisecond.
+ * @return true when all jobs are completed (could be succeed or failed) within the given idleTime range, or {@value #WAIT_MAX_MILLIS}, whichever is smaller
+ * @throws InterruptedException
+ */
+ public boolean blockTilDone( int jobCount, long idleTime ) throws InterruptedException
+ {
+ final long waitTime = Math.min( idleTime, WAIT_MAX_MILLIS );
+ long lastChangeTime = System.currentTimeMillis();
+ long timeNotChanged = 0;
+ int currentCount;
+ int startCount = getDoneCount();
+
+ do {
+ currentCount = getDoneCount();
+
+ if ( startCount == currentCount ) {
+// if ( timeNotChanged > idleTime ) {
+ if ( timeNotChanged > waitTime ) {
+ return false;
+ }
+
+ timeNotChanged = System.currentTimeMillis() - lastChangeTime;
+ }
+ else {
+ timeNotChanged = 0;
+ startCount = currentCount;
+ lastChangeTime = System.currentTimeMillis();
+ }
+
+ synchronized ( lock ) {
+ lock.wait( waitTime );
+ }
+ } while ( currentCount < jobCount );
+
+ return true;
+ }
+}