forked from apache/eventmesh
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
15 changed files
with
247 additions
and
101 deletions.
There are no files selected for viewing
29 changes: 29 additions & 0 deletions
29
...ver/src/main/java/com/apache/eventmesh/admin/server/web/db/entity/EventMeshJobDetail.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,29 @@ | ||
| package com.apache.eventmesh.admin.server.web.db.entity; | ||
|
|
||
| import lombok.Data; | ||
| import org.apache.eventmesh.common.remote.JobState; | ||
| import org.apache.eventmesh.common.remote.job.JobTransportType; | ||
| import org.apache.eventmesh.common.remote.offset.RecordPosition; | ||
|
|
||
| import java.util.Map; | ||
|
|
||
| @Data | ||
| public class EventMeshJobDetail { | ||
| private Integer id; | ||
|
|
||
| private String name; | ||
|
|
||
| private JobTransportType transportType; | ||
|
|
||
| private Map<String, Object> sourceConnectorConfig; | ||
|
|
||
| private String sourceConnectorDesc; | ||
|
|
||
| private Map<String, Object> sinkConnectorConfig; | ||
|
|
||
| private String sinkConnectorDesc; | ||
|
|
||
| private RecordPosition position; | ||
|
|
||
| private JobState state; | ||
| } |
16 changes: 16 additions & 0 deletions
16
...ain/java/com/apache/eventmesh/admin/server/web/db/service/EventMeshJobInfoExtService.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,16 @@ | ||
| package com.apache.eventmesh.admin.server.web.db.service; | ||
|
|
||
| import com.apache.eventmesh.admin.server.web.db.entity.EventMeshJobDetail; | ||
| import com.apache.eventmesh.admin.server.web.db.entity.EventMeshJobInfo; | ||
| import com.baomidou.mybatisplus.extension.service.IService; | ||
| import org.apache.eventmesh.common.remote.JobState; | ||
|
|
||
| /** | ||
| * @author sodafang | ||
| * @description 针对表【event_mesh_job_info】的数据库操作Service | ||
| * @createDate 2024-05-09 15:51:45 | ||
| */ | ||
| public interface EventMeshJobInfoExtService extends IService<EventMeshJobInfo> { | ||
| boolean updateJobState(Integer jobID, JobState state); | ||
| EventMeshJobDetail getJobDetail(Integer jobID); | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
13 changes: 13 additions & 0 deletions
13
...va/com/apache/eventmesh/admin/server/web/db/service/EventMeshMysqlPositionExtService.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,13 @@ | ||
| package com.apache.eventmesh.admin.server.web.db.service; | ||
|
|
||
| import com.apache.eventmesh.admin.server.web.db.entity.EventMeshMysqlPosition; | ||
| import com.baomidou.mybatisplus.extension.service.IService; | ||
|
|
||
| /** | ||
| * @author sodafang | ||
| * @description 针对表【event_mesh_mysql_position】的数据库操作Service | ||
| * @createDate 2024-05-14 17:15:03 | ||
| */ | ||
| public interface EventMeshMysqlPositionExtService extends IService<EventMeshMysqlPosition> { | ||
| boolean saveOrUpdateByJob(EventMeshMysqlPosition position); | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
13 changes: 13 additions & 0 deletions
13
...com/apache/eventmesh/admin/server/web/db/service/EventMeshRuntimeHeartbeatExtService.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,13 @@ | ||
| package com.apache.eventmesh.admin.server.web.db.service; | ||
|
|
||
| import com.apache.eventmesh.admin.server.web.db.entity.EventMeshRuntimeHeartbeat; | ||
| import com.baomidou.mybatisplus.extension.service.IService; | ||
|
|
||
| /** | ||
| * @author sodafang | ||
| * @description 针对表【event_mesh_runtime_heartbeat】的数据库操作Service | ||
| * @createDate 2024-05-14 17:15:03 | ||
| */ | ||
| public interface EventMeshRuntimeHeartbeatExtService extends IService<EventMeshRuntimeHeartbeat> { | ||
| boolean saveOrUpdateByRuntimeAddress(EventMeshRuntimeHeartbeat entity); | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
44 changes: 44 additions & 0 deletions
44
...com/apache/eventmesh/admin/server/web/db/service/impl/EventMeshJobInfoServiceExtImpl.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,44 @@ | ||
| package com.apache.eventmesh.admin.server.web.db.service.impl; | ||
|
|
||
| import com.apache.eventmesh.admin.server.web.db.entity.EventMeshJobDetail; | ||
| import com.apache.eventmesh.admin.server.web.db.entity.EventMeshJobInfo; | ||
| import com.apache.eventmesh.admin.server.web.db.mapper.EventMeshJobInfoMapper; | ||
| import com.apache.eventmesh.admin.server.web.db.service.EventMeshJobInfoExtService; | ||
| import com.baomidou.mybatisplus.core.toolkit.Wrappers; | ||
| import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.apache.eventmesh.common.remote.JobState; | ||
| import org.springframework.stereotype.Service; | ||
|
|
||
| /** | ||
| * @author sodafang | ||
| * @description 针对表【event_mesh_job_info】的数据库操作Service实现 | ||
| * @createDate 2024-05-09 15:51:45 | ||
| */ | ||
| @Service | ||
| @Slf4j | ||
| public class EventMeshJobInfoServiceExtImpl extends ServiceImpl<EventMeshJobInfoMapper, EventMeshJobInfo> | ||
| implements EventMeshJobInfoExtService { | ||
|
|
||
| @Override | ||
| public boolean updateJobState(Integer jobID, JobState state) { | ||
| if (jobID == null || state == null) { | ||
| return false; | ||
| } | ||
| EventMeshJobInfo jobInfo = new EventMeshJobInfo(); | ||
| jobInfo.setJobID(jobID); | ||
| jobInfo.setState(state.ordinal()); | ||
| update(jobInfo, Wrappers.<EventMeshJobInfo>update().notIn("state",JobState.DELETE.ordinal(), | ||
| JobState.COMPLETE.ordinal())); | ||
| return true; | ||
| } | ||
|
|
||
| @Override | ||
| public EventMeshJobDetail getJobDetail(Integer jobID) { | ||
| return null; | ||
| } | ||
| } | ||
|
|
||
|
|
||
|
|
||
|
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
62 changes: 62 additions & 0 deletions
62
...ache/eventmesh/admin/server/web/db/service/impl/EventMeshMysqlPositionServiceExtImpl.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,62 @@ | ||
| package com.apache.eventmesh.admin.server.web.db.service.impl; | ||
|
|
||
| import com.apache.eventmesh.admin.server.web.db.entity.EventMeshMysqlPosition; | ||
| import com.apache.eventmesh.admin.server.web.db.entity.EventMeshPositionReporterHistory; | ||
| import com.apache.eventmesh.admin.server.web.db.mapper.EventMeshMysqlPositionMapper; | ||
| import com.apache.eventmesh.admin.server.web.db.service.EventMeshMysqlPositionExtService; | ||
| import com.apache.eventmesh.admin.server.web.db.service.EventMeshPositionReporterHistoryService; | ||
| import com.baomidou.mybatisplus.core.toolkit.Wrappers; | ||
| import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.apache.eventmesh.common.utils.JsonUtils; | ||
| import org.springframework.beans.factory.annotation.Autowired; | ||
| import org.springframework.stereotype.Service; | ||
|
|
||
| /** | ||
| * @author sodafang | ||
| * @description 针对表【event_mesh_mysql_position】的数据库操作Service实现 | ||
| * @createDate 2024-05-14 17:15:03 | ||
| */ | ||
| @Service | ||
| @Slf4j | ||
| public class EventMeshMysqlPositionServiceExtImpl extends ServiceImpl<EventMeshMysqlPositionMapper, EventMeshMysqlPosition> | ||
| implements EventMeshMysqlPositionExtService { | ||
|
|
||
| @Autowired | ||
| EventMeshPositionReporterHistoryService historyService; | ||
|
|
||
| @Override | ||
| public boolean saveOrUpdateByJob(EventMeshMysqlPosition position) { | ||
| EventMeshMysqlPosition old = getOne(Wrappers.<EventMeshMysqlPosition>query().eq("jobId", position.getJobID())); | ||
| if (old == null) { | ||
| return save(position); | ||
| } else { | ||
| if (old.getPosition() >= position.getPosition()) { | ||
| log.info("job [{}] report position [{}] by runtime [{}] less than db position [{}] by [{}]", | ||
| position.getJobID(), position.getPosition(), position.getAddress(), old.getPosition(), old.getAddress()); | ||
| return true; | ||
| } | ||
| try { | ||
| return update(position, Wrappers.<EventMeshMysqlPosition>update().eq("updateTime", old.getUpdateTime())); | ||
| } finally { | ||
| if (old.getAddress()!= null && !old.getAddress().equals(position.getAddress())) { | ||
| EventMeshPositionReporterHistory history = new EventMeshPositionReporterHistory(); | ||
| history.setRecord(JsonUtils.toJSONString(position)); | ||
| history.setJob(old.getJobID()); | ||
| history.setAddress(old.getAddress()); | ||
| log.info("job [{}] position reporter changed old [{}], now [{}]", position.getJobID(), old, position); | ||
| try { | ||
| historyService.save(history); | ||
| } catch (Exception e) { | ||
| log.warn("save job [{}] mysql position reporter changed history fail, now reporter [{}], old " + | ||
| "[{}]", position.getJobID(), position.getAddress(), old.getAddress(), e); | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
|
|
||
|
|
||
|
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
62 changes: 62 additions & 0 deletions
62
...e/eventmesh/admin/server/web/db/service/impl/EventMeshRuntimeHeartbeatServiceExtImpl.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,62 @@ | ||
| package com.apache.eventmesh.admin.server.web.db.service.impl; | ||
|
|
||
| import com.apache.eventmesh.admin.server.web.db.entity.EventMeshRuntimeHeartbeat; | ||
| import com.apache.eventmesh.admin.server.web.db.entity.EventMeshRuntimeHistory; | ||
| import com.apache.eventmesh.admin.server.web.db.mapper.EventMeshRuntimeHeartbeatMapper; | ||
| import com.apache.eventmesh.admin.server.web.db.service.EventMeshRuntimeHeartbeatExtService; | ||
| import com.apache.eventmesh.admin.server.web.db.service.EventMeshRuntimeHistoryService; | ||
| import com.baomidou.mybatisplus.core.toolkit.Wrappers; | ||
| import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.springframework.beans.factory.annotation.Autowired; | ||
| import org.springframework.stereotype.Service; | ||
|
|
||
| /** | ||
| * @author sodafang | ||
| * @description 针对表【event_mesh_runtime_heartbeat】的数据库操作Service实现 | ||
| * @createDate 2024-05-14 17:15:03 | ||
| */ | ||
| @Service | ||
| @Slf4j | ||
| public class EventMeshRuntimeHeartbeatServiceExtImpl extends ServiceImpl<EventMeshRuntimeHeartbeatMapper, EventMeshRuntimeHeartbeat> | ||
| implements EventMeshRuntimeHeartbeatExtService { | ||
|
|
||
| @Autowired | ||
| EventMeshRuntimeHistoryService historyService; | ||
|
|
||
| @Override | ||
| public boolean saveOrUpdateByRuntimeAddress(EventMeshRuntimeHeartbeat entity) { | ||
| EventMeshRuntimeHeartbeat old = getOne(Wrappers.<EventMeshRuntimeHeartbeat>query().eq("runtimeAddr", | ||
| entity.getRuntimeAddr())); | ||
| if (old == null) { | ||
| return save(entity); | ||
| } else { | ||
| if (Long.parseLong(old.getReportTime()) >= Long.parseLong(entity.getReportTime())) { | ||
| log.info("update heartbeat record ignore, current report time late than db, job " + | ||
| "[{}], remote [{}]", entity.getJobID(), entity.getRuntimeAddr()); | ||
| return true; | ||
| } | ||
| try { | ||
| return update(entity, Wrappers.<EventMeshRuntimeHeartbeat>update().eq("updateTime", old.getUpdateTime())); | ||
| } finally { | ||
| if (old.getJobID() != null && !old.getJobID().equals(entity.getJobID())) { | ||
| EventMeshRuntimeHistory history = new EventMeshRuntimeHistory(); | ||
| history.setAddress(old.getAdminAddr()); | ||
| history.setJob(old.getJobID()); | ||
| try { | ||
| historyService.save(history); | ||
| } catch (Exception e) { | ||
| log.warn("save runtime job changed history fail", e); | ||
| } | ||
|
|
||
| log.info("runtime [{}] changed job, old job [{}], now [{}]",entity.getRuntimeAddr(),old.getJobID(), | ||
| entity.getJobID()); | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
|
|
||
|
|
||
|
|
Oops, something went wrong.