Skip to content

Commit

Permalink
Remove JobStatusTraceEvent.Source
Browse files Browse the repository at this point in the history
  • Loading branch information
terrymanu committed Oct 14, 2023
1 parent ea0c02b commit 33ec1f6
Show file tree
Hide file tree
Showing 12 changed files with 25 additions and 36 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -44,8 +44,6 @@ public final class JobStatusTraceEvent implements JobEvent {

private final String slaveId;

private final Source source;

private final String executionType;

private final String shardingItems;
Expand All @@ -59,8 +57,4 @@ public final class JobStatusTraceEvent implements JobEvent {
public enum State {
TASK_STAGING, TASK_RUNNING, TASK_FINISHED, TASK_KILLED, TASK_LOST, TASK_FAILED, TASK_ERROR, TASK_DROPPED, TASK_GONE, TASK_GONE_BY_OPERATOR, TASK_UNREACHABLE, TASK_UNKNOWN
}

public enum Source {
CLOUD_SCHEDULER, CLOUD_EXECUTOR, LITE_EXECUTOR
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@
import lombok.extern.slf4j.Slf4j;
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;
import org.apache.shardingsphere.elasticjob.tracing.event.JobStatusTraceEvent.State;
import org.apache.shardingsphere.elasticjob.tracing.exception.WrapException;
import org.apache.shardingsphere.elasticjob.tracing.rdb.type.DatabaseType;
Expand Down Expand Up @@ -353,12 +352,11 @@ public boolean addJobStatusTraceEvent(final JobStatusTraceEvent jobStatusTraceEv
preparedStatement.setString(3, originalTaskId);
preparedStatement.setString(4, jobStatusTraceEvent.getTaskId());
preparedStatement.setString(5, jobStatusTraceEvent.getSlaveId());
preparedStatement.setString(6, jobStatusTraceEvent.getSource().toString());
preparedStatement.setString(7, jobStatusTraceEvent.getExecutionType());
preparedStatement.setString(8, jobStatusTraceEvent.getShardingItems());
preparedStatement.setString(9, jobStatusTraceEvent.getState().toString());
preparedStatement.setString(10, truncateString(jobStatusTraceEvent.getMessage()));
preparedStatement.setTimestamp(11, new Timestamp(jobStatusTraceEvent.getCreationTime().getTime()));
preparedStatement.setString(6, jobStatusTraceEvent.getExecutionType());
preparedStatement.setString(7, jobStatusTraceEvent.getShardingItems());
preparedStatement.setString(8, jobStatusTraceEvent.getState().toString());
preparedStatement.setString(9, truncateString(jobStatusTraceEvent.getMessage()));
preparedStatement.setTimestamp(10, new Timestamp(jobStatusTraceEvent.getCreationTime().getTime()));
preparedStatement.execute();
result = true;
} catch (final SQLException ex) {
Expand Down Expand Up @@ -399,8 +397,8 @@ List<JobStatusTraceEvent> getJobStatusTraceEvents(final String taskId) {
try (ResultSet resultSet = preparedStatement.executeQuery()) {
while (resultSet.next()) {
JobStatusTraceEvent jobStatusTraceEvent = new JobStatusTraceEvent(resultSet.getString(1), resultSet.getString(2), resultSet.getString(3), resultSet.getString(4),
resultSet.getString(5), Source.valueOf(resultSet.getString(6)), resultSet.getString(7), resultSet.getString(8),
State.valueOf(resultSet.getString(9)), resultSet.getString(10), new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(resultSet.getString(11)));
resultSet.getString(5), resultSet.getString(6), resultSet.getString(7),
State.valueOf(resultSet.getString(8)), resultSet.getString(9), new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(resultSet.getString(10)));
result.add(jobStatusTraceEvent);
}
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,6 @@ JOB_EXECUTION_LOG.UPDATE_FAILURE=UPDATE JOB_EXECUTION_LOG SET is_success = ?, co
JOB_STATUS_TRACE_LOG.TABLE.CREATE=CREATE TABLE JOB_STATUS_TRACE_LOG (id VARCHAR(40) NOT NULL, job_name VARCHAR(100) NOT NULL, original_task_id VARCHAR(255) NOT NULL, task_id VARCHAR(255) NOT NULL, slave_id VARCHAR(50) NOT NULL, source VARCHAR(50) NOT NULL, execution_type VARCHAR(20) NOT NULL, sharding_item VARCHAR(100) NOT NULL, state VARCHAR(20) NOT NULL, message VARCHAR(4000) NULL, creation_time TIMESTAMP NULL, PRIMARY KEY (id))
TASK_ID_STATE_INDEX.INDEX.CREATE=CREATE INDEX TASK_ID_STATE_INDEX ON JOB_STATUS_TRACE_LOG (task_id(128), state)

JOB_STATUS_TRACE_LOG.INSERT=INSERT INTO JOB_STATUS_TRACE_LOG (id, job_name, original_task_id, task_id, slave_id, source, execution_type, sharding_item, state, message, creation_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
JOB_STATUS_TRACE_LOG.INSERT=INSERT INTO JOB_STATUS_TRACE_LOG (id, job_name, original_task_id, task_id, slave_id, execution_type, sharding_item, state, message, creation_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
JOB_STATUS_TRACE_LOG.SELECT=SELECT * FROM JOB_STATUS_TRACE_LOG WHERE task_id = ?
JOB_STATUS_TRACE_LOG.SELECT_ORIGINAL_TASK_ID=SELECT * FROM (SELECT ROWNUMBER() OVER() AS ROW, A.* FROM JOB_STATUS_TRACE_LOG A WHERE A.TASK_ID = '4' AND A.STATE= 'TASK_STAGING') AS B WHERE B.ROW = 1
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,6 @@ JOB_EXECUTION_LOG.UPDATE_FAILURE=UPDATE JOB_EXECUTION_LOG SET is_success = ?, co
JOB_STATUS_TRACE_LOG.TABLE.CREATE=CREATE TABLE IF NOT EXISTS JOB_STATUS_TRACE_LOG (id VARCHAR(40) NOT NULL, job_name VARCHAR(100) NOT NULL, original_task_id VARCHAR(255) NOT NULL, task_id VARCHAR(255) NOT NULL, slave_id VARCHAR(50) NOT NULL, source VARCHAR(50) NOT NULL, execution_type VARCHAR(20) NOT NULL, sharding_item VARCHAR(100) NOT NULL, state VARCHAR(20) NOT NULL, message VARCHAR(4000) NULL, creation_time TIMESTAMP NULL, PRIMARY KEY (id))
TASK_ID_STATE_INDEX.INDEX.CREATE=CREATE INDEX IF NOT EXISTS TASK_ID_STATE_INDEX ON JOB_STATUS_TRACE_LOG (task_id, state)

JOB_STATUS_TRACE_LOG.INSERT=INSERT INTO JOB_STATUS_TRACE_LOG (id, job_name, original_task_id, task_id, slave_id, source, execution_type, sharding_item, state, message, creation_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
JOB_STATUS_TRACE_LOG.INSERT=INSERT INTO JOB_STATUS_TRACE_LOG (id, job_name, original_task_id, task_id, slave_id, execution_type, sharding_item, state, message, creation_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
JOB_STATUS_TRACE_LOG.SELECT=SELECT * FROM JOB_STATUS_TRACE_LOG WHERE task_id = ?
JOB_STATUS_TRACE_LOG.SELECT_ORIGINAL_TASK_ID=SELECT original_task_id FROM JOB_STATUS_TRACE_LOG WHERE task_id = ? and state= 'TASK_STAGING' LIMIT 1
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,6 @@ JOB_STATUS_TRACE_LOG.TABLE.CREATE= CREATE TABLE \

TASK_ID_STATE_INDEX.INDEX.CREATE=CREATE INDEX TASK_ID_STATE_INDEX ON JOB_STATUS_TRACE_LOG (task_id(128), state)

JOB_STATUS_TRACE_LOG.INSERT=INSERT INTO JOB_STATUS_TRACE_LOG (id, job_name, original_task_id, task_id, slave_id, source, execution_type, sharding_item, state, message, creation_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
JOB_STATUS_TRACE_LOG.INSERT=INSERT INTO JOB_STATUS_TRACE_LOG (id, job_name, original_task_id, task_id, slave_id, execution_type, sharding_item, state, message, creation_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
JOB_STATUS_TRACE_LOG.SELECT=SELECT id, job_name, original_task_id, task_id, slave_id, source, execution_type, sharding_item, state, message, creation_time FROM JOB_STATUS_TRACE_LOG WHERE task_id = ?
JOB_STATUS_TRACE_LOG.SELECT_ORIGINAL_TASK_ID=SELECT original_task_id FROM JOB_STATUS_TRACE_LOG WHERE task_id = ? and state= 'TASK_STAGING' LIMIT 1
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,6 @@ JOB_EXECUTION_LOG.UPDATE_FAILURE=UPDATE JOB_EXECUTION_LOG SET is_success = ?, co
JOB_STATUS_TRACE_LOG.TABLE.CREATE=CREATE TABLE JOB_STATUS_TRACE_LOG (id VARCHAR(40) NOT NULL, job_name VARCHAR(100) NOT NULL, original_task_id VARCHAR(255) NOT NULL, task_id VARCHAR(255) NOT NULL, slave_id VARCHAR(50) NOT NULL, source VARCHAR(50) NOT NULL, execution_type VARCHAR(20) NOT NULL, sharding_item VARCHAR(100) NOT NULL, state VARCHAR(20) NOT NULL, message VARCHAR(4000) NULL, creation_time TIMESTAMP NULL, PRIMARY KEY (id))
TASK_ID_STATE_INDEX.INDEX.CREATE=CREATE INDEX TASK_ID_STATE_INDEX ON JOB_STATUS_TRACE_LOG (task_id, state)

JOB_STATUS_TRACE_LOG.INSERT=INSERT INTO JOB_STATUS_TRACE_LOG (id, job_name, original_task_id, task_id, slave_id, source, execution_type, sharding_item, state, message, creation_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
JOB_STATUS_TRACE_LOG.INSERT=INSERT INTO JOB_STATUS_TRACE_LOG (id, job_name, original_task_id, task_id, slave_id, execution_type, sharding_item, state, message, creation_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
JOB_STATUS_TRACE_LOG.SELECT=SELECT * FROM JOB_STATUS_TRACE_LOG WHERE task_id = ?
JOB_STATUS_TRACE_LOG.SELECT_ORIGINAL_TASK_ID=SELECT original_task_id FROM JOB_STATUS_TRACE_LOG WHERE task_id = ? and state= 'TASK_STAGING' and ROWNUM = 1
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,6 @@ JOB_EXECUTION_LOG.UPDATE_FAILURE=UPDATE JOB_EXECUTION_LOG SET is_success = ?, co
JOB_STATUS_TRACE_LOG.TABLE.CREATE=CREATE TABLE JOB_STATUS_TRACE_LOG (id VARCHAR(40) NOT NULL, job_name VARCHAR(100) NOT NULL, original_task_id VARCHAR(255) NOT NULL, task_id VARCHAR(255) NOT NULL, slave_id VARCHAR(50) NOT NULL, source VARCHAR(50) NOT NULL, execution_type VARCHAR(20) NOT NULL, sharding_item VARCHAR(100) NOT NULL, state VARCHAR(20) NOT NULL, message VARCHAR(4000) NULL, creation_time TIMESTAMP NULL, PRIMARY KEY (id))
TASK_ID_STATE_INDEX.INDEX.CREATE=CREATE INDEX TASK_ID_STATE_INDEX ON JOB_STATUS_TRACE_LOG (task_id, state)

JOB_STATUS_TRACE_LOG.INSERT=INSERT INTO JOB_STATUS_TRACE_LOG (id, job_name, original_task_id, task_id, slave_id, source, execution_type, sharding_item, state, message, creation_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
JOB_STATUS_TRACE_LOG.INSERT=INSERT INTO JOB_STATUS_TRACE_LOG (id, job_name, original_task_id, task_id, slave_id, execution_type, sharding_item, state, message, creation_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
JOB_STATUS_TRACE_LOG.SELECT=SELECT * FROM JOB_STATUS_TRACE_LOG WHERE task_id=?
JOB_STATUS_TRACE_LOG.SELECT_ORIGINAL_TASK_ID=SELECT original_task_id FROM JOB_STATUS_TRACE_LOG WHERE task_id=? and state='TASK_STAGING' LIMIT 1
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,6 @@ JOB_EXECUTION_LOG.UPDATE_FAILURE=UPDATE JOB_EXECUTION_LOG SET is_success = ?, co
JOB_STATUS_TRACE_LOG.TABLE.CREATE=CREATE TABLE JOB_STATUS_TRACE_LOG (id CHARACTER(40) NOT NULL, job_name CHARACTER(100) NOT NULL, original_task_id CHARACTER(255) NOT NULL, task_id CHARACTER(255) NOT NULL, slave_id CHARACTER(50) NOT NULL, source CHARACTER(50) NOT NULL, execution_type CHARACTER(20) NOT NULL, sharding_item CHARACTER(100) NOT NULL, state CHARACTER(20) NOT NULL, message CHARACTER VARYING(4000) NULL, creation_time TIMESTAMP NULL, PRIMARY KEY (id))
TASK_ID_STATE_INDEX.INDEX.CREATE=CREATE INDEX TASK_ID_STATE_INDEX ON JOB_STATUS_TRACE_LOG (task_id, state)

JOB_STATUS_TRACE_LOG.INSERT=INSERT INTO JOB_STATUS_TRACE_LOG (id, job_name, original_task_id, task_id, slave_id, source, execution_type, sharding_item, state, message, creation_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
JOB_STATUS_TRACE_LOG.INSERT=INSERT INTO JOB_STATUS_TRACE_LOG (id, job_name, original_task_id, task_id, slave_id, execution_type, sharding_item, state, message, creation_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
JOB_STATUS_TRACE_LOG.SELECT=SELECT * FROM JOB_STATUS_TRACE_LOG WHERE task_id=?
JOB_STATUS_TRACE_LOG.SELECT_ORIGINAL_TASK_ID=SELECT original_task_id FROM JOB_STATUS_TRACE_LOG WHERE task_id=? and state='TASK_STAGING' LIMIT 1
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,6 @@ JOB_EXECUTION_LOG.UPDATE_FAILURE=UPDATE JOB_EXECUTION_LOG SET is_success = ?, co
JOB_STATUS_TRACE_LOG.TABLE.CREATE=CREATE TABLE JOB_STATUS_TRACE_LOG (id VARCHAR(40) NOT NULL, job_name VARCHAR(100) NOT NULL, original_task_id VARCHAR(255) NOT NULL, task_id VARCHAR(255) NOT NULL, slave_id VARCHAR(50) NOT NULL, source VARCHAR(50) NOT NULL, execution_type VARCHAR(20) NOT NULL, sharding_item VARCHAR(100) NOT NULL, state VARCHAR(20) NOT NULL, message VARCHAR(4000) NULL, creation_time DATETIME NULL, PRIMARY KEY (id))
TASK_ID_STATE_INDEX.INDEX.CREATE=CREATE INDEX TASK_ID_STATE_INDEX ON JOB_STATUS_TRACE_LOG (task_id, state)

JOB_STATUS_TRACE_LOG.INSERT=INSERT INTO JOB_STATUS_TRACE_LOG (id, job_name, original_task_id, task_id, slave_id, source, execution_type, sharding_item, state, message, creation_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
JOB_STATUS_TRACE_LOG.INSERT=INSERT INTO JOB_STATUS_TRACE_LOG (id, job_name, original_task_id, task_id, slave_id, execution_type, sharding_item, state, message, creation_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
JOB_STATUS_TRACE_LOG.SELECT=SELECT * FROM JOB_STATUS_TRACE_LOG WHERE task_id = ?
JOB_STATUS_TRACE_LOG.SELECT_ORIGINAL_TASK_ID=SELECT TOP 1 original_task_id FROM JOB_STATUS_TRACE_LOG WHERE task_id = ? and state = 'TASK_STAGING'
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,6 @@
import org.apache.shardingsphere.elasticjob.tracing.api.TracingConfiguration;
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;
import org.apache.shardingsphere.elasticjob.tracing.event.JobStatusTraceEvent.State;
import org.apache.shardingsphere.elasticjob.tracing.rdb.storage.RDBJobEventStorage;
import org.junit.jupiter.api.BeforeEach;
Expand Down Expand Up @@ -77,7 +76,7 @@ void assertPostJobExecutionEvent() {

@Test
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.");
JobStatusTraceEvent jobStatusTraceEvent = new JobStatusTraceEvent(JOB_NAME, "fake_task_id", "fake_slave_id", "READY", "0", State.TASK_RUNNING, "message is empty.");
jobTracingEventBus.post(jobStatusTraceEvent);
verify(repository, atMost(1)).addJobStatusTraceEvent(jobStatusTraceEvent);
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,6 @@
import org.apache.commons.dbcp2.BasicDataSource;
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;
import org.apache.shardingsphere.elasticjob.tracing.event.JobStatusTraceEvent.State;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
Expand Down Expand Up @@ -65,13 +64,13 @@ void assertAddJobExecutionEvent() {
@Test
void assertAddJobStatusTraceEvent() {
assertTrue(storage.addJobStatusTraceEvent(
new JobStatusTraceEvent("test_job", "fake_task_id", "fake_slave_id", Source.LITE_EXECUTOR, "READY", "0", State.TASK_RUNNING, "message is empty.")));
new JobStatusTraceEvent("test_job", "fake_task_id", "fake_slave_id", "READY", "0", State.TASK_RUNNING, "message is empty.")));
}

@Test
void assertAddJobStatusTraceEventWhenFailoverWithTaskStagingState() {
JobStatusTraceEvent jobStatusTraceEvent = new JobStatusTraceEvent(
"test_job", "fake_failover_task_id", "fake_slave_id", Source.LITE_EXECUTOR, "FAILOVER", "0", State.TASK_STAGING, "message is empty.");
"test_job", "fake_failover_task_id", "fake_slave_id", "FAILOVER", "0", State.TASK_STAGING, "message is empty.");
jobStatusTraceEvent.setOriginalTaskId("original_fake_failover_task_id");
assertThat(storage.getJobStatusTraceEvents("fake_failover_task_id").size(), is(0));
storage.addJobStatusTraceEvent(jobStatusTraceEvent);
Expand All @@ -81,11 +80,11 @@ void assertAddJobStatusTraceEventWhenFailoverWithTaskStagingState() {
@Test
void assertAddJobStatusTraceEventWhenFailoverWithTaskFailedState() {
JobStatusTraceEvent stagingJobStatusTraceEvent = new JobStatusTraceEvent(
"test_job", "fake_failed_failover_task_id", "fake_slave_id", Source.LITE_EXECUTOR, "FAILOVER", "0", State.TASK_STAGING, "message is empty.");
"test_job", "fake_failed_failover_task_id", "fake_slave_id", "FAILOVER", "0", State.TASK_STAGING, "message is empty.");
stagingJobStatusTraceEvent.setOriginalTaskId("original_fake_failed_failover_task_id");
storage.addJobStatusTraceEvent(stagingJobStatusTraceEvent);
JobStatusTraceEvent failedJobStatusTraceEvent = new JobStatusTraceEvent(
"test_job", "fake_failed_failover_task_id", "fake_slave_id", Source.LITE_EXECUTOR, "FAILOVER", "0", State.TASK_FAILED, "message is empty.");
"test_job", "fake_failed_failover_task_id", "fake_slave_id", "FAILOVER", "0", State.TASK_FAILED, "message is empty.");
storage.addJobStatusTraceEvent(failedJobStatusTraceEvent);
List<JobStatusTraceEvent> jobStatusTraceEvents = storage.getJobStatusTraceEvents("fake_failed_failover_task_id");
assertThat(jobStatusTraceEvents.size(), is(2));
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,22 +20,21 @@
import com.google.common.base.Strings;
import lombok.extern.slf4j.Slf4j;
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.exception.JobExecutionEnvironmentException;
import org.apache.shardingsphere.elasticjob.infra.listener.ElasticJobListener;
import org.apache.shardingsphere.elasticjob.infra.listener.ShardingContexts;
import org.apache.shardingsphere.elasticjob.engine.internal.config.ConfigurationService;
import org.apache.shardingsphere.elasticjob.engine.internal.failover.FailoverService;
import org.apache.shardingsphere.elasticjob.engine.internal.sharding.ExecutionContextService;
import org.apache.shardingsphere.elasticjob.engine.internal.sharding.ExecutionService;
import org.apache.shardingsphere.elasticjob.engine.internal.sharding.ShardingService;
import org.apache.shardingsphere.elasticjob.executor.JobFacade;
import org.apache.shardingsphere.elasticjob.infra.context.TaskContext;
import org.apache.shardingsphere.elasticjob.infra.exception.JobExecutionEnvironmentException;
import org.apache.shardingsphere.elasticjob.infra.listener.ElasticJobListener;
import org.apache.shardingsphere.elasticjob.infra.listener.ShardingContexts;
import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
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;
import org.apache.shardingsphere.elasticjob.tracing.event.JobStatusTraceEvent.Source;
import org.apache.shardingsphere.elasticjob.tracing.event.JobStatusTraceEvent.State;

import java.util.Collection;
Expand Down Expand Up @@ -164,7 +163,7 @@ public void postJobExecutionEvent(final JobExecutionEvent jobExecutionEvent) {
public void postJobStatusTraceEvent(final String taskId, final State state, final String message) {
TaskContext taskContext = TaskContext.from(taskId);
jobTracingEventBus.post(new JobStatusTraceEvent(taskContext.getMetaInfo().getJobName(), taskContext.getId(),
taskContext.getSlaveId(), Source.LITE_EXECUTOR, taskContext.getType().name(), taskContext.getMetaInfo().getShardingItems().toString(), state, message));
taskContext.getSlaveId(), taskContext.getType().name(), taskContext.getMetaInfo().getShardingItems().toString(), state, message));
if (!Strings.isNullOrEmpty(message)) {
log.trace(message);
}
Expand Down

0 comments on commit 33ec1f6

Please sign in to comment.