You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by zh...@apache.org on 2020/10/19 13:45:27 UTC
[shardingsphere-elasticjob] branch master updated: Rename
JobTracingEventBus (#1612)
This is an automated email from the ASF dual-hosted git repository.
zhangliang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere-elasticjob.git
The following commit(s) were added to refs/heads/master by this push:
new 727b1cb Rename JobTracingEventBus (#1612)
727b1cb is described below
commit 727b1cbf3b2f53f0193b18701e00d9b5611d122c
Author: Liang Zhang <te...@163.com>
AuthorDate: Mon Oct 19 21:45:14 2020 +0800
Rename JobTracingEventBus (#1612)
* For code format
* Rename JobTracingEventBus
---
.../cloud/executor/local/LocalTaskExecutor.java | 4 ++--
.../cloud/executor/prod/TaskExecutor.java | 8 ++++----
.../elasticjob/cloud/facade/CloudJobFacade.java | 8 ++++----
.../cloud/executor/facade/CloudJobFacadeTest.java | 8 ++++----
.../console/controller/CloudJobController.java | 2 +-
.../cloud/scheduler/env/BootstrapEnvironment.java | 2 +-
.../cloud/scheduler/mesos/SchedulerEngine.java | 6 +++---
.../cloud/scheduler/mesos/SchedulerService.java | 21 +++++++++++----------
.../mesos/TaskLaunchScheduledService.java | 6 +++---
.../scheduler/env/BootstrapEnvironmentTest.java | 2 +-
.../cloud/scheduler/mesos/SchedulerEngineTest.java | 4 ++--
.../mesos/TaskLaunchScheduledServiceTest.java | 10 +++++-----
.../lite/internal/schedule/JobScheduler.java | 4 ++--
.../lite/internal/schedule/LiteJobFacade.java | 12 ++++++------
.../lite/internal/schedule/LiteJobFacadeTest.java | 8 ++++----
.../{JobEventBus.java => JobTracingEventBus.java} | 10 +++++-----
...entBusTest.java => JobTracingEventBusTest.java} | 22 +++++++++++-----------
.../src/test/resources/logback-test.xml | 2 +-
.../rdb/listener/RDBTracingListenerTest.java | 10 +++++-----
19 files changed, 75 insertions(+), 74 deletions(-)
diff --git a/elasticjob-cloud/elasticjob-cloud-executor/src/main/java/org/apache/shardingsphere/elasticjob/cloud/executor/local/LocalTaskExecutor.java b/elasticjob-cloud/elasticjob-cloud-executor/src/main/java/org/apache/shardingsphere/elasticjob/cloud/executor/local/LocalTaskExecutor.java
index 9bb9559..fee6878 100755
--- a/elasticjob-cloud/elasticjob-cloud-executor/src/main/java/org/apache/shardingsphere/elasticjob/cloud/executor/local/LocalTaskExecutor.java
+++ b/elasticjob-cloud/elasticjob-cloud-executor/src/main/java/org/apache/shardingsphere/elasticjob/cloud/executor/local/LocalTaskExecutor.java
@@ -26,7 +26,7 @@ import org.apache.shardingsphere.elasticjob.executor.ElasticJobExecutor;
import org.apache.shardingsphere.elasticjob.executor.JobFacade;
import org.apache.shardingsphere.elasticjob.infra.context.ShardingItemParameters;
import org.apache.shardingsphere.elasticjob.infra.listener.ShardingContexts;
-import org.apache.shardingsphere.elasticjob.tracing.JobEventBus;
+import org.apache.shardingsphere.elasticjob.tracing.JobTracingEventBus;
import java.util.HashMap;
import java.util.Map;
@@ -57,7 +57,7 @@ public final class LocalTaskExecutor {
* Execute job.
*/
public void execute() {
- createElasticJobExecutor(new CloudJobFacade(getShardingContexts(), jobConfiguration, new JobEventBus())).execute();
+ createElasticJobExecutor(new CloudJobFacade(getShardingContexts(), jobConfiguration, new JobTracingEventBus())).execute();
}
private ElasticJobExecutor createElasticJobExecutor(final JobFacade jobFacade) {
diff --git a/elasticjob-cloud/elasticjob-cloud-executor/src/main/java/org/apache/shardingsphere/elasticjob/cloud/executor/prod/TaskExecutor.java b/elasticjob-cloud/elasticjob-cloud-executor/src/main/java/org/apache/shardingsphere/elasticjob/cloud/executor/prod/TaskExecutor.java
index 377fa37..9bbf575 100755
--- a/elasticjob-cloud/elasticjob-cloud-executor/src/main/java/org/apache/shardingsphere/elasticjob/cloud/executor/prod/TaskExecutor.java
+++ b/elasticjob-cloud/elasticjob-cloud-executor/src/main/java/org/apache/shardingsphere/elasticjob/cloud/executor/prod/TaskExecutor.java
@@ -37,7 +37,7 @@ import org.apache.shardingsphere.elasticjob.infra.exception.ExceptionUtils;
import org.apache.shardingsphere.elasticjob.infra.listener.ShardingContexts;
import org.apache.shardingsphere.elasticjob.infra.pojo.JobConfigurationPOJO;
import org.apache.shardingsphere.elasticjob.infra.yaml.YamlEngine;
-import org.apache.shardingsphere.elasticjob.tracing.JobEventBus;
+import org.apache.shardingsphere.elasticjob.tracing.JobTracingEventBus;
import org.apache.shardingsphere.elasticjob.tracing.api.TracingConfiguration;
import javax.sql.DataSource;
@@ -59,7 +59,7 @@ public final class TaskExecutor implements Executor {
private volatile ElasticJobExecutor jobExecutor;
- private volatile JobEventBus jobEventBus = new JobEventBus();
+ private volatile JobTracingEventBus jobTracingEventBus = new JobTracingEventBus();
public TaskExecutor(final ElasticJob elasticJob) {
this(elasticJob, null);
@@ -78,7 +78,7 @@ public final class TaskExecutor implements Executor {
dataSource.setUrl(data.get("event_trace_rdb_url"));
dataSource.setPassword(data.get("event_trace_rdb_password"));
dataSource.setUsername(data.get("event_trace_rdb_username"));
- jobEventBus = new JobEventBus(new TracingConfiguration<DataSource>("RDB", dataSource));
+ jobTracingEventBus = new JobTracingEventBus(new TracingConfiguration<DataSource>("RDB", dataSource));
}
}
@@ -133,7 +133,7 @@ public final class TaskExecutor implements Executor {
ShardingContexts shardingContexts = (ShardingContexts) data.get("shardingContext");
JobConfiguration jobConfig = YamlEngine.unmarshal(data.get("jobConfigContext").toString(), JobConfigurationPOJO.class).toJobConfiguration();
try {
- JobFacade jobFacade = new CloudJobFacade(shardingContexts, jobConfig, jobEventBus);
+ JobFacade jobFacade = new CloudJobFacade(shardingContexts, jobConfig, jobTracingEventBus);
if (isTransient(jobConfig)) {
getJobExecutor(jobFacade).execute();
executorDriver.sendStatusUpdate(Protos.TaskStatus.newBuilder().setTaskId(taskInfo.getTaskId()).setState(Protos.TaskState.TASK_FINISHED).build());
diff --git a/elasticjob-cloud/elasticjob-cloud-executor/src/main/java/org/apache/shardingsphere/elasticjob/cloud/facade/CloudJobFacade.java b/elasticjob-cloud/elasticjob-cloud-executor/src/main/java/org/apache/shardingsphere/elasticjob/cloud/facade/CloudJobFacade.java
index 383e5be..e1f324c 100755
--- a/elasticjob-cloud/elasticjob-cloud-executor/src/main/java/org/apache/shardingsphere/elasticjob/cloud/facade/CloudJobFacade.java
+++ b/elasticjob-cloud/elasticjob-cloud-executor/src/main/java/org/apache/shardingsphere/elasticjob/cloud/facade/CloudJobFacade.java
@@ -22,7 +22,7 @@ import org.apache.shardingsphere.elasticjob.api.JobConfiguration;
import org.apache.shardingsphere.elasticjob.executor.JobFacade;
import org.apache.shardingsphere.elasticjob.infra.context.TaskContext;
import org.apache.shardingsphere.elasticjob.infra.listener.ShardingContexts;
-import org.apache.shardingsphere.elasticjob.tracing.JobEventBus;
+import org.apache.shardingsphere.elasticjob.tracing.JobTracingEventBus;
import org.apache.shardingsphere.elasticjob.tracing.event.JobExecutionEvent;
import org.apache.shardingsphere.elasticjob.tracing.event.JobStatusTraceEvent;
import org.apache.shardingsphere.elasticjob.tracing.event.JobStatusTraceEvent.Source;
@@ -40,7 +40,7 @@ public final class CloudJobFacade implements JobFacade {
private final JobConfiguration jobConfig;
- private final JobEventBus jobEventBus;
+ private final JobTracingEventBus jobTracingEventBus;
@Override
public JobConfiguration loadJobConfiguration(final boolean fromCache) {
@@ -103,13 +103,13 @@ public final class CloudJobFacade implements JobFacade {
@Override
public void postJobExecutionEvent(final JobExecutionEvent jobExecutionEvent) {
- jobEventBus.post(jobExecutionEvent);
+ jobTracingEventBus.post(jobExecutionEvent);
}
@Override
public void postJobStatusTraceEvent(final String taskId, final State state, final String message) {
TaskContext taskContext = TaskContext.from(taskId);
- jobEventBus.post(new JobStatusTraceEvent(taskContext.getMetaInfo().getJobName(), taskContext.getId(), taskContext.getSlaveId(),
+ jobTracingEventBus.post(new JobStatusTraceEvent(taskContext.getMetaInfo().getJobName(), taskContext.getId(), taskContext.getSlaveId(),
Source.CLOUD_EXECUTOR, taskContext.getType().toString(), String.valueOf(taskContext.getMetaInfo().getShardingItems()), state, message));
}
}
diff --git a/elasticjob-cloud/elasticjob-cloud-executor/src/test/java/org/apache/shardingsphere/elasticjob/cloud/executor/facade/CloudJobFacadeTest.java b/elasticjob-cloud/elasticjob-cloud-executor/src/test/java/org/apache/shardingsphere/elasticjob/cloud/executor/facade/CloudJobFacadeTest.java
index d958c86..475ac79 100755
--- a/elasticjob-cloud/elasticjob-cloud-executor/src/test/java/org/apache/shardingsphere/elasticjob/cloud/executor/facade/CloudJobFacadeTest.java
+++ b/elasticjob-cloud/elasticjob-cloud-executor/src/test/java/org/apache/shardingsphere/elasticjob/cloud/executor/facade/CloudJobFacadeTest.java
@@ -24,7 +24,7 @@ import org.apache.shardingsphere.elasticjob.executor.JobFacade;
import org.apache.shardingsphere.elasticjob.infra.context.ExecutionType;
import org.apache.shardingsphere.elasticjob.infra.exception.JobExecutionEnvironmentException;
import org.apache.shardingsphere.elasticjob.infra.listener.ShardingContexts;
-import org.apache.shardingsphere.elasticjob.tracing.JobEventBus;
+import org.apache.shardingsphere.elasticjob.tracing.JobTracingEventBus;
import org.apache.shardingsphere.elasticjob.tracing.event.JobExecutionEvent;
import org.apache.shardingsphere.elasticjob.tracing.event.JobStatusTraceEvent.State;
import org.junit.Before;
@@ -47,14 +47,14 @@ public final class CloudJobFacadeTest {
private ShardingContexts shardingContexts;
@Mock
- private JobEventBus eventBus;
+ private JobTracingEventBus jobTracingEventBus;
private JobFacade jobFacade;
@Before
public void setUp() {
shardingContexts = getShardingContexts();
- jobFacade = new CloudJobFacade(shardingContexts, getJobConfiguration(), eventBus);
+ jobFacade = new CloudJobFacade(shardingContexts, getJobConfiguration(), jobTracingEventBus);
}
private ShardingContexts getShardingContexts() {
@@ -126,7 +126,7 @@ public final class CloudJobFacadeTest {
public void assertPostJobExecutionEvent() {
JobExecutionEvent jobExecutionEvent = new JobExecutionEvent("localhost", "127.0.0.1", "fake_task_id", "test_job", JobExecutionEvent.ExecutionSource.NORMAL_TRIGGER, 0);
jobFacade.postJobExecutionEvent(jobExecutionEvent);
- verify(eventBus).post(jobExecutionEvent);
+ verify(jobTracingEventBus).post(jobExecutionEvent);
}
@Test
diff --git a/elasticjob-cloud/elasticjob-cloud-scheduler/src/main/java/org/apache/shardingsphere/elasticjob/cloud/console/controller/CloudJobController.java b/elasticjob-cloud/elasticjob-cloud-scheduler/src/main/java/org/apache/shardingsphere/elasticjob/cloud/console/controller/CloudJobController.java
index f5919aa..d5548f9 100755
--- a/elasticjob-cloud/elasticjob-cloud-scheduler/src/main/java/org/apache/shardingsphere/elasticjob/cloud/console/controller/CloudJobController.java
+++ b/elasticjob-cloud/elasticjob-cloud-scheduler/src/main/java/org/apache/shardingsphere/elasticjob/cloud/console/controller/CloudJobController.java
@@ -99,7 +99,7 @@ public final class CloudJobController implements RestfulController {
public static void init(final CoordinatorRegistryCenter regCenter, final ProducerManager producerManager) {
CloudJobController.regCenter = regCenter;
CloudJobController.producerManager = producerManager;
- Optional<TracingConfiguration> tracingConfiguration = BootstrapEnvironment.getINSTANCE().getTracingConfiguration();
+ Optional<TracingConfiguration<?>> tracingConfiguration = BootstrapEnvironment.getINSTANCE().getTracingConfiguration();
jobEventRdbSearch = tracingConfiguration.map(tracingConfiguration1 -> new JobEventRdbSearch((DataSource) tracingConfiguration1.getStorage())).orElse(null);
}
diff --git a/elasticjob-cloud/elasticjob-cloud-scheduler/src/main/java/org/apache/shardingsphere/elasticjob/cloud/scheduler/env/BootstrapEnvironment.java b/elasticjob-cloud/elasticjob-cloud-scheduler/src/main/java/org/apache/shardingsphere/elasticjob/cloud/scheduler/env/BootstrapEnvironment.java
index 2ecd8b3..8acd829 100755
--- a/elasticjob-cloud/elasticjob-cloud-scheduler/src/main/java/org/apache/shardingsphere/elasticjob/cloud/scheduler/env/BootstrapEnvironment.java
+++ b/elasticjob-cloud/elasticjob-cloud-scheduler/src/main/java/org/apache/shardingsphere/elasticjob/cloud/scheduler/env/BootstrapEnvironment.java
@@ -127,7 +127,7 @@ public final class BootstrapEnvironment {
*
* @return tracing configuration
*/
- public Optional<TracingConfiguration> getTracingConfiguration() {
+ public Optional<TracingConfiguration<?>> getTracingConfiguration() {
String driver = getValue(EnvironmentArgument.EVENT_TRACE_RDB_DRIVER);
String url = getValue(EnvironmentArgument.EVENT_TRACE_RDB_URL);
String username = getValue(EnvironmentArgument.EVENT_TRACE_RDB_USERNAME);
diff --git a/elasticjob-cloud/elasticjob-cloud-scheduler/src/main/java/org/apache/shardingsphere/elasticjob/cloud/scheduler/mesos/SchedulerEngine.java b/elasticjob-cloud/elasticjob-cloud-scheduler/src/main/java/org/apache/shardingsphere/elasticjob/cloud/scheduler/mesos/SchedulerEngine.java
index 3d4e600..3332a5f 100755
--- a/elasticjob-cloud/elasticjob-cloud-scheduler/src/main/java/org/apache/shardingsphere/elasticjob/cloud/scheduler/mesos/SchedulerEngine.java
+++ b/elasticjob-cloud/elasticjob-cloud-scheduler/src/main/java/org/apache/shardingsphere/elasticjob/cloud/scheduler/mesos/SchedulerEngine.java
@@ -26,7 +26,7 @@ import org.apache.mesos.SchedulerDriver;
import org.apache.shardingsphere.elasticjob.infra.context.TaskContext;
import org.apache.shardingsphere.elasticjob.cloud.scheduler.ha.FrameworkIDService;
import org.apache.shardingsphere.elasticjob.cloud.scheduler.statistics.StatisticManager;
-import org.apache.shardingsphere.elasticjob.tracing.JobEventBus;
+import org.apache.shardingsphere.elasticjob.tracing.JobTracingEventBus;
import org.apache.shardingsphere.elasticjob.tracing.event.JobStatusTraceEvent;
import java.util.List;
@@ -42,7 +42,7 @@ public final class SchedulerEngine implements Scheduler {
private final FacadeService facadeService;
- private final JobEventBus jobEventBus;
+ private final JobTracingEventBus jobTracingEventBus;
private final FrameworkIDService frameworkIDService;
@@ -83,7 +83,7 @@ public final class SchedulerEngine implements Scheduler {
TaskContext taskContext = TaskContext.from(taskId);
String jobName = taskContext.getMetaInfo().getJobName();
log.trace("call statusUpdate task state is: {}, task id is: {}", taskStatus.getState(), taskId);
- jobEventBus.post(new JobStatusTraceEvent(jobName, taskContext.getId(), taskContext.getSlaveId(), JobStatusTraceEvent.Source.CLOUD_SCHEDULER, taskContext.getType().toString(),
+ jobTracingEventBus.post(new JobStatusTraceEvent(jobName, taskContext.getId(), taskContext.getSlaveId(), JobStatusTraceEvent.Source.CLOUD_SCHEDULER, taskContext.getType().toString(),
String.valueOf(taskContext.getMetaInfo().getShardingItems()), JobStatusTraceEvent.State.valueOf(taskStatus.getState().name()), taskStatus.getMessage()));
switch (taskStatus.getState()) {
case TASK_RUNNING:
diff --git a/elasticjob-cloud/elasticjob-cloud-scheduler/src/main/java/org/apache/shardingsphere/elasticjob/cloud/scheduler/mesos/SchedulerService.java b/elasticjob-cloud/elasticjob-cloud-scheduler/src/main/java/org/apache/shardingsphere/elasticjob/cloud/scheduler/mesos/SchedulerService.java
index aed52ad..c12a22b 100755
--- a/elasticjob-cloud/elasticjob-cloud-scheduler/src/main/java/org/apache/shardingsphere/elasticjob/cloud/scheduler/mesos/SchedulerService.java
+++ b/elasticjob-cloud/elasticjob-cloud-scheduler/src/main/java/org/apache/shardingsphere/elasticjob/cloud/scheduler/mesos/SchedulerService.java
@@ -19,7 +19,6 @@ package org.apache.shardingsphere.elasticjob.cloud.scheduler.mesos;
import com.google.common.util.concurrent.Service;
import com.netflix.fenzo.TaskScheduler;
-import java.util.Optional;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.mesos.MesosSchedulerDriver;
@@ -36,9 +35,11 @@ import org.apache.shardingsphere.elasticjob.cloud.scheduler.state.disable.app.Cl
import org.apache.shardingsphere.elasticjob.cloud.scheduler.state.disable.job.CloudJobDisableListener;
import org.apache.shardingsphere.elasticjob.cloud.scheduler.statistics.StatisticManager;
import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
-import org.apache.shardingsphere.elasticjob.tracing.JobEventBus;
+import org.apache.shardingsphere.elasticjob.tracing.JobTracingEventBus;
import org.apache.shardingsphere.elasticjob.tracing.api.TracingConfiguration;
+import java.util.Optional;
+
/**
* Scheduler service.
*/
@@ -77,19 +78,19 @@ public final class SchedulerService {
facadeService = new FacadeService(regCenter);
statisticManager = StatisticManager.getInstance(regCenter, env.getTracingConfiguration().orElse(null));
TaskScheduler taskScheduler = getTaskScheduler();
- JobEventBus jobEventBus = getJobEventBus();
- schedulerDriver = getSchedulerDriver(taskScheduler, jobEventBus, new FrameworkIDService(regCenter));
+ JobTracingEventBus jobTracingEventBus = getJobTracingEventBus();
+ schedulerDriver = getSchedulerDriver(taskScheduler, jobTracingEventBus, new FrameworkIDService(regCenter));
producerManager = new ProducerManager(schedulerDriver, regCenter);
cloudJobConfigurationListener = new CloudJobConfigurationListener(regCenter, producerManager);
cloudJobDisableListener = new CloudJobDisableListener(regCenter, producerManager);
cloudAppConfigurationListener = new CloudAppConfigurationListener(regCenter, producerManager);
cloudAppDisableListener = new CloudAppDisableListener(regCenter, producerManager);
- taskLaunchScheduledService = new TaskLaunchScheduledService(schedulerDriver, taskScheduler, facadeService, jobEventBus);
+ taskLaunchScheduledService = new TaskLaunchScheduledService(schedulerDriver, taskScheduler, facadeService, jobTracingEventBus);
reconcileService = new ReconcileService(schedulerDriver, facadeService);
consoleBootstrap = new ConsoleBootstrap(regCenter, env.getRestfulServerConfiguration(), producerManager, reconcileService);
}
- private SchedulerDriver getSchedulerDriver(final TaskScheduler taskScheduler, final JobEventBus jobEventBus, final FrameworkIDService frameworkIDService) {
+ private SchedulerDriver getSchedulerDriver(final TaskScheduler taskScheduler, final JobTracingEventBus jobTracingEventBus, final FrameworkIDService frameworkIDService) {
Protos.FrameworkInfo.Builder builder = Protos.FrameworkInfo.newBuilder();
frameworkIDService.fetch().ifPresent(frameworkID -> builder.setId(Protos.FrameworkID.newBuilder().setValue(frameworkID).build()));
Optional<String> role = env.getMesosRole();
@@ -103,7 +104,7 @@ public final class SchedulerService {
Protos.FrameworkInfo frameworkInfo = builder.setUser(mesosConfig.getUser()).setName(frameworkName)
.setHostname(mesosConfig.getHostname()).setFailoverTimeout(MesosConfiguration.FRAMEWORK_FAILOVER_TIMEOUT_SECONDS)
.setWebuiUrl(WEB_UI_PROTOCOL + env.getFrameworkHostPort()).setCheckpoint(true).build();
- return new MesosSchedulerDriver(new SchedulerEngine(taskScheduler, facadeService, jobEventBus, frameworkIDService, statisticManager), frameworkInfo, mesosConfig.getUrl());
+ return new MesosSchedulerDriver(new SchedulerEngine(taskScheduler, facadeService, jobTracingEventBus, frameworkIDService, statisticManager), frameworkInfo, mesosConfig.getUrl());
}
private TaskScheduler getTaskScheduler() {
@@ -115,9 +116,9 @@ public final class SchedulerService {
}).build();
}
- private JobEventBus getJobEventBus() {
- Optional<TracingConfiguration> tracingConfiguration = env.getTracingConfiguration();
- return tracingConfiguration.map(JobEventBus::new).orElseGet(JobEventBus::new);
+ private JobTracingEventBus getJobTracingEventBus() {
+ Optional<TracingConfiguration<?>> tracingConfiguration = env.getTracingConfiguration();
+ return tracingConfiguration.map(JobTracingEventBus::new).orElseGet(JobTracingEventBus::new);
}
/**
diff --git a/elasticjob-cloud/elasticjob-cloud-scheduler/src/main/java/org/apache/shardingsphere/elasticjob/cloud/scheduler/mesos/TaskLaunchScheduledService.java b/elasticjob-cloud/elasticjob-cloud-scheduler/src/main/java/org/apache/shardingsphere/elasticjob/cloud/scheduler/mesos/TaskLaunchScheduledService.java
index 7fbe548..3fada70 100755
--- a/elasticjob-cloud/elasticjob-cloud-scheduler/src/main/java/org/apache/shardingsphere/elasticjob/cloud/scheduler/mesos/TaskLaunchScheduledService.java
+++ b/elasticjob-cloud/elasticjob-cloud-scheduler/src/main/java/org/apache/shardingsphere/elasticjob/cloud/scheduler/mesos/TaskLaunchScheduledService.java
@@ -44,7 +44,7 @@ import org.apache.shardingsphere.elasticjob.infra.context.TaskContext.MetaInfo;
import org.apache.shardingsphere.elasticjob.infra.json.GsonFactory;
import org.apache.shardingsphere.elasticjob.infra.listener.ShardingContexts;
import org.apache.shardingsphere.elasticjob.script.props.ScriptJobProperties;
-import org.apache.shardingsphere.elasticjob.tracing.JobEventBus;
+import org.apache.shardingsphere.elasticjob.tracing.JobTracingEventBus;
import org.apache.shardingsphere.elasticjob.tracing.event.JobStatusTraceEvent;
import org.apache.shardingsphere.elasticjob.tracing.event.JobStatusTraceEvent.Source;
@@ -71,7 +71,7 @@ public final class TaskLaunchScheduledService extends AbstractScheduledService {
private final FacadeService facadeService;
- private final JobEventBus jobEventBus;
+ private final JobTracingEventBus jobTracingEventBus;
private final BootstrapEnvironment env = BootstrapEnvironment.getINSTANCE();
@@ -118,7 +118,7 @@ public final class TaskLaunchScheduledService extends AbstractScheduledService {
}
for (TaskContext each : taskContextsList) {
facadeService.addRunning(each);
- jobEventBus.post(createJobStatusTraceEvent(each));
+ jobTracingEventBus.post(createJobStatusTraceEvent(each));
}
facadeService.removeLaunchTasksFromQueue(taskContextsList);
for (Entry<List<OfferID>, List<TaskInfo>> each : offerIdTaskInfoMap.entrySet()) {
diff --git a/elasticjob-cloud/elasticjob-cloud-scheduler/src/test/java/org/apache/shardingsphere/elasticjob/cloud/scheduler/env/BootstrapEnvironmentTest.java b/elasticjob-cloud/elasticjob-cloud-scheduler/src/test/java/org/apache/shardingsphere/elasticjob/cloud/scheduler/env/BootstrapEnvironmentTest.java
index f662dbf..15a1578 100755
--- a/elasticjob-cloud/elasticjob-cloud-scheduler/src/test/java/org/apache/shardingsphere/elasticjob/cloud/scheduler/env/BootstrapEnvironmentTest.java
+++ b/elasticjob-cloud/elasticjob-cloud-scheduler/src/test/java/org/apache/shardingsphere/elasticjob/cloud/scheduler/env/BootstrapEnvironmentTest.java
@@ -75,7 +75,7 @@ public final class BootstrapEnvironmentTest {
properties.setProperty(BootstrapEnvironment.EnvironmentArgument.EVENT_TRACE_RDB_USERNAME.getKey(), "sa");
properties.setProperty(BootstrapEnvironment.EnvironmentArgument.EVENT_TRACE_RDB_PASSWORD.getKey(), "password");
ReflectionUtils.setFieldValue(bootstrapEnvironment, "properties", properties);
- Optional<TracingConfiguration> tracingConfiguration = bootstrapEnvironment.getTracingConfiguration();
+ Optional<TracingConfiguration<?>> tracingConfiguration = bootstrapEnvironment.getTracingConfiguration();
tracingConfiguration.ifPresent(tracingConfiguration1 -> assertThat(tracingConfiguration1.getStorage(), instanceOf(BasicDataSource.class)));
}
diff --git a/elasticjob-cloud/elasticjob-cloud-scheduler/src/test/java/org/apache/shardingsphere/elasticjob/cloud/scheduler/mesos/SchedulerEngineTest.java b/elasticjob-cloud/elasticjob-cloud-scheduler/src/test/java/org/apache/shardingsphere/elasticjob/cloud/scheduler/mesos/SchedulerEngineTest.java
index 2855979..d4122d2 100755
--- a/elasticjob-cloud/elasticjob-cloud-scheduler/src/test/java/org/apache/shardingsphere/elasticjob/cloud/scheduler/mesos/SchedulerEngineTest.java
+++ b/elasticjob-cloud/elasticjob-cloud-scheduler/src/test/java/org/apache/shardingsphere/elasticjob/cloud/scheduler/mesos/SchedulerEngineTest.java
@@ -30,7 +30,7 @@ import org.apache.shardingsphere.elasticjob.cloud.scheduler.state.running.Runnin
import org.apache.shardingsphere.elasticjob.cloud.scheduler.statistics.StatisticManager;
import org.apache.shardingsphere.elasticjob.infra.context.TaskContext;
import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
-import org.apache.shardingsphere.elasticjob.tracing.JobEventBus;
+import org.apache.shardingsphere.elasticjob.tracing.JobTracingEventBus;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -68,7 +68,7 @@ public final class SchedulerEngineTest {
@Before
public void setUp() {
- schedulerEngine = new SchedulerEngine(taskScheduler, facadeService, new JobEventBus(), frameworkIDService, statisticManager);
+ schedulerEngine = new SchedulerEngine(taskScheduler, facadeService, new JobTracingEventBus(), frameworkIDService, statisticManager);
ReflectionUtils.setFieldValue(schedulerEngine, "facadeService", facadeService);
when(facadeService.load("test_job")).thenReturn(Optional.of(CloudJobConfigurationBuilder.createCloudJobConfiguration("test_job")));
new RunningService(mock(CoordinatorRegistryCenter.class)).clear();
diff --git a/elasticjob-cloud/elasticjob-cloud-scheduler/src/test/java/org/apache/shardingsphere/elasticjob/cloud/scheduler/mesos/TaskLaunchScheduledServiceTest.java b/elasticjob-cloud/elasticjob-cloud-scheduler/src/test/java/org/apache/shardingsphere/elasticjob/cloud/scheduler/mesos/TaskLaunchScheduledServiceTest.java
index bd682c5..3ca3344 100755
--- a/elasticjob-cloud/elasticjob-cloud-scheduler/src/test/java/org/apache/shardingsphere/elasticjob/cloud/scheduler/mesos/TaskLaunchScheduledServiceTest.java
+++ b/elasticjob-cloud/elasticjob-cloud-scheduler/src/test/java/org/apache/shardingsphere/elasticjob/cloud/scheduler/mesos/TaskLaunchScheduledServiceTest.java
@@ -34,7 +34,7 @@ import org.apache.shardingsphere.elasticjob.cloud.scheduler.mesos.fixture.OfferB
import org.apache.shardingsphere.elasticjob.infra.context.ExecutionType;
import org.apache.shardingsphere.elasticjob.infra.context.TaskContext;
import org.apache.shardingsphere.elasticjob.infra.context.TaskContext.MetaInfo;
-import org.apache.shardingsphere.elasticjob.tracing.JobEventBus;
+import org.apache.shardingsphere.elasticjob.tracing.JobTracingEventBus;
import org.apache.shardingsphere.elasticjob.tracing.event.JobStatusTraceEvent;
import org.junit.After;
import org.junit.Before;
@@ -70,14 +70,14 @@ public final class TaskLaunchScheduledServiceTest {
private FacadeService facadeService;
@Mock
- private JobEventBus jobEventBus;
+ private JobTracingEventBus jobTracingEventBus;
private TaskLaunchScheduledService taskLaunchScheduledService;
@Before
public void setUp() {
when(facadeService.loadAppConfig("test_app")).thenReturn(Optional.of(CloudAppConfigurationBuilder.createCloudAppConfiguration("test_app")));
- taskLaunchScheduledService = new TaskLaunchScheduledService(schedulerDriver, taskScheduler, facadeService, jobEventBus);
+ taskLaunchScheduledService = new TaskLaunchScheduledService(schedulerDriver, taskScheduler, facadeService, jobTracingEventBus);
taskLaunchScheduledService.startUp();
}
@@ -100,7 +100,7 @@ public final class TaskLaunchScheduledServiceTest {
taskLaunchScheduledService.runOneIteration();
verify(facadeService).removeLaunchTasksFromQueue(ArgumentMatchers.anyList());
verify(facadeService).loadAppConfig("test_app");
- verify(jobEventBus).post(ArgumentMatchers.<JobStatusTraceEvent>any());
+ verify(jobTracingEventBus).post(ArgumentMatchers.<JobStatusTraceEvent>any());
}
@Test
@@ -118,7 +118,7 @@ public final class TaskLaunchScheduledServiceTest {
verify(facadeService).removeLaunchTasksFromQueue(ArgumentMatchers.anyList());
verify(facadeService).isRunning(TaskContext.from(String.format("%s@-@0@-@%s@-@unassigned-slave@-@0", "script_job", ExecutionType.READY)));
verify(facadeService).loadAppConfig("test_app");
- verify(jobEventBus).post(ArgumentMatchers.<JobStatusTraceEvent>any());
+ verify(jobTracingEventBus).post(ArgumentMatchers.<JobStatusTraceEvent>any());
}
private TaskAssignmentResult mockTaskAssignmentResult(final String taskName, final ExecutionType executionType) {
diff --git a/elasticjob-lite/elasticjob-lite-core/src/main/java/org/apache/shardingsphere/elasticjob/lite/internal/schedule/JobScheduler.java b/elasticjob-lite/elasticjob-lite-core/src/main/java/org/apache/shardingsphere/elasticjob/lite/internal/schedule/JobScheduler.java
index bbe16e4..0a344ff 100644
--- a/elasticjob-lite/elasticjob-lite-core/src/main/java/org/apache/shardingsphere/elasticjob/lite/internal/schedule/JobScheduler.java
+++ b/elasticjob-lite/elasticjob-lite-core/src/main/java/org/apache/shardingsphere/elasticjob/lite/internal/schedule/JobScheduler.java
@@ -95,13 +95,13 @@ public final class JobScheduler {
setGuaranteeServiceForElasticJobListeners(regCenter, jobListeners);
jobScheduleController = createJobScheduleController();
}
-
+
private Collection<ElasticJobListener> getElasticJobListeners(final JobConfiguration jobConfig) {
return jobConfig.getJobListenerTypes().stream()
.map(type -> ElasticJobListenerFactory.createListener(type).orElseThrow(() -> new IllegalArgumentException(String.format("Can not find job listener type '%s'.", type))))
.collect(Collectors.toList());
}
-
+
private Optional<TracingConfiguration<?>> findTracingConfiguration(final JobConfiguration jobConfig) {
return jobConfig.getExtraConfigurations().stream().filter(each -> each instanceof TracingConfiguration).findFirst().map(extraConfig -> (TracingConfiguration<?>) extraConfig);
}
diff --git a/elasticjob-lite/elasticjob-lite-core/src/main/java/org/apache/shardingsphere/elasticjob/lite/internal/schedule/LiteJobFacade.java b/elasticjob-lite/elasticjob-lite-core/src/main/java/org/apache/shardingsphere/elasticjob/lite/internal/schedule/LiteJobFacade.java
index ad6fa88..6896500 100644
--- a/elasticjob-lite/elasticjob-lite-core/src/main/java/org/apache/shardingsphere/elasticjob/lite/internal/schedule/LiteJobFacade.java
+++ b/elasticjob-lite/elasticjob-lite-core/src/main/java/org/apache/shardingsphere/elasticjob/lite/internal/schedule/LiteJobFacade.java
@@ -31,7 +31,7 @@ import org.apache.shardingsphere.elasticjob.lite.internal.sharding.ExecutionCont
import org.apache.shardingsphere.elasticjob.lite.internal.sharding.ExecutionService;
import org.apache.shardingsphere.elasticjob.lite.internal.sharding.ShardingService;
import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
-import org.apache.shardingsphere.elasticjob.tracing.JobEventBus;
+import org.apache.shardingsphere.elasticjob.tracing.JobTracingEventBus;
import org.apache.shardingsphere.elasticjob.tracing.api.TracingConfiguration;
import org.apache.shardingsphere.elasticjob.tracing.event.JobExecutionEvent;
import org.apache.shardingsphere.elasticjob.tracing.event.JobStatusTraceEvent;
@@ -59,16 +59,16 @@ public final class LiteJobFacade implements JobFacade {
private final Collection<ElasticJobListener> elasticJobListeners;
- private final JobEventBus jobEventBus;
+ private final JobTracingEventBus jobTracingEventBus;
- public LiteJobFacade(final CoordinatorRegistryCenter regCenter, final String jobName, final Collection<ElasticJobListener> elasticJobListeners, final TracingConfiguration tracingConfig) {
+ public LiteJobFacade(final CoordinatorRegistryCenter regCenter, final String jobName, final Collection<ElasticJobListener> elasticJobListeners, final TracingConfiguration<?> tracingConfig) {
configService = new ConfigurationService(regCenter, jobName);
shardingService = new ShardingService(regCenter, jobName);
executionContextService = new ExecutionContextService(regCenter, jobName);
executionService = new ExecutionService(regCenter, jobName);
failoverService = new FailoverService(regCenter, jobName);
this.elasticJobListeners = elasticJobListeners;
- this.jobEventBus = null == tracingConfig ? new JobEventBus() : new JobEventBus(tracingConfig);
+ this.jobTracingEventBus = null == tracingConfig ? new JobTracingEventBus() : new JobTracingEventBus(tracingConfig);
}
@Override
@@ -155,13 +155,13 @@ public final class LiteJobFacade implements JobFacade {
@Override
public void postJobExecutionEvent(final JobExecutionEvent jobExecutionEvent) {
- jobEventBus.post(jobExecutionEvent);
+ jobTracingEventBus.post(jobExecutionEvent);
}
@Override
public void postJobStatusTraceEvent(final String taskId, final State state, final String message) {
TaskContext taskContext = TaskContext.from(taskId);
- jobEventBus.post(new JobStatusTraceEvent(taskContext.getMetaInfo().getJobName(), taskContext.getId(),
+ jobTracingEventBus.post(new JobStatusTraceEvent(taskContext.getMetaInfo().getJobName(), taskContext.getId(),
taskContext.getSlaveId(), Source.LITE_EXECUTOR, taskContext.getType().name(), taskContext.getMetaInfo().getShardingItems().toString(), state, message));
if (!Strings.isNullOrEmpty(message)) {
log.trace(message);
diff --git a/elasticjob-lite/elasticjob-lite-core/src/test/java/org/apache/shardingsphere/elasticjob/lite/internal/schedule/LiteJobFacadeTest.java b/elasticjob-lite/elasticjob-lite-core/src/test/java/org/apache/shardingsphere/elasticjob/lite/internal/schedule/LiteJobFacadeTest.java
index 9aedf95..4056e78 100644
--- a/elasticjob-lite/elasticjob-lite-core/src/test/java/org/apache/shardingsphere/elasticjob/lite/internal/schedule/LiteJobFacadeTest.java
+++ b/elasticjob-lite/elasticjob-lite-core/src/test/java/org/apache/shardingsphere/elasticjob/lite/internal/schedule/LiteJobFacadeTest.java
@@ -29,7 +29,7 @@ import org.apache.shardingsphere.elasticjob.lite.internal.sharding.ExecutionCont
import org.apache.shardingsphere.elasticjob.lite.internal.sharding.ExecutionService;
import org.apache.shardingsphere.elasticjob.lite.internal.sharding.ShardingService;
import org.apache.shardingsphere.elasticjob.lite.util.ReflectionUtils;
-import org.apache.shardingsphere.elasticjob.tracing.JobEventBus;
+import org.apache.shardingsphere.elasticjob.tracing.JobTracingEventBus;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -64,7 +64,7 @@ public final class LiteJobFacadeTest {
private FailoverService failoverService;
@Mock
- private JobEventBus jobEventBus;
+ private JobTracingEventBus jobTracingEventBus;
@Mock
private ElasticJobListenerCaller caller;
@@ -79,7 +79,7 @@ public final class LiteJobFacadeTest {
ReflectionUtils.setFieldValue(liteJobFacade, "executionContextService", executionContextService);
ReflectionUtils.setFieldValue(liteJobFacade, "executionService", executionService);
ReflectionUtils.setFieldValue(liteJobFacade, "failoverService", failoverService);
- ReflectionUtils.setFieldValue(liteJobFacade, "jobEventBus", jobEventBus);
+ ReflectionUtils.setFieldValue(liteJobFacade, "jobTracingEventBus", jobTracingEventBus);
}
@Test
@@ -210,6 +210,6 @@ public final class LiteJobFacadeTest {
@Test
public void assertPostJobExecutionEvent() {
liteJobFacade.postJobExecutionEvent(null);
- verify(jobEventBus).post(null);
+ verify(jobTracingEventBus).post(null);
}
}
diff --git a/elasticjob-tracing/elasticjob-tracing-api/src/main/java/org/apache/shardingsphere/elasticjob/tracing/JobEventBus.java b/elasticjob-tracing/elasticjob-tracing-api/src/main/java/org/apache/shardingsphere/elasticjob/tracing/JobTracingEventBus.java
similarity index 92%
rename from elasticjob-tracing/elasticjob-tracing-api/src/main/java/org/apache/shardingsphere/elasticjob/tracing/JobEventBus.java
rename to elasticjob-tracing/elasticjob-tracing-api/src/main/java/org/apache/shardingsphere/elasticjob/tracing/JobTracingEventBus.java
index cd87e40..50a18cc 100644
--- a/elasticjob-tracing/elasticjob-tracing-api/src/main/java/org/apache/shardingsphere/elasticjob/tracing/JobEventBus.java
+++ b/elasticjob-tracing/elasticjob-tracing-api/src/main/java/org/apache/shardingsphere/elasticjob/tracing/JobTracingEventBus.java
@@ -33,10 +33,10 @@ import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
- * Job event bus.
+ * Job tracing event bus.
*/
@Slf4j
-public final class JobEventBus {
+public final class JobTracingEventBus {
private final ExecutorService executorService;
@@ -44,12 +44,12 @@ public final class JobEventBus {
private volatile boolean isRegistered;
- public JobEventBus() {
+ public JobTracingEventBus() {
executorService = null;
eventBus = null;
}
- public JobEventBus(final TracingConfiguration tracingConfig) {
+ public JobTracingEventBus(final TracingConfiguration<?> tracingConfig) {
executorService = createExecutorService(Runtime.getRuntime().availableProcessors() * 2);
eventBus = new AsyncEventBus(executorService);
register(tracingConfig);
@@ -62,7 +62,7 @@ public final class JobEventBus {
return MoreExecutors.listeningDecorator(MoreExecutors.getExitingExecutorService(threadPoolExecutor));
}
- private void register(final TracingConfiguration tracingConfig) {
+ private void register(final TracingConfiguration<?> tracingConfig) {
try {
eventBus.register(TracingListenerFactory.getListener(tracingConfig));
isRegistered = true;
diff --git a/elasticjob-tracing/elasticjob-tracing-api/src/test/java/org/apache/shardingsphere/elasticjob/tracing/JobEventBusTest.java b/elasticjob-tracing/elasticjob-tracing-api/src/test/java/org/apache/shardingsphere/elasticjob/tracing/JobTracingEventBusTest.java
similarity index 72%
rename from elasticjob-tracing/elasticjob-tracing-api/src/test/java/org/apache/shardingsphere/elasticjob/tracing/JobEventBusTest.java
rename to elasticjob-tracing/elasticjob-tracing-api/src/test/java/org/apache/shardingsphere/elasticjob/tracing/JobTracingEventBusTest.java
index a416ace..3d1cd73 100644
--- a/elasticjob-tracing/elasticjob-tracing-api/src/test/java/org/apache/shardingsphere/elasticjob/tracing/JobEventBusTest.java
+++ b/elasticjob-tracing/elasticjob-tracing-api/src/test/java/org/apache/shardingsphere/elasticjob/tracing/JobTracingEventBusTest.java
@@ -38,7 +38,7 @@ import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@RunWith(MockitoJUnitRunner.class)
-public final class JobEventBusTest {
+public final class JobTracingEventBusTest {
@Mock
private JobEventCaller jobEventCaller;
@@ -46,19 +46,19 @@ public final class JobEventBusTest {
@Mock
private EventBus eventBus;
- private JobEventBus jobEventBus;
+ private JobTracingEventBus jobTracingEventBus;
@Test
public void assertRegisterFailure() {
- jobEventBus = new JobEventBus(new TracingConfiguration<>("FAIL", null));
+ jobTracingEventBus = new JobTracingEventBus(new TracingConfiguration<>("FAIL", null));
assertIsRegistered(false);
}
@Test
public void assertPost() throws InterruptedException {
- jobEventBus = new JobEventBus(new TracingConfiguration<>("TEST", jobEventCaller));
+ jobTracingEventBus = new JobTracingEventBus(new TracingConfiguration<>("TEST", jobEventCaller));
assertIsRegistered(true);
- jobEventBus.post(new JobExecutionEvent("localhost", "127.0.0.1", "fake_task_id", "test_event_bus_job", JobExecutionEvent.ExecutionSource.NORMAL_TRIGGER, 0));
+ jobTracingEventBus.post(new JobExecutionEvent("localhost", "127.0.0.1", "fake_task_id", "test_event_bus_job", JobExecutionEvent.ExecutionSource.NORMAL_TRIGGER, 0));
while (!TestTracingListener.isExecutionEventCalled()) {
Thread.sleep(100L);
}
@@ -67,19 +67,19 @@ public final class JobEventBusTest {
@Test
public void assertPostWithoutListener() throws ReflectiveOperationException {
- jobEventBus = new JobEventBus();
+ jobTracingEventBus = new JobTracingEventBus();
assertIsRegistered(false);
- Field field = JobEventBus.class.getDeclaredField("eventBus");
+ Field field = JobTracingEventBus.class.getDeclaredField("eventBus");
field.setAccessible(true);
- field.set(jobEventBus, eventBus);
- jobEventBus.post(new JobExecutionEvent("localhost", "127.0.0.1", "fake_task_id", "test_event_bus_job", JobExecutionEvent.ExecutionSource.NORMAL_TRIGGER, 0));
+ field.set(jobTracingEventBus, eventBus);
+ jobTracingEventBus.post(new JobExecutionEvent("localhost", "127.0.0.1", "fake_task_id", "test_event_bus_job", JobExecutionEvent.ExecutionSource.NORMAL_TRIGGER, 0));
verify(eventBus, times(0)).post(ArgumentMatchers.<JobEvent>any());
}
@SneakyThrows
private void assertIsRegistered(final boolean actual) {
- Field field = JobEventBus.class.getDeclaredField("isRegistered");
+ Field field = JobTracingEventBus.class.getDeclaredField("isRegistered");
field.setAccessible(true);
- assertThat(field.get(jobEventBus), is(actual));
+ assertThat(field.get(jobTracingEventBus), is(actual));
}
}
diff --git a/elasticjob-tracing/elasticjob-tracing-api/src/test/resources/logback-test.xml b/elasticjob-tracing/elasticjob-tracing-api/src/test/resources/logback-test.xml
index ecc5b0f..6c61d95 100644
--- a/elasticjob-tracing/elasticjob-tracing-api/src/test/resources/logback-test.xml
+++ b/elasticjob-tracing/elasticjob-tracing-api/src/test/resources/logback-test.xml
@@ -37,5 +37,5 @@
</root>
<logger name="RDBTracingListenerConfiguration" level="OFF" />
- <logger name="org.apache.shardingsphere.elasticjob.tracing.JobEventBus" level="OFF" />
+ <logger name="org.apache.shardingsphere.elasticjob.tracing.JobTracingEventBus" level="OFF" />
</configuration>
diff --git a/elasticjob-tracing/elasticjob-tracing-rdb/src/test/java/org/apache/shardingsphere/elasticjob/tracing/rdb/listener/RDBTracingListenerTest.java b/elasticjob-tracing/elasticjob-tracing-rdb/src/test/java/org/apache/shardingsphere/elasticjob/tracing/rdb/listener/RDBTracingListenerTest.java
index 049677c..7c74c62 100644
--- a/elasticjob-tracing/elasticjob-tracing-rdb/src/test/java/org/apache/shardingsphere/elasticjob/tracing/rdb/listener/RDBTracingListenerTest.java
+++ b/elasticjob-tracing/elasticjob-tracing-rdb/src/test/java/org/apache/shardingsphere/elasticjob/tracing/rdb/listener/RDBTracingListenerTest.java
@@ -19,7 +19,7 @@ package org.apache.shardingsphere.elasticjob.tracing.rdb.listener;
import lombok.SneakyThrows;
import org.apache.commons.dbcp.BasicDataSource;
-import org.apache.shardingsphere.elasticjob.tracing.JobEventBus;
+import org.apache.shardingsphere.elasticjob.tracing.JobTracingEventBus;
import org.apache.shardingsphere.elasticjob.tracing.api.TracingConfiguration;
import org.apache.shardingsphere.elasticjob.tracing.event.JobExecutionEvent;
import org.apache.shardingsphere.elasticjob.tracing.event.JobStatusTraceEvent;
@@ -47,7 +47,7 @@ public final class RDBTracingListenerTest {
@Mock
private RDBJobEventStorage repository;
- private JobEventBus jobEventBus;
+ private JobTracingEventBus jobTracingEventBus;
@Before
public void setUp() throws SQLException {
@@ -58,7 +58,7 @@ public final class RDBTracingListenerTest {
dataSource.setPassword("");
RDBTracingListener tracingListener = new RDBTracingListener(dataSource);
setRepository(tracingListener);
- jobEventBus = new JobEventBus(new TracingConfiguration<DataSource>("RDB", dataSource));
+ jobTracingEventBus = new JobTracingEventBus(new TracingConfiguration<DataSource>("RDB", dataSource));
}
@SneakyThrows
@@ -71,14 +71,14 @@ public final class RDBTracingListenerTest {
@Test
public void assertPostJobExecutionEvent() {
JobExecutionEvent jobExecutionEvent = new JobExecutionEvent("localhost", "127.0.0.1", "fake_task_id", JOB_NAME, JobExecutionEvent.ExecutionSource.NORMAL_TRIGGER, 0);
- jobEventBus.post(jobExecutionEvent);
+ jobTracingEventBus.post(jobExecutionEvent);
verify(repository, atMost(1)).addJobExecutionEvent(jobExecutionEvent);
}
@Test
public void assertPostJobStatusTraceEvent() {
JobStatusTraceEvent jobStatusTraceEvent = new JobStatusTraceEvent(JOB_NAME, "fake_task_id", "fake_slave_id", Source.LITE_EXECUTOR, "READY", "0", State.TASK_RUNNING, "message is empty.");
- jobEventBus.post(jobStatusTraceEvent);
+ jobTracingEventBus.post(jobStatusTraceEvent);
verify(repository, atMost(1)).addJobStatusTraceEvent(jobStatusTraceEvent);
}
}