LocalTaskExecutor { public void execute() { createElasticJobExecutor(new CloudJobFacade(getShardingContexts(), jobConfiguration, new JobEventBus())).execute(); } LocalTaskExecutor(final ElasticJob elasticJob, final JobConfiguration jobConfiguration, final int shardingItem);  LocalTaskExecutor(final String elasticJobType, final JobConfiguration jobConfiguration, final int shardingItem); void execute();  }
@Test public void assertSimpleJob() { TestSimpleJob simpleJob = new TestSimpleJob(); new LocalTaskExecutor(simpleJob, JobConfiguration.newBuilder(TestSimpleJob.class.getSimpleName(), 3) .cron("*/2 * * * * ?").shardingItemParameters("0=A,1=B").build(), 1).execute(); assertThat(simpleJob.getShardingParameters(), is(Collections.singletonList("B"))); }
@Test public void assertDataflowJob() { TestDataflowJob dataflowJob = new TestDataflowJob(); new LocalTaskExecutor(dataflowJob, JobConfiguration.newBuilder(TestDataflowJob.class.getSimpleName(), 3) .cron("*/2 * * * * ?").setProperty(DataflowJobProperties.STREAM_PROCESS_KEY, Boolean.FALSE.toString()).build(), 1).execute(); assertFalse(dataflowJob.getOutput().isEmpty()); assertThat(dataflowJob.getOutput(), is(Collections.singletonList("1-d"))); }
@Test public void assertScriptJob() { new LocalTaskExecutor(new TestDataflowJob(), JobConfiguration.newBuilder("TestScriptJob", 3) .cron("*/2 * * * * ?").setProperty(ScriptJobProperties.SCRIPT_KEY, "echo test").build(), 1).execute(); }
@Test(expected = JobConfigurationException.class) public void assertNotExistsJobType() { new LocalTaskExecutor("not exist", JobConfiguration.newBuilder("not exist", 3).cron("*/2 * * * * ?").build(), 1).execute(); }