-
Notifications
You must be signed in to change notification settings - Fork 1.3k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[Bugfix]修复删除Kafka集群后，Connect集群任务出现NPE问题 (#1129)
原因： 首先，删除Kafka集群后，没有将DB中的Connect集群进行删除。随后，进行Connect集群指标采集时，由于所在的Kafka集群已经不存在了。最终，导致NPE； 解决： 发布一个Kafka集群删除事件，触发MetaDataService子类，将其在DB中的数据进行删除。 遗留： 当前MetaDataService仅在部分元信息同步类中实现，导致当前DB中的脏数据清理不彻底，后续等MetaDataService在所有元信息同步类中实现后，便可彻底清理数据。 PS：当前修复已保证NPE问题不会再出现。
- Loading branch information
Showing
5 changed files
with
84 additions
and
1 deletion.
There are no files selected for viewing
16 changes: 16 additions & 0 deletions
16
...iaojukeji/know/streaming/km/common/bean/event/cluster/connect/ClusterPhyDeletedEvent.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.xiaojukeji.know.streaming.km.common.bean.event.cluster.connect; | ||
|
|
||
| import com.xiaojukeji.know.streaming.km.common.bean.event.cluster.ClusterPhyBaseEvent; | ||
| import lombok.Getter; | ||
|
|
||
| /** | ||
| * 集群删除事件 | ||
| * @author zengqiao | ||
| * @date 23/08/15 | ||
| */ | ||
| @Getter | ||
| public class ClusterPhyDeletedEvent extends ClusterPhyBaseEvent { | ||
| public ClusterPhyDeletedEvent(Object source, Long clusterPhyId) { | ||
| super(source, clusterPhyId); | ||
| } | ||
| } |
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
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
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
53 changes: 53 additions & 0 deletions
53
...va/com/xiaojukeji/know/streaming/km/task/service/listener/TaskClusterDeletedListener.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,53 @@ | ||
| package com.xiaojukeji.know.streaming.km.task.service.listener; | ||
|
|
||
| import com.didiglobal.logi.log.ILog; | ||
| import com.didiglobal.logi.log.LogFactory; | ||
| import com.xiaojukeji.know.streaming.km.common.bean.event.cluster.connect.ClusterPhyDeletedEvent; | ||
| import com.xiaojukeji.know.streaming.km.common.component.SpringTool; | ||
| import com.xiaojukeji.know.streaming.km.common.utils.BackoffUtils; | ||
| import com.xiaojukeji.know.streaming.km.common.utils.FutureUtil; | ||
| import com.xiaojukeji.know.streaming.km.core.service.meta.MetaDataService; | ||
| import org.springframework.context.ApplicationListener; | ||
| import org.springframework.stereotype.Service; | ||
|
|
||
| @Service | ||
| public class TaskClusterDeletedListener implements ApplicationListener<ClusterPhyDeletedEvent> { | ||
| private static final ILog LOGGER = LogFactory.getLog(TaskClusterDeletedListener.class); | ||
|
|
||
| @Override | ||
| public void onApplicationEvent(ClusterPhyDeletedEvent event) { | ||
| LOGGER.info("method=onApplicationEvent||clusterPhyId={}||msg=listened delete cluster", event.getClusterPhyId()); | ||
|
|
||
| // 交由KS自定义的线程池，异步执行任务 | ||
| FutureUtil.quickStartupFutureUtil.submitTask( | ||
| () -> { | ||
| // 延迟60秒，避免正在运行的任务，将数据写入DB中 | ||
| BackoffUtils.backoff(60000); | ||
|
|
||
| for (MetaDataService metaDataService: SpringTool.getBeansOfType(MetaDataService.class).values()) { | ||
| LOGGER.info( | ||
| "method=onApplicationEvent||clusterPhyId={}||className={}||msg=delete cluster data in db starting", | ||
| event.getClusterPhyId(), metaDataService.getClass().getSimpleName() | ||
| ); | ||
|
|
||
| try { | ||
| // 删除数据 | ||
| metaDataService.deleteInDBByKafkaClusterId(event.getClusterPhyId()); | ||
|
|
||
| LOGGER.info( | ||
| "method=onApplicationEvent||clusterPhyId={}||className={}||msg=delete cluster data in db finished", | ||
| event.getClusterPhyId(), metaDataService.getClass().getSimpleName() | ||
| ); | ||
| } catch (Exception e) { | ||
| LOGGER.error( | ||
| "method=onApplicationEvent||clusterPhyId={}||className={}||msg=delete cluster data in db failed||errMsg=exception", | ||
| event.getClusterPhyId(), metaDataService.getClass().getSimpleName(), e | ||
| ); | ||
| } | ||
| } | ||
| } | ||
| ); | ||
|
|
||
|
|
||
| } | ||
| } |