(rocketmq-eventbridge) branch feat/runtime-manager updated: feat:watch the cluster and worker.

shenlin Sun, 26 Nov 2023 03:32:28 -0800

This is an automated email from the ASF dual-hosted git repository.

shenlin pushed a commit to branch feat/runtime-manager
in repository https://gitbox.apache.org/repos/asf/rocketmq-eventbridge.git



The following commit(s) were added to refs/heads/feat/runtime-manager by this 
push:
     new 81f7bc3  feat:watch the cluster and worker.
81f7bc3 is described below

commit 81f7bc3ae92115039bffb1f8cf9671ed5073b56a
Author: 2011shenlin <[email protected]>
AuthorDate: Sun Nov 26 19:31:43 2023 +0800

    feat:watch the cluster and worker.
---
 adapter/persistence/pom.xml                        |   5 +
 .../mybatis/mapper/EventClusterMapper.java}        |  10 +-
 .../repository/MybatisClusterRepository.java}      |  25 +-
 .../worker/mybatis/mapper/EventWorkerMapper.java}  |  21 +-
 .../repository/MybatisWorkerRepository.java        |  54 ++
 .../resources/db/migration/V10__init_cluster.sql   |  30 +
 .../resources/db/migration/V11__init_worker.sql    |  32 +
 .../main/resources/mybatis/EventClusterMapper.xml  |  52 ++
 .../main/resources/mybatis/EventWorkerMapper.xml   |  63 ++
 adapter/pom.xml                                    |   1 +
 .../adapter/runtime/manager/cluster/Cluster.java   |   7 +-
 .../runtime/manager/cluster/ClusterService.java    |  10 +-
 .../manager/repository/ClusterRepository.java      |   2 +-
 .../manager/repository/WorkerRepository.java       |   4 +-
 .../runtime/manager/scale/ClusterWorkerScale.java  |   2 +
 .../runtime/manager/watch/WatchCluster.java        |  48 +-
 .../adapter/runtime/manager/watch/WatchWorker.java |  37 +-
 .../adapter/runtime/manager/worker/Worker.java     |  12 +-
 .../runtime/manager/worker/WorkerService.java      |  20 +-
 pom.xml                                            |  10 +-
 rocketmq_eventbridge.mv.db                         | Bin 81920 -> 81920 bytes
 rocketmq_eventbridge.trace.db                      | 657 +++------------------
 scan.py                                            |   0
 start/pom.xml                                      |  13 +-
 .../rocketmq/eventbridge/demo/DefaultDemo.java     |   6 +-
 25 files changed, 467 insertions(+), 654 deletions(-)

diff --git a/adapter/persistence/pom.xml b/adapter/persistence/pom.xml
index eac0c8e..b206a1d 100644
--- a/adapter/persistence/pom.xml
+++ b/adapter/persistence/pom.xml
@@ -71,6 +71,11 @@
             <groupId>org.apache.rocketmq</groupId>
             <artifactId>rocketmq-tools</artifactId>
         </dependency>
+        <dependency>
+            
<artifactId>rocketmq-eventbridge-adapter-runtime-manager</artifactId>
+            <groupId>org.apache.rocketmq</groupId>
+            <version>1.0.0</version>
+        </dependency>
         <dependency>
             <groupId>org.flywaydb</groupId>
             <artifactId>flyway-core</artifactId>
diff --git 
a/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/repository/ClusterRepository.java
 
b/adapter/persistence/src/main/java/org/apache/rocketmq/eventbridge/adapter/persistence/cluster/mybatis/mapper/EventClusterMapper.java
similarity index 79%
copy from 
adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/repository/ClusterRepository.java
copy to 
adapter/persistence/src/main/java/org/apache/rocketmq/eventbridge/adapter/persistence/cluster/mybatis/mapper/EventClusterMapper.java
index 38c1c58..ea62dc5 100644
--- 
a/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/repository/ClusterRepository.java
+++ 
b/adapter/persistence/src/main/java/org/apache/rocketmq/eventbridge/adapter/persistence/cluster/mybatis/mapper/EventClusterMapper.java
@@ -15,14 +15,14 @@
  *  limitations under the License.
  */
 
-package org.apache.rocketmq.eventbridge.adapter.runtime.manager.repository;
+package 
org.apache.rocketmq.eventbridge.adapter.persistence.cluster.mybatis.mapper;
 
 import java.util.List;
+import org.apache.ibatis.annotations.Param;
 import org.apache.rocketmq.eventbridge.adapter.runtime.manager.cluster.Cluster;
 
-public interface ClusterRepository {
-
+public interface EventClusterMapper {
     List<Cluster> listCluster();
 
-    boolean updateCluster(Cluster cluster);
-}
+    int updateCluster(@Param("id") int id, @Param("md5") String md5);
+}
\ No newline at end of file
diff --git 
a/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/repository/ClusterRepository.java
 
b/adapter/persistence/src/main/java/org/apache/rocketmq/eventbridge/adapter/persistence/cluster/mybatis/repository/MybatisClusterRepository.java
similarity index 51%
copy from 
adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/repository/ClusterRepository.java
copy to 
adapter/persistence/src/main/java/org/apache/rocketmq/eventbridge/adapter/persistence/cluster/mybatis/repository/MybatisClusterRepository.java
index 38c1c58..19ad2da 100644
--- 
a/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/repository/ClusterRepository.java
+++ 
b/adapter/persistence/src/main/java/org/apache/rocketmq/eventbridge/adapter/persistence/cluster/mybatis/repository/MybatisClusterRepository.java
@@ -15,14 +15,29 @@
  *  limitations under the License.
  */
 
-package org.apache.rocketmq.eventbridge.adapter.runtime.manager.repository;
+package 
org.apache.rocketmq.eventbridge.adapter.persistence.cluster.mybatis.repository;
 
 import java.util.List;
+import 
org.apache.rocketmq.eventbridge.adapter.persistence.cluster.mybatis.mapper.EventClusterMapper;
 import org.apache.rocketmq.eventbridge.adapter.runtime.manager.cluster.Cluster;
+import 
org.apache.rocketmq.eventbridge.adapter.runtime.manager.repository.ClusterRepository;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Repository;
 
-public interface ClusterRepository {
+@Repository
+public class MybatisClusterRepository implements ClusterRepository {
 
-    List<Cluster> listCluster();
+    @Autowired
+    EventClusterMapper eventClusterMapper;
 
-    boolean updateCluster(Cluster cluster);
-}
+    @Override public List<Cluster> listCluster() {
+        List<Cluster> clusterDOList = eventClusterMapper.listCluster();
+        return clusterDOList;
+    }
+
+    @Override
+    public boolean updateCluster(int clusterId, String md5) {
+        return eventClusterMapper.updateCluster(clusterId,md5)==1;
+    }
+
+}
\ No newline at end of file
diff --git 
a/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/repository/WorkerRepository.java
 
b/adapter/persistence/src/main/java/org/apache/rocketmq/eventbridge/adapter/persistence/worker/mybatis/mapper/EventWorkerMapper.java
similarity index 62%
copy from 
adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/repository/WorkerRepository.java
copy to 
adapter/persistence/src/main/java/org/apache/rocketmq/eventbridge/adapter/persistence/worker/mybatis/mapper/EventWorkerMapper.java
index c35fefb..d7a27d7 100644
--- 
a/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/repository/WorkerRepository.java
+++ 
b/adapter/persistence/src/main/java/org/apache/rocketmq/eventbridge/adapter/persistence/worker/mybatis/mapper/EventWorkerMapper.java
@@ -15,21 +15,18 @@
  *  limitations under the License.
  */
 
-package org.apache.rocketmq.eventbridge.adapter.runtime.manager.repository;
+package 
org.apache.rocketmq.eventbridge.adapter.persistence.worker.mybatis.mapper;
 
 import java.util.List;
+import org.apache.ibatis.annotations.Param;
 import org.apache.rocketmq.eventbridge.adapter.runtime.manager.worker.Worker;
 
-public interface WorkerRepository {
+public interface EventWorkerMapper {
+    List<Worker> listWorkersByCluster(@Param("clusterId") int clusterId);
 
-    List<Worker> listWorkers(int clusterId);
+    List<Worker> listWorkers();
 
-    boolean createWorker(Worker worker);
-
-    boolean updateWorker(Worker worker);
-
-    boolean deleteWorker(int workerId);
-
-    boolean deleteWorkers(int clusterId);
-
-}
+    int createWorker(@Param("id") String id, @Param("clusterId") int 
clusterId, @Param("name") String name,
+        @Param("image") String image, @Param("resources") String resources, 
@Param("config") String config,
+        @Param("status") String status, @Param("md5") String md5);
+}
\ No newline at end of file
diff --git 
a/adapter/persistence/src/main/java/org/apache/rocketmq/eventbridge/adapter/persistence/worker/mybatis/repository/MybatisWorkerRepository.java
 
b/adapter/persistence/src/main/java/org/apache/rocketmq/eventbridge/adapter/persistence/worker/mybatis/repository/MybatisWorkerRepository.java
new file mode 100644
index 0000000..4c4a2ee
--- /dev/null
+++ 
b/adapter/persistence/src/main/java/org/apache/rocketmq/eventbridge/adapter/persistence/worker/mybatis/repository/MybatisWorkerRepository.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+package 
org.apache.rocketmq.eventbridge.adapter.persistence.worker.mybatis.repository;
+
+import java.util.List;
+import 
org.apache.rocketmq.eventbridge.adapter.persistence.worker.mybatis.mapper.EventWorkerMapper;
+import 
org.apache.rocketmq.eventbridge.adapter.runtime.manager.repository.WorkerRepository;
+import org.apache.rocketmq.eventbridge.adapter.runtime.manager.worker.Worker;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Repository;
+
+@Repository public class MybatisWorkerRepository implements WorkerRepository {
+
+    @Autowired EventWorkerMapper eventWorkerMapper;
+
+    @Override public List<Worker> listWorkersByCluster(int clusterId) {
+        return eventWorkerMapper.listWorkersByCluster(clusterId);
+    }
+
+    @Override public List<Worker> listWorkers() {
+        return eventWorkerMapper.listWorkers();
+    }
+
+    @Override public boolean createWorker(Worker worker) {
+        return eventWorkerMapper.createWorker(worker.getId(), 
worker.getClusterId(), worker.getName(), worker.getImage(), 
worker.getResources(), worker.getConfig(), worker.getStatus(), worker.getMd5()) 
== 1;
+    }
+
+    @Override public boolean updateWorker(Worker worker) {
+        return false;
+    }
+
+    @Override public boolean deleteWorker(int workerId) {
+        return false;
+    }
+
+    @Override public boolean deleteWorkers(int clusterId) {
+        return false;
+    }
+}
\ No newline at end of file
diff --git 
a/adapter/persistence/src/main/resources/db/migration/V10__init_cluster.sql 
b/adapter/persistence/src/main/resources/db/migration/V10__init_cluster.sql
new file mode 100644
index 0000000..216b98b
--- /dev/null
+++ b/adapter/persistence/src/main/resources/db/migration/V10__init_cluster.sql
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+CREATE TABLE IF NOT EXISTS  `event_cluster` (
+  `id` int(128) unsigned NOT NULL AUTO_INCREMENT COMMENT 'cluster id',
+  `name` varchar(128) DEFAULT NULL COMMENT 'cluster name',
+  `resources` varchar(1024) DEFAULT NULL COMMENT 'the spec of worker on 
cluster',
+  `replica` int(11) DEFAULT '2' COMMENT 'the replica of cluster',
+  `image` varchar(1024) DEFAULT NULL COMMENT 'the image of cluster',
+  `md5` varchar(4096) DEFAULT NULL,
+  `gmt_create` datetime DEFAULT NULL COMMENT 'create time',
+  `gmt_modify` datetime DEFAULT NULL COMMENT 'modify time',
+  PRIMARY KEY (`id`),
+  UNIQUE KEY `event_cluster_index_unique_name` (`name`)
+) ENGINE=InnoDB  DEFAULT CHARSET=utf8
+;
\ No newline at end of file
diff --git 
a/adapter/persistence/src/main/resources/db/migration/V11__init_worker.sql 
b/adapter/persistence/src/main/resources/db/migration/V11__init_worker.sql
new file mode 100644
index 0000000..49c9a43
--- /dev/null
+++ b/adapter/persistence/src/main/resources/db/migration/V11__init_worker.sql
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+CREATE TABLE `event_worker` (
+  `id` varchar(128) NOT NULL,
+  `cluster_id` varchar(128) DEFAULT NULL COMMENT 'the cluster id of worker',
+  `name` varchar(128) DEFAULT NULL COMMENT 'worker name',
+  `image` varchar(1024) DEFAULT NULL COMMENT 'the image of cluster',
+  `resources` varchar(1024) DEFAULT NULL COMMENT 'the spec of worker',
+  `config` text DEFAULT NULL COMMENT 'the config of worker',
+  `status` varchar(128) DEFAULT NULL COMMENT 'Worker name',
+  `gmt_create` datetime DEFAULT NULL COMMENT 'create time',
+  `gmt_modify` datetime DEFAULT NULL COMMENT 'modify time',
+  `md5` varchar(128) DEFAULT NULL COMMENT 'worker config md5',
+  PRIMARY KEY (`id`),
+  UNIQUE KEY `connect_worker_name_index`(`name`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8
+;
diff --git 
a/adapter/persistence/src/main/resources/mybatis/EventClusterMapper.xml 
b/adapter/persistence/src/main/resources/mybatis/EventClusterMapper.xml
new file mode 100644
index 0000000..77b816c
--- /dev/null
+++ b/adapter/persistence/src/main/resources/mybatis/EventClusterMapper.xml
@@ -0,0 +1,52 @@
+<!-- Licensed to the Apache Software Foundation (ASF) under one or more 
contributor
+       license agreements. See the NOTICE file distributed with this work for 
additional
+       information regarding copyright ownership. The ASF licenses this file to
+       You under the Apache License, Version 2.0 (the "License"); you may not 
use
+       this file except in compliance with the License. You may obtain a copy 
of
+       the License at http://www.apache.org/licenses/LICENSE-2.0 Unless 
required
+       by applicable law or agreed to in writing, software distributed under 
the
+       License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR 
CONDITIONS
+       OF ANY KIND, either express or implied. See the License for the specific
+       language governing permissions and limitations under the License. -->
+
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" 
"http://mybatis.org/dtd/mybatis-3-mapper.dtd"; >
+<mapper 
namespace="org.apache.rocketmq.eventbridge.adapter.persistence.cluster.mybatis.mapper.EventClusterMapper">
+
+    <sql id="TABLE_NAME">
+        event_cluster
+    </sql>
+
+    <sql id="COLUMN_LIST">
+        id,name,resources,replica,image,md5,gmt_create,gmt_modify
+    </sql>
+    <resultMap id="resultMap" 
type="org.apache.rocketmq.eventbridge.adapter.runtime.manager.cluster.Cluster">
+        <result property="id" column="id"/>
+        <result property="name" column="name"/>
+        <result property="resources" column="resources"/>
+        <result property="replica" column="replica"/>
+        <result property="image" column="image"/>
+        <result property="md5" column="md5"/>
+        <result property="gmtCreate" column="gmt_create"/>
+        <result property="gmtModify" column="gmt_modify"/>
+    </resultMap>
+
+    <select id="listCluster" resultMap="resultMap">
+        SELECT
+        <include refid="COLUMN_LIST"/>
+        FROM
+        <include refid="TABLE_NAME"/>
+    </select>
+
+
+    <update id="updateCluster" parameterType="java.util.Map">
+        UPDATE
+        <include refid="TABLE_NAME"/>
+        <set>
+            gmt_modify = now()
+            <if test="md5 != null">
+                ,md5= #{md5}
+            </if>
+        </set>
+        WHERE id = #{id}
+    </update>
+</mapper>
\ No newline at end of file
diff --git 
a/adapter/persistence/src/main/resources/mybatis/EventWorkerMapper.xml 
b/adapter/persistence/src/main/resources/mybatis/EventWorkerMapper.xml
new file mode 100644
index 0000000..730df31
--- /dev/null
+++ b/adapter/persistence/src/main/resources/mybatis/EventWorkerMapper.xml
@@ -0,0 +1,63 @@
+<!-- Licensed to the Apache Software Foundation (ASF) under one or more 
contributor
+       license agreements. See the NOTICE file distributed with this work for 
additional
+       information regarding copyright ownership. The ASF licenses this file to
+       You under the Apache License, Version 2.0 (the "License"); you may not 
use
+       this file except in compliance with the License. You may obtain a copy 
of
+       the License at http://www.apache.org/licenses/LICENSE-2.0 Unless 
required
+       by applicable law or agreed to in writing, software distributed under 
the
+       License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR 
CONDITIONS
+       OF ANY KIND, either express or implied. See the License for the specific
+       language governing permissions and limitations under the License. -->
+
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" 
"http://mybatis.org/dtd/mybatis-3-mapper.dtd"; >
+<mapper 
namespace="org.apache.rocketmq.eventbridge.adapter.persistence.worker.mybatis.mapper.EventWorkerMapper">
+
+    <sql id="TABLE_NAME">
+        event_worker
+    </sql>
+
+    <sql id="COLUMN_LIST">
+        
id,cluster_id,name,image,resources,config,status,md5,gmt_create,gmt_modify
+    </sql>
+
+
+    <resultMap id="resultMap" 
type="org.apache.rocketmq.eventbridge.adapter.runtime.manager.worker.Worker">
+        <result property="id" column="id"/>
+        <result property="clusterId" column="cluster_id"/>
+        <result property="name" column="name"/>
+        <result property="image" column="image"/>
+        <result property="resources" column="resources"/>
+        <result property="config" column="config"/>
+        <result property="status" column="status"/>
+        <result property="md5" column="md5"/>
+        <result property="gmtCreate" column="gmt_create"/>
+        <result property="gmtModify" column="gmt_modify"/>
+    </resultMap>
+
+    <select id="listWorkersByCluster" resultMap="resultMap">
+        SELECT
+        <include refid="COLUMN_LIST"/>
+        FROM
+        <include refid="TABLE_NAME"/>
+        WHERE cluster_id = #{clusterId}
+    </select>
+
+    <select id="listWorkers" resultMap="resultMap">
+        SELECT
+        <include refid="COLUMN_LIST"/>
+        FROM
+        <include refid="TABLE_NAME"/>
+    </select>
+
+
+    <insert id="createWorker" parameterType="java.util.Map">
+        INSERT INTO
+        <include refid="TABLE_NAME"/>
+        (
+        <include refid="COLUMN_LIST"/>
+        )
+        VALUES
+        
(#{id},#{clusterId},#{name},#{image},#{resources},#{config},#{status},#{md5},now(),now())
+    </insert>
+
+</mapper>
\ No newline at end of file
diff --git a/adapter/pom.xml b/adapter/pom.xml
index dc61d42..010492c 100644
--- a/adapter/pom.xml
+++ b/adapter/pom.xml
@@ -28,6 +28,7 @@
         <module>runtime</module>
         <module>storage</module>
         <module>benchmark</module>
+        <module>runtime-manager</module>
     </modules>
 
 </project>
\ No newline at end of file
diff --git 
a/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/cluster/Cluster.java
 
b/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/cluster/Cluster.java
index 2e0cb83..340aeee 100644
--- 
a/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/cluster/Cluster.java
+++ 
b/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/cluster/Cluster.java
@@ -28,9 +28,8 @@ public class Cluster {
     private String name;
     private String resources;
     private int replica;
-    private String currentImageId;
-    private String expectImageId;
+    private String image;
     private String md5;
-    private Date ctime;
-    private Date mtime;
+    private Date gmtCreate;
+    private Date gmtModify;
 }
\ No newline at end of file
diff --git 
a/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/cluster/ClusterService.java
 
b/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/cluster/ClusterService.java
index 4e51c3b..7ac59ca 100644
--- 
a/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/cluster/ClusterService.java
+++ 
b/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/cluster/ClusterService.java
@@ -21,9 +21,13 @@ import com.google.common.base.Strings;
 import java.util.List;
 import org.apache.commons.codec.digest.DigestUtils;
 import 
org.apache.rocketmq.eventbridge.adapter.runtime.manager.repository.ClusterRepository;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
 
+@Service
 public class ClusterService {
 
+    @Autowired
     private ClusterRepository clusterRepository;
 
     public boolean createCluster() {
@@ -43,7 +47,7 @@ public class ClusterService {
     }
 
     public String calMD5(Cluster cluster) {
-        String str = cluster.getName() + cluster.getResources() + 
cluster.getReplica() + cluster.getExpectImageId();
+        String str = cluster.getName() + cluster.getResources() + 
cluster.getReplica() + cluster.getImage();
         return DigestUtils.md5Hex(str);
     }
 
@@ -56,8 +60,8 @@ public class ClusterService {
         }
     }
 
-    public boolean refreshMD5(Cluster cluster) {
-        return true;
+    public boolean refreshMD5(int clusterId, String md5) {
+        return clusterRepository.updateCluster(clusterId, md5);
     }
 
     public int calLoad(Cluster cluster) {
diff --git 
a/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/repository/ClusterRepository.java
 
b/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/repository/ClusterRepository.java
index 38c1c58..9ef88bb 100644
--- 
a/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/repository/ClusterRepository.java
+++ 
b/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/repository/ClusterRepository.java
@@ -24,5 +24,5 @@ public interface ClusterRepository {
 
     List<Cluster> listCluster();
 
-    boolean updateCluster(Cluster cluster);
+    boolean updateCluster(int clusterId, String md5);
 }
diff --git 
a/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/repository/WorkerRepository.java
 
b/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/repository/WorkerRepository.java
index c35fefb..0ecb786 100644
--- 
a/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/repository/WorkerRepository.java
+++ 
b/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/repository/WorkerRepository.java
@@ -22,7 +22,9 @@ import 
org.apache.rocketmq.eventbridge.adapter.runtime.manager.worker.Worker;
 
 public interface WorkerRepository {
 
-    List<Worker> listWorkers(int clusterId);
+    List<Worker> listWorkersByCluster(int clusterId);
+
+    List<Worker> listWorkers();
 
     boolean createWorker(Worker worker);
 
diff --git 
a/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/scale/ClusterWorkerScale.java
 
b/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/scale/ClusterWorkerScale.java
index 1b48daf..be22fa0 100644
--- 
a/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/scale/ClusterWorkerScale.java
+++ 
b/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/scale/ClusterWorkerScale.java
@@ -25,7 +25,9 @@ import org.apache.rocketmq.common.ThreadFactoryImpl;
 import org.apache.rocketmq.eventbridge.adapter.runtime.manager.cluster.Cluster;
 import 
org.apache.rocketmq.eventbridge.adapter.runtime.manager.cluster.ClusterService;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
 
+@Service
 public class ClusterWorkerScale {
     @Autowired
     ClusterService clusterService;
diff --git 
a/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/watch/WatchCluster.java
 
b/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/watch/WatchCluster.java
index bd77c83..bbe270e 100644
--- 
a/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/watch/WatchCluster.java
+++ 
b/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/watch/WatchCluster.java
@@ -17,46 +17,72 @@
 
 package org.apache.rocketmq.eventbridge.adapter.runtime.manager.watch;
 
+import com.google.common.base.Strings;
 import java.util.List;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
+import javax.annotation.PostConstruct;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.rocketmq.common.ThreadFactoryImpl;
 import org.apache.rocketmq.eventbridge.adapter.runtime.manager.cluster.Cluster;
 import 
org.apache.rocketmq.eventbridge.adapter.runtime.manager.cluster.ClusterService;
+import org.apache.rocketmq.eventbridge.adapter.runtime.manager.worker.Worker;
 import 
org.apache.rocketmq.eventbridge.adapter.runtime.manager.worker.WorkerService;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
+@Service
+@Slf4j
 public class WatchCluster {
 
     @Autowired
     ClusterService clusterService;
 
+    @Autowired
+    WorkerService workerService;
+
     private final ScheduledExecutorService scheduledExecutorService = 
Executors.newSingleThreadScheduledExecutor(
         new ThreadFactoryImpl(WatchCluster.class.getSimpleName()));
 
+    public WatchCluster(ClusterService clusterService) {
+        this.clusterService = clusterService;
+    }
+
+    @PostConstruct
     public void start() {
         this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
             @Override
             public void run() {
-                List<Cluster> clusters = clusterService.listCluster();
-                clusters.forEach(cluster -> {
-                    if (!clusterService.isFinalState(cluster)) {
-                        watchTheClusterReplica(cluster);
-                        watchTheClusterImageId(cluster);
-                        watchTheClusterResources(cluster);
-                        clusterService.refreshMD5(cluster);
-                    }
+                try {
+                    List<Cluster> clusters = clusterService.listCluster();
+                    clusters.forEach(cluster -> {
+                        String md5 = clusterService.calMD5(cluster);
+                        if (!Strings.isNullOrEmpty(md5) && 
!md5.equals(cluster.getMd5())) {
+                            watchTheClusterReplica(cluster);
+                            watchTheClusterImageId(cluster);
+                            watchTheClusterResources(cluster);
+                            clusterService.refreshMD5(cluster.getId(), md5);
+                        }
 
-                });
+                    });
+                } catch (Throwable e) {
+                    log.error("WatchCluster failed.", e);
+                }
             }
-        }, 3, 60, TimeUnit.SECONDS);
+        }, 3, 5, TimeUnit.SECONDS);
     }
 
     @Transactional
     private void watchTheClusterReplica(Cluster cluster) {
-
+        List<Worker> workers = 
workerService.listWorkersByCluster(cluster.getId());
+        if (workers != null && cluster.getReplica() > workers.size()) {
+            IntStream.range(workers.size(), 
cluster.getReplica()).forEach(value -> {
+                workerService.createWorker(cluster.getId(), "worker-" + value, 
cluster.getImage(), cluster.getResources(), null, null);
+            });
+        }
     }
 
     @Transactional
diff --git 
a/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/watch/WatchWorker.java
 
b/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/watch/WatchWorker.java
index c7c2dc8..c00a694 100644
--- 
a/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/watch/WatchWorker.java
+++ 
b/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/watch/WatchWorker.java
@@ -24,39 +24,50 @@ import java.util.Map;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import javax.annotation.PostConstruct;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.rocketmq.common.ThreadFactoryImpl;
 import 
org.apache.rocketmq.eventbridge.adapter.runtime.manager.repository.WorkerInstanceRepository;
 import org.apache.rocketmq.eventbridge.adapter.runtime.manager.worker.Worker;
 import 
org.apache.rocketmq.eventbridge.adapter.runtime.manager.worker.WorkerResource;
 import 
org.apache.rocketmq.eventbridge.adapter.runtime.manager.worker.WorkerService;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
 
+@Service
+@Slf4j
 public class WatchWorker {
 
     @Autowired
     WorkerService workerService;
-    @Autowired
-    WorkerInstanceRepository workerInstanceRepository;
+//    @Autowired
+//    WorkerInstanceRepository workerInstanceRepository;
 
     private final ScheduledExecutorService scheduledExecutorService = 
Executors.newSingleThreadScheduledExecutor(
         new ThreadFactoryImpl(WatchWorker.class.getSimpleName()));
 
+    @PostConstruct
     public void start() {
         this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
             @Override
             public void run() {
-                List<Worker> workers = workerService.listWorkers();
-                workers.forEach(worker -> {
-                    if (!workerService.isFinalState(worker)) {
-                        Map<String, Object> environments = new 
Gson().fromJson(worker.getConfig(), new TypeToken<Map<String, Object>>() {
-                        }.getType());
-                        WorkerResource workerResource = new 
Gson().fromJson(worker.getResources(), WorkerResource.class);
-                        
workerInstanceRepository.applyWorkerInstance(worker.getName(), 
worker.getImageTag(), workerResource, environments);
-                        workerService.refreshMD5(worker);
-                    }
+                try {
+                    List<Worker> workers = workerService.listWorkers();
+                    workers.forEach(worker -> {
+                        if (!workerService.isFinalState(worker)) {
+                            Map<String, Object> environments = new 
Gson().fromJson(worker.getConfig(), new TypeToken<Map<String, Object>>() {
+                            }.getType());
+//                            WorkerResource workerResource = new 
Gson().fromJson(worker.getResources(), WorkerResource.class);
+                            log.info("applyWorkerInstance, workerName: {}, 
workerImageTag: {}, workerResource: {}, environments: {}", worker.getName(), 
worker.getImage(), worker.getResources(), new Gson().toJson(environments));
+//                        
workerInstanceRepository.applyWorkerInstance(worker.getName(), 
worker.getImageTag(), workerResource, environments);
+                            workerService.refreshMD5(worker);
+                        }
 
-                });
+                    });
+                } catch (Throwable e) {
+                    log.error("WatchWorker failed.", e);
+                }
             }
-        }, 3, 60, TimeUnit.SECONDS);
+        }, 3, 5, TimeUnit.SECONDS);
     }
 }
\ No newline at end of file
diff --git 
a/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/worker/Worker.java
 
b/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/worker/Worker.java
index 96aba43..e47e4aa 100644
--- 
a/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/worker/Worker.java
+++ 
b/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/worker/Worker.java
@@ -18,20 +18,22 @@
 package org.apache.rocketmq.eventbridge.adapter.runtime.manager.worker;
 
 import java.util.Date;
+import lombok.Builder;
 import lombok.Data;
 import lombok.EqualsAndHashCode;
 
 @EqualsAndHashCode(of = {"name"})
 @Data
+@Builder
 public class Worker {
-    private int id;
-    private String clusterId;
+    private String id;
+    private int clusterId;
     private String name;
-    private String imageTag;
+    private String image;
     private String resources;
     private String config;
     private String status;
-    private Date ctime;
-    private Date mtime;
     private String md5;
+    private Date gmtCreate;
+    private Date gmtModify;
 }
\ No newline at end of file
diff --git 
a/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/worker/WorkerService.java
 
b/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/worker/WorkerService.java
index 5709c62..d1d4eb9 100644
--- 
a/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/worker/WorkerService.java
+++ 
b/adapter/runtime-manager/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/manager/worker/WorkerService.java
@@ -18,19 +18,31 @@
 package org.apache.rocketmq.eventbridge.adapter.runtime.manager.worker;
 
 import java.util.List;
+import java.util.UUID;
+import 
org.apache.rocketmq.eventbridge.adapter.runtime.manager.repository.WorkerRepository;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
 
-public class WorkerService {
+@Service public class WorkerService {
 
-    private boolean createWorker() {
-        return true;
+    @Autowired private WorkerRepository workerRepository;
+
+    public boolean createWorker(int clusterId, String workerName, String 
image, String resources, String config,
+        String status) {
+        Worker worker = 
Worker.builder().id(UUID.randomUUID().toString()).clusterId(clusterId).name(workerName).image(image).resources(resources).config(config).status(status).build();
+        return workerRepository.createWorker(worker);
     }
 
     private boolean applyTasksToWorker() {
         return true;
     }
 
+    public List<Worker> listWorkersByCluster(int clusterId) {
+        return workerRepository.listWorkersByCluster(clusterId);
+    }
+
     public List<Worker> listWorkers() {
-        return null;
+        return workerRepository.listWorkers();
     }
 
     public boolean isFinalState(Worker worker) {
diff --git a/pom.xml b/pom.xml
index f3d1bcb..41ee0f3 100644
--- a/pom.xml
+++ b/pom.xml
@@ -151,11 +151,11 @@
                 <artifactId>rocketmq-eventbridge-infrastructure</artifactId>
                 <version>${project.version}</version>
             </dependency>
-            <dependency>
-                <groupId>org.apache.rocketmq</groupId>
-                <artifactId>rocketmq-test-demo</artifactId>
-                <version>${project.version}</version>
-            </dependency>
+<!--            <dependency>-->
+<!--                <groupId>org.apache.rocketmq</groupId>-->
+<!--                <artifactId>rocketmq-test-demo</artifactId>-->
+<!--                <version>${project.version}</version>-->
+<!--            </dependency>-->
 
             <!-- Framework -->
             <dependency>
diff --git a/rocketmq_eventbridge.mv.db b/rocketmq_eventbridge.mv.db
index e5daf95..1a098df 100644
Binary files a/rocketmq_eventbridge.mv.db and b/rocketmq_eventbridge.mv.db 
differ
diff --git a/rocketmq_eventbridge.trace.db b/rocketmq_eventbridge.trace.db
index 8db537c..bcce0ee 100644
--- a/rocketmq_eventbridge.trace.db
+++ b/rocketmq_eventbridge.trace.db
@@ -1,4 +1,4 @@
-2023-06-12 15:47:31 jdbc[3]: exception
+2023-11-26 18:41:02 jdbc[3]: exception
 org.h2.jdbc.JdbcSQLException: Table "EVENT_TARGET_RUNNER" not found; SQL 
statement:
 SELECT
          
@@ -10,584 +10,85 @@ SELECT
      
         WHERE
         id>=0 [42102-197]
-2023-06-25 10:24:37 jdbc[13370]: exception
-2023-08-13 15:17:19 jdbc[18]: exception
-org.h2.jdbc.JdbcSQLException: Database is already closed (to disable automatic 
closing at VM shutdown, add ";DB_CLOSE_ON_EXIT=FALSE" to the db URL) [90121-197]
+2023-11-26 18:42:05 jdbc[8]: exception
+org.h2.jdbc.JdbcSQLException: Cannot parse "TIMESTAMP" constant ; SQL 
statement:
+INSERT INTO 
"PUBLIC"."EVENT_CLUSTER"("ID","NAME","RESOURCES","REPLICA","IMAGE","MD5","GMT_CREATE","GMT_MODIFY")VALUES(?,?,?,?,?,?,?,?)
 -- (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8) [22007-197]
        at org.h2.message.DbException.getJdbcSQLException(DbException.java:357)
-       at org.h2.message.DbException.get(DbException.java:179)
-       at org.h2.message.DbException.get(DbException.java:155)
-       at org.h2.message.DbException.get(DbException.java:144)
-       at org.h2.jdbc.JdbcConnection.checkClosed(JdbcConnection.java:1526)
-       at org.h2.jdbc.JdbcConnection.checkClosed(JdbcConnection.java:1502)
-       at org.h2.jdbc.JdbcConnection.createStatement(JdbcConnection.java:217)
-       at com.zaxxer.hikari.pool.PoolBase.isConnectionAlive(PoolBase.java:153)
-       at com.zaxxer.hikari.pool.HikariPool.getConnection(HikariPool.java:172)
-       at com.zaxxer.hikari.pool.HikariPool.getConnection(HikariPool.java:148)
-       at 
com.zaxxer.hikari.HikariDataSource.getConnection(HikariDataSource.java:100)
-       at 
org.springframework.jdbc.datasource.DataSourceUtils.fetchConnection(DataSourceUtils.java:151)
-       at 
org.springframework.jdbc.datasource.DataSourceUtils.doGetConnection(DataSourceUtils.java:115)
-       at 
org.springframework.jdbc.datasource.DataSourceUtils.getConnection(DataSourceUtils.java:78)
-       at 
org.mybatis.spring.transaction.SpringManagedTransaction.openConnection(SpringManagedTransaction.java:80)
-       at 
org.mybatis.spring.transaction.SpringManagedTransaction.getConnection(SpringManagedTransaction.java:67)
-       at 
org.apache.ibatis.executor.BaseExecutor.getConnection(BaseExecutor.java:337)
-       at 
org.apache.ibatis.executor.SimpleExecutor.prepareStatement(SimpleExecutor.java:86)
-       at 
org.apache.ibatis.executor.SimpleExecutor.doQuery(SimpleExecutor.java:62)
-       at 
org.apache.ibatis.executor.BaseExecutor.queryFromDatabase(BaseExecutor.java:325)
-       at org.apache.ibatis.executor.BaseExecutor.query(BaseExecutor.java:156)
-       at 
org.apache.ibatis.executor.CachingExecutor.query(CachingExecutor.java:109)
-       at 
org.apache.ibatis.executor.CachingExecutor.query(CachingExecutor.java:89)
-       at 
org.apache.ibatis.session.defaults.DefaultSqlSession.selectList(DefaultSqlSession.java:147)
-       at 
org.apache.ibatis.session.defaults.DefaultSqlSession.selectList(DefaultSqlSession.java:140)
-       at sun.reflect.GeneratedMethodAccessor51.invoke(Unknown Source)
-       at 
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
-       at java.lang.reflect.Method.invoke(Method.java:498)
-       at 
org.mybatis.spring.SqlSessionTemplate$SqlSessionInterceptor.invoke(SqlSessionTemplate.java:426)
-       at com.sun.proxy.$Proxy81.selectList(Unknown Source)
-       at 
org.mybatis.spring.SqlSessionTemplate.selectList(SqlSessionTemplate.java:223)
-       at 
org.apache.ibatis.binding.MapperMethod.executeForMany(MapperMethod.java:147)
-       at org.apache.ibatis.binding.MapperMethod.execute(MapperMethod.java:80)
-       at 
org.apache.ibatis.binding.MapperProxy$PlainMethodInvoker.invoke(MapperProxy.java:152)
-       at org.apache.ibatis.binding.MapperProxy.invoke(MapperProxy.java:85)
-       at com.sun.proxy.$Proxy92.listEventTargetRunners(Unknown Source)
-       at 
org.apache.rocketmq.eventbridge.adapter.persistence.target.mybatis.repository.MybatisEventTargetRunnerRepository.listEventTargetRunners(MybatisEventTargetRunnerRepository.java:77)
-       at 
org.apache.rocketmq.eventbridge.adapter.persistence.target.mybatis.repository.MybatisEventTargetRunnerRepository$$FastClassBySpringCGLIB$$d4e0cf7.invoke(<generated>)
-       at 
org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:204)
-       at 
org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:746)
-       at 
org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
-       at 
org.springframework.dao.support.PersistenceExceptionTranslationInterceptor.invoke(PersistenceExceptionTranslationInterceptor.java:139)
-       at 
org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:185)
-       at 
org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:688)
-       at 
org.apache.rocketmq.eventbridge.adapter.persistence.target.mybatis.repository.MybatisEventTargetRunnerRepository$$EnhancerBySpringCGLIB$$ae71ebee.listEventTargetRunners(<generated>)
-       at 
org.apache.rocketmq.eventbridge.adapter.runtime.service.TargetRunnerConfigOnDBObserver.getLatestTargetRunnerConfig(TargetRunnerConfigOnDBObserver.java:62)
-       at 
org.apache.rocketmq.eventbridge.adapter.runtime.service.AbstractTargetRunnerConfigObserver.diff(AbstractTargetRunnerConfigObserver.java:110)
-       at 
org.apache.rocketmq.eventbridge.adapter.runtime.service.TargetRunnerConfigOnDBObserver.lambda$addListen$0(TargetRunnerConfigOnDBObserver.java:82)
-       at 
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
-       at 
java.util.concurrent.FutureTask.runAndReset$$$capture(FutureTask.java:308)
-       at java.util.concurrent.FutureTask.runAndReset(FutureTask.java)
-       at 
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
-       at 
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
-       at 
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
-       at 
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
+       at org.h2.message.DbException.get(DbException.java:168)
+       at org.h2.value.ValueTimestamp.parse(ValueTimestamp.java:147)
+       at org.h2.value.Value.convertTo(Value.java:1071)
+       at org.h2.table.Column.convert(Column.java:177)
+       at org.h2.command.dml.Insert.insertRows(Insert.java:166)
+       at org.h2.command.dml.Insert.update(Insert.java:134)
+       at org.h2.command.CommandContainer.update(CommandContainer.java:102)
+       at org.h2.command.Command.executeUpdate(Command.java:261)
+       at 
org.h2.jdbc.JdbcPreparedStatement.executeUpdateInternal(JdbcPreparedStatement.java:199)
+       at 
org.h2.jdbc.JdbcPreparedStatement.executeUpdate(JdbcPreparedStatement.java:153)
+       at org.h2.result.UpdatableRow.insertRow(UpdatableRow.java:321)
+       at org.h2.jdbc.JdbcResultSet.insertRow(JdbcResultSet.java:2994)
+       at org.h2.server.web.WebApp.editResult(WebApp.java:1082)
+       at org.h2.server.web.WebApp.process(WebApp.java:227)
+       at org.h2.server.web.WebApp.processRequest(WebApp.java:170)
+       at org.h2.server.web.WebThread.process(WebThread.java:134)
+       at org.h2.server.web.WebThread.run(WebThread.java:90)
        at java.lang.Thread.run(Thread.java:748)
-2023-08-13 15:17:19 jdbc[19]: exception
-org.h2.jdbc.JdbcSQLException: Database is already closed (to disable automatic 
closing at VM shutdown, add ";DB_CLOSE_ON_EXIT=FALSE" to the db URL) [90121-197]
+Caused by: java.lang.StringIndexOutOfBoundsException: String index out of 
range: 0
+       at java.lang.String.charAt(String.java:658)
+       at org.h2.util.DateTimeUtils.parseDateValue(DateTimeUtils.java:337)
+       at org.h2.util.DateTimeUtils.parseTimestamp(DateTimeUtils.java:460)
+       at org.h2.value.ValueTimestamp.parse(ValueTimestamp.java:145)
+       ... 16 more
+2023-11-26 18:42:05 jdbc[8]: exception
+org.h2.jdbc.JdbcSQLException: Cannot parse "TIMESTAMP" constant ; SQL 
statement:
+INSERT INTO 
"PUBLIC"."EVENT_CLUSTER"("ID","NAME","RESOURCES","REPLICA","IMAGE","MD5","GMT_CREATE","GMT_MODIFY")VALUES(?,?,?,?,?,?,?,?)
 -- (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8) [22007-197]
        at org.h2.message.DbException.getJdbcSQLException(DbException.java:357)
-       at org.h2.message.DbException.get(DbException.java:179)
-       at org.h2.message.DbException.get(DbException.java:155)
-       at org.h2.message.DbException.get(DbException.java:144)
-       at org.h2.jdbc.JdbcConnection.checkClosed(JdbcConnection.java:1526)
-       at org.h2.jdbc.JdbcConnection.checkClosed(JdbcConnection.java:1502)
-       at org.h2.jdbc.JdbcConnection.createStatement(JdbcConnection.java:217)
-       at com.zaxxer.hikari.pool.PoolBase.isConnectionAlive(PoolBase.java:153)
-       at com.zaxxer.hikari.pool.HikariPool.getConnection(HikariPool.java:172)
-       at com.zaxxer.hikari.pool.HikariPool.getConnection(HikariPool.java:148)
-       at 
com.zaxxer.hikari.HikariDataSource.getConnection(HikariDataSource.java:100)
-       at 
org.springframework.jdbc.datasource.DataSourceUtils.fetchConnection(DataSourceUtils.java:151)
-       at 
org.springframework.jdbc.datasource.DataSourceUtils.doGetConnection(DataSourceUtils.java:115)
-       at 
org.springframework.jdbc.datasource.DataSourceUtils.getConnection(DataSourceUtils.java:78)
-       at 
org.mybatis.spring.transaction.SpringManagedTransaction.openConnection(SpringManagedTransaction.java:80)
-       at 
org.mybatis.spring.transaction.SpringManagedTransaction.getConnection(SpringManagedTransaction.java:67)
-       at 
org.apache.ibatis.executor.BaseExecutor.getConnection(BaseExecutor.java:337)
-       at 
org.apache.ibatis.executor.SimpleExecutor.prepareStatement(SimpleExecutor.java:86)
-       at 
org.apache.ibatis.executor.SimpleExecutor.doQuery(SimpleExecutor.java:62)
-       at 
org.apache.ibatis.executor.BaseExecutor.queryFromDatabase(BaseExecutor.java:325)
-       at org.apache.ibatis.executor.BaseExecutor.query(BaseExecutor.java:156)
-       at 
org.apache.ibatis.executor.CachingExecutor.query(CachingExecutor.java:109)
-       at 
org.apache.ibatis.executor.CachingExecutor.query(CachingExecutor.java:89)
-       at 
org.apache.ibatis.session.defaults.DefaultSqlSession.selectList(DefaultSqlSession.java:147)
-       at 
org.apache.ibatis.session.defaults.DefaultSqlSession.selectList(DefaultSqlSession.java:140)
-       at sun.reflect.GeneratedMethodAccessor51.invoke(Unknown Source)
-       at 
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
-       at java.lang.reflect.Method.invoke(Method.java:498)
-       at 
org.mybatis.spring.SqlSessionTemplate$SqlSessionInterceptor.invoke(SqlSessionTemplate.java:426)
-       at com.sun.proxy.$Proxy81.selectList(Unknown Source)
-       at 
org.mybatis.spring.SqlSessionTemplate.selectList(SqlSessionTemplate.java:223)
-       at 
org.apache.ibatis.binding.MapperMethod.executeForMany(MapperMethod.java:147)
-       at org.apache.ibatis.binding.MapperMethod.execute(MapperMethod.java:80)
-       at 
org.apache.ibatis.binding.MapperProxy$PlainMethodInvoker.invoke(MapperProxy.java:152)
-       at org.apache.ibatis.binding.MapperProxy.invoke(MapperProxy.java:85)
-       at com.sun.proxy.$Proxy92.listEventTargetRunners(Unknown Source)
-       at 
org.apache.rocketmq.eventbridge.adapter.persistence.target.mybatis.repository.MybatisEventTargetRunnerRepository.listEventTargetRunners(MybatisEventTargetRunnerRepository.java:77)
-       at 
org.apache.rocketmq.eventbridge.adapter.persistence.target.mybatis.repository.MybatisEventTargetRunnerRepository$$FastClassBySpringCGLIB$$d4e0cf7.invoke(<generated>)
-       at 
org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:204)
-       at 
org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:746)
-       at 
org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
-       at 
org.springframework.dao.support.PersistenceExceptionTranslationInterceptor.invoke(PersistenceExceptionTranslationInterceptor.java:139)
-       at 
org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:185)
-       at 
org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:688)
-       at 
org.apache.rocketmq.eventbridge.adapter.persistence.target.mybatis.repository.MybatisEventTargetRunnerRepository$$EnhancerBySpringCGLIB$$ae71ebee.listEventTargetRunners(<generated>)
-       at 
org.apache.rocketmq.eventbridge.adapter.runtime.service.TargetRunnerConfigOnDBObserver.getLatestTargetRunnerConfig(TargetRunnerConfigOnDBObserver.java:62)
-       at 
org.apache.rocketmq.eventbridge.adapter.runtime.service.AbstractTargetRunnerConfigObserver.diff(AbstractTargetRunnerConfigObserver.java:110)
-       at 
org.apache.rocketmq.eventbridge.adapter.runtime.service.TargetRunnerConfigOnDBObserver.lambda$addListen$0(TargetRunnerConfigOnDBObserver.java:82)
-       at 
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
-       at 
java.util.concurrent.FutureTask.runAndReset$$$capture(FutureTask.java:308)
-       at java.util.concurrent.FutureTask.runAndReset(FutureTask.java)
-       at 
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
-       at 
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
-       at 
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
-       at 
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
-       at java.lang.Thread.run(Thread.java:748)
-2023-08-13 15:17:19 jdbc[20]: exception
-org.h2.jdbc.JdbcSQLException: Database is already closed (to disable automatic 
closing at VM shutdown, add ";DB_CLOSE_ON_EXIT=FALSE" to the db URL) [90121-197]
-       at org.h2.message.DbException.getJdbcSQLException(DbException.java:357)
-       at org.h2.message.DbException.get(DbException.java:179)
-       at org.h2.message.DbException.get(DbException.java:155)
-       at org.h2.message.DbException.get(DbException.java:144)
-       at org.h2.jdbc.JdbcConnection.checkClosed(JdbcConnection.java:1526)
-       at org.h2.jdbc.JdbcConnection.checkClosed(JdbcConnection.java:1502)
-       at org.h2.jdbc.JdbcConnection.createStatement(JdbcConnection.java:217)
-       at com.zaxxer.hikari.pool.PoolBase.isConnectionAlive(PoolBase.java:153)
-       at com.zaxxer.hikari.pool.HikariPool.getConnection(HikariPool.java:172)
-       at com.zaxxer.hikari.pool.HikariPool.getConnection(HikariPool.java:148)
-       at 
com.zaxxer.hikari.HikariDataSource.getConnection(HikariDataSource.java:100)
-       at 
org.springframework.jdbc.datasource.DataSourceUtils.fetchConnection(DataSourceUtils.java:151)
-       at 
org.springframework.jdbc.datasource.DataSourceUtils.doGetConnection(DataSourceUtils.java:115)
-       at 
org.springframework.jdbc.datasource.DataSourceUtils.getConnection(DataSourceUtils.java:78)
-       at 
org.mybatis.spring.transaction.SpringManagedTransaction.openConnection(SpringManagedTransaction.java:80)
-       at 
org.mybatis.spring.transaction.SpringManagedTransaction.getConnection(SpringManagedTransaction.java:67)
-       at 
org.apache.ibatis.executor.BaseExecutor.getConnection(BaseExecutor.java:337)
-       at 
org.apache.ibatis.executor.SimpleExecutor.prepareStatement(SimpleExecutor.java:86)
-       at 
org.apache.ibatis.executor.SimpleExecutor.doQuery(SimpleExecutor.java:62)
-       at 
org.apache.ibatis.executor.BaseExecutor.queryFromDatabase(BaseExecutor.java:325)
-       at org.apache.ibatis.executor.BaseExecutor.query(BaseExecutor.java:156)
-       at 
org.apache.ibatis.executor.CachingExecutor.query(CachingExecutor.java:109)
-       at 
org.apache.ibatis.executor.CachingExecutor.query(CachingExecutor.java:89)
-       at 
org.apache.ibatis.session.defaults.DefaultSqlSession.selectList(DefaultSqlSession.java:147)
-       at 
org.apache.ibatis.session.defaults.DefaultSqlSession.selectList(DefaultSqlSession.java:140)
-       at sun.reflect.GeneratedMethodAccessor51.invoke(Unknown Source)
-       at 
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
-       at java.lang.reflect.Method.invoke(Method.java:498)
-       at 
org.mybatis.spring.SqlSessionTemplate$SqlSessionInterceptor.invoke(SqlSessionTemplate.java:426)
-       at com.sun.proxy.$Proxy81.selectList(Unknown Source)
-       at 
org.mybatis.spring.SqlSessionTemplate.selectList(SqlSessionTemplate.java:223)
-       at 
org.apache.ibatis.binding.MapperMethod.executeForMany(MapperMethod.java:147)
-       at org.apache.ibatis.binding.MapperMethod.execute(MapperMethod.java:80)
-       at 
org.apache.ibatis.binding.MapperProxy$PlainMethodInvoker.invoke(MapperProxy.java:152)
-       at org.apache.ibatis.binding.MapperProxy.invoke(MapperProxy.java:85)
-       at com.sun.proxy.$Proxy92.listEventTargetRunners(Unknown Source)
-       at 
org.apache.rocketmq.eventbridge.adapter.persistence.target.mybatis.repository.MybatisEventTargetRunnerRepository.listEventTargetRunners(MybatisEventTargetRunnerRepository.java:77)
-       at 
org.apache.rocketmq.eventbridge.adapter.persistence.target.mybatis.repository.MybatisEventTargetRunnerRepository$$FastClassBySpringCGLIB$$d4e0cf7.invoke(<generated>)
-       at 
org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:204)
-       at 
org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:746)
-       at 
org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
-       at 
org.springframework.dao.support.PersistenceExceptionTranslationInterceptor.invoke(PersistenceExceptionTranslationInterceptor.java:139)
-       at 
org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:185)
-       at 
org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:688)
-       at 
org.apache.rocketmq.eventbridge.adapter.persistence.target.mybatis.repository.MybatisEventTargetRunnerRepository$$EnhancerBySpringCGLIB$$ae71ebee.listEventTargetRunners(<generated>)
-       at 
org.apache.rocketmq.eventbridge.adapter.runtime.service.TargetRunnerConfigOnDBObserver.getLatestTargetRunnerConfig(TargetRunnerConfigOnDBObserver.java:62)
-       at 
org.apache.rocketmq.eventbridge.adapter.runtime.service.AbstractTargetRunnerConfigObserver.diff(AbstractTargetRunnerConfigObserver.java:110)
-       at 
org.apache.rocketmq.eventbridge.adapter.runtime.service.TargetRunnerConfigOnDBObserver.lambda$addListen$0(TargetRunnerConfigOnDBObserver.java:82)
-       at 
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
-       at 
java.util.concurrent.FutureTask.runAndReset$$$capture(FutureTask.java:308)
-       at java.util.concurrent.FutureTask.runAndReset(FutureTask.java)
-       at 
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
-       at 
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
-       at 
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
-       at 
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
-       at java.lang.Thread.run(Thread.java:748)
-2023-08-13 15:17:19 jdbc[21]: exception
-org.h2.jdbc.JdbcSQLException: Database is already closed (to disable automatic 
closing at VM shutdown, add ";DB_CLOSE_ON_EXIT=FALSE" to the db URL) [90121-197]
-       at org.h2.message.DbException.getJdbcSQLException(DbException.java:357)
-       at org.h2.message.DbException.get(DbException.java:179)
-       at org.h2.message.DbException.get(DbException.java:155)
-       at org.h2.message.DbException.get(DbException.java:144)
-       at org.h2.jdbc.JdbcConnection.checkClosed(JdbcConnection.java:1526)
-       at org.h2.jdbc.JdbcConnection.checkClosed(JdbcConnection.java:1502)
-       at org.h2.jdbc.JdbcConnection.createStatement(JdbcConnection.java:217)
-       at com.zaxxer.hikari.pool.PoolBase.isConnectionAlive(PoolBase.java:153)
-       at com.zaxxer.hikari.pool.HikariPool.getConnection(HikariPool.java:172)
-       at com.zaxxer.hikari.pool.HikariPool.getConnection(HikariPool.java:148)
-       at 
com.zaxxer.hikari.HikariDataSource.getConnection(HikariDataSource.java:100)
-       at 
org.springframework.jdbc.datasource.DataSourceUtils.fetchConnection(DataSourceUtils.java:151)
-       at 
org.springframework.jdbc.datasource.DataSourceUtils.doGetConnection(DataSourceUtils.java:115)
-       at 
org.springframework.jdbc.datasource.DataSourceUtils.getConnection(DataSourceUtils.java:78)
-       at 
org.mybatis.spring.transaction.SpringManagedTransaction.openConnection(SpringManagedTransaction.java:80)
-       at 
org.mybatis.spring.transaction.SpringManagedTransaction.getConnection(SpringManagedTransaction.java:67)
-       at 
org.apache.ibatis.executor.BaseExecutor.getConnection(BaseExecutor.java:337)
-       at 
org.apache.ibatis.executor.SimpleExecutor.prepareStatement(SimpleExecutor.java:86)
-       at 
org.apache.ibatis.executor.SimpleExecutor.doQuery(SimpleExecutor.java:62)
-       at 
org.apache.ibatis.executor.BaseExecutor.queryFromDatabase(BaseExecutor.java:325)
-       at org.apache.ibatis.executor.BaseExecutor.query(BaseExecutor.java:156)
-       at 
org.apache.ibatis.executor.CachingExecutor.query(CachingExecutor.java:109)
-       at 
org.apache.ibatis.executor.CachingExecutor.query(CachingExecutor.java:89)
-       at 
org.apache.ibatis.session.defaults.DefaultSqlSession.selectList(DefaultSqlSession.java:147)
-       at 
org.apache.ibatis.session.defaults.DefaultSqlSession.selectList(DefaultSqlSession.java:140)
-       at sun.reflect.GeneratedMethodAccessor51.invoke(Unknown Source)
-       at 
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
-       at java.lang.reflect.Method.invoke(Method.java:498)
-       at 
org.mybatis.spring.SqlSessionTemplate$SqlSessionInterceptor.invoke(SqlSessionTemplate.java:426)
-       at com.sun.proxy.$Proxy81.selectList(Unknown Source)
-       at 
org.mybatis.spring.SqlSessionTemplate.selectList(SqlSessionTemplate.java:223)
-       at 
org.apache.ibatis.binding.MapperMethod.executeForMany(MapperMethod.java:147)
-       at org.apache.ibatis.binding.MapperMethod.execute(MapperMethod.java:80)
-       at 
org.apache.ibatis.binding.MapperProxy$PlainMethodInvoker.invoke(MapperProxy.java:152)
-       at org.apache.ibatis.binding.MapperProxy.invoke(MapperProxy.java:85)
-       at com.sun.proxy.$Proxy92.listEventTargetRunners(Unknown Source)
-       at 
org.apache.rocketmq.eventbridge.adapter.persistence.target.mybatis.repository.MybatisEventTargetRunnerRepository.listEventTargetRunners(MybatisEventTargetRunnerRepository.java:77)
-       at 
org.apache.rocketmq.eventbridge.adapter.persistence.target.mybatis.repository.MybatisEventTargetRunnerRepository$$FastClassBySpringCGLIB$$d4e0cf7.invoke(<generated>)
-       at 
org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:204)
-       at 
org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:746)
-       at 
org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
-       at 
org.springframework.dao.support.PersistenceExceptionTranslationInterceptor.invoke(PersistenceExceptionTranslationInterceptor.java:139)
-       at 
org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:185)
-       at 
org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:688)
-       at 
org.apache.rocketmq.eventbridge.adapter.persistence.target.mybatis.repository.MybatisEventTargetRunnerRepository$$EnhancerBySpringCGLIB$$ae71ebee.listEventTargetRunners(<generated>)
-       at 
org.apache.rocketmq.eventbridge.adapter.runtime.service.TargetRunnerConfigOnDBObserver.getLatestTargetRunnerConfig(TargetRunnerConfigOnDBObserver.java:62)
-       at 
org.apache.rocketmq.eventbridge.adapter.runtime.service.AbstractTargetRunnerConfigObserver.diff(AbstractTargetRunnerConfigObserver.java:110)
-       at 
org.apache.rocketmq.eventbridge.adapter.runtime.service.TargetRunnerConfigOnDBObserver.lambda$addListen$0(TargetRunnerConfigOnDBObserver.java:82)
-       at 
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
-       at 
java.util.concurrent.FutureTask.runAndReset$$$capture(FutureTask.java:308)
-       at java.util.concurrent.FutureTask.runAndReset(FutureTask.java)
-       at 
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
-       at 
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
-       at 
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
-       at 
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
-       at java.lang.Thread.run(Thread.java:748)
-2023-08-13 15:17:19 jdbc[22]: exception
-org.h2.jdbc.JdbcSQLException: Database is already closed (to disable automatic 
closing at VM shutdown, add ";DB_CLOSE_ON_EXIT=FALSE" to the db URL) [90121-197]
-       at org.h2.message.DbException.getJdbcSQLException(DbException.java:357)
-       at org.h2.message.DbException.get(DbException.java:179)
-       at org.h2.message.DbException.get(DbException.java:155)
-       at org.h2.message.DbException.get(DbException.java:144)
-       at org.h2.jdbc.JdbcConnection.checkClosed(JdbcConnection.java:1526)
-       at org.h2.jdbc.JdbcConnection.checkClosed(JdbcConnection.java:1502)
-       at org.h2.jdbc.JdbcConnection.createStatement(JdbcConnection.java:217)
-       at com.zaxxer.hikari.pool.PoolBase.isConnectionAlive(PoolBase.java:153)
-       at com.zaxxer.hikari.pool.HikariPool.getConnection(HikariPool.java:172)
-       at com.zaxxer.hikari.pool.HikariPool.getConnection(HikariPool.java:148)
-       at 
com.zaxxer.hikari.HikariDataSource.getConnection(HikariDataSource.java:100)
-       at 
org.springframework.jdbc.datasource.DataSourceUtils.fetchConnection(DataSourceUtils.java:151)
-       at 
org.springframework.jdbc.datasource.DataSourceUtils.doGetConnection(DataSourceUtils.java:115)
-       at 
org.springframework.jdbc.datasource.DataSourceUtils.getConnection(DataSourceUtils.java:78)
-       at 
org.mybatis.spring.transaction.SpringManagedTransaction.openConnection(SpringManagedTransaction.java:80)
-       at 
org.mybatis.spring.transaction.SpringManagedTransaction.getConnection(SpringManagedTransaction.java:67)
-       at 
org.apache.ibatis.executor.BaseExecutor.getConnection(BaseExecutor.java:337)
-       at 
org.apache.ibatis.executor.SimpleExecutor.prepareStatement(SimpleExecutor.java:86)
-       at 
org.apache.ibatis.executor.SimpleExecutor.doQuery(SimpleExecutor.java:62)
-       at 
org.apache.ibatis.executor.BaseExecutor.queryFromDatabase(BaseExecutor.java:325)
-       at org.apache.ibatis.executor.BaseExecutor.query(BaseExecutor.java:156)
-       at 
org.apache.ibatis.executor.CachingExecutor.query(CachingExecutor.java:109)
-       at 
org.apache.ibatis.executor.CachingExecutor.query(CachingExecutor.java:89)
-       at 
org.apache.ibatis.session.defaults.DefaultSqlSession.selectList(DefaultSqlSession.java:147)
-       at 
org.apache.ibatis.session.defaults.DefaultSqlSession.selectList(DefaultSqlSession.java:140)
-       at sun.reflect.GeneratedMethodAccessor51.invoke(Unknown Source)
-       at 
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
-       at java.lang.reflect.Method.invoke(Method.java:498)
-       at 
org.mybatis.spring.SqlSessionTemplate$SqlSessionInterceptor.invoke(SqlSessionTemplate.java:426)
-       at com.sun.proxy.$Proxy81.selectList(Unknown Source)
-       at 
org.mybatis.spring.SqlSessionTemplate.selectList(SqlSessionTemplate.java:223)
-       at 
org.apache.ibatis.binding.MapperMethod.executeForMany(MapperMethod.java:147)
-       at org.apache.ibatis.binding.MapperMethod.execute(MapperMethod.java:80)
-       at 
org.apache.ibatis.binding.MapperProxy$PlainMethodInvoker.invoke(MapperProxy.java:152)
-       at org.apache.ibatis.binding.MapperProxy.invoke(MapperProxy.java:85)
-       at com.sun.proxy.$Proxy92.listEventTargetRunners(Unknown Source)
-       at 
org.apache.rocketmq.eventbridge.adapter.persistence.target.mybatis.repository.MybatisEventTargetRunnerRepository.listEventTargetRunners(MybatisEventTargetRunnerRepository.java:77)
-       at 
org.apache.rocketmq.eventbridge.adapter.persistence.target.mybatis.repository.MybatisEventTargetRunnerRepository$$FastClassBySpringCGLIB$$d4e0cf7.invoke(<generated>)
-       at 
org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:204)
-       at 
org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:746)
-       at 
org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
-       at 
org.springframework.dao.support.PersistenceExceptionTranslationInterceptor.invoke(PersistenceExceptionTranslationInterceptor.java:139)
-       at 
org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:185)
-       at 
org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:688)
-       at 
org.apache.rocketmq.eventbridge.adapter.persistence.target.mybatis.repository.MybatisEventTargetRunnerRepository$$EnhancerBySpringCGLIB$$ae71ebee.listEventTargetRunners(<generated>)
-       at 
org.apache.rocketmq.eventbridge.adapter.runtime.service.TargetRunnerConfigOnDBObserver.getLatestTargetRunnerConfig(TargetRunnerConfigOnDBObserver.java:62)
-       at 
org.apache.rocketmq.eventbridge.adapter.runtime.service.AbstractTargetRunnerConfigObserver.diff(AbstractTargetRunnerConfigObserver.java:110)
-       at 
org.apache.rocketmq.eventbridge.adapter.runtime.service.TargetRunnerConfigOnDBObserver.lambda$addListen$0(TargetRunnerConfigOnDBObserver.java:82)
-       at 
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
-       at 
java.util.concurrent.FutureTask.runAndReset$$$capture(FutureTask.java:308)
-       at java.util.concurrent.FutureTask.runAndReset(FutureTask.java)
-       at 
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
-       at 
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
-       at 
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
-       at 
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
-       at java.lang.Thread.run(Thread.java:748)
-2023-08-15 20:23:33 jdbc[3413]: exception
-org.h2.jdbc.JdbcSQLException: Database is already closed (to disable automatic 
closing at VM shutdown, add ";DB_CLOSE_ON_EXIT=FALSE" to the db URL) [90121-197]
-       at org.h2.message.DbException.getJdbcSQLException(DbException.java:357)
-       at org.h2.message.DbException.get(DbException.java:179)
-       at org.h2.message.DbException.get(DbException.java:155)
-       at org.h2.message.DbException.get(DbException.java:144)
-       at org.h2.jdbc.JdbcConnection.checkClosed(JdbcConnection.java:1526)
-       at org.h2.jdbc.JdbcConnection.checkClosed(JdbcConnection.java:1502)
-       at org.h2.jdbc.JdbcConnection.createStatement(JdbcConnection.java:217)
-       at com.zaxxer.hikari.pool.PoolBase.isConnectionAlive(PoolBase.java:153)
-       at com.zaxxer.hikari.pool.HikariPool.getConnection(HikariPool.java:172)
-       at com.zaxxer.hikari.pool.HikariPool.getConnection(HikariPool.java:148)
-       at 
com.zaxxer.hikari.HikariDataSource.getConnection(HikariDataSource.java:100)
-       at 
org.springframework.jdbc.datasource.DataSourceUtils.fetchConnection(DataSourceUtils.java:151)
-       at 
org.springframework.jdbc.datasource.DataSourceUtils.doGetConnection(DataSourceUtils.java:115)
-       at 
org.springframework.jdbc.datasource.DataSourceUtils.getConnection(DataSourceUtils.java:78)
-       at 
org.mybatis.spring.transaction.SpringManagedTransaction.openConnection(SpringManagedTransaction.java:80)
-       at 
org.mybatis.spring.transaction.SpringManagedTransaction.getConnection(SpringManagedTransaction.java:67)
-       at 
org.apache.ibatis.executor.BaseExecutor.getConnection(BaseExecutor.java:337)
-       at 
org.apache.ibatis.executor.SimpleExecutor.prepareStatement(SimpleExecutor.java:86)
-       at 
org.apache.ibatis.executor.SimpleExecutor.doQuery(SimpleExecutor.java:62)
-       at 
org.apache.ibatis.executor.BaseExecutor.queryFromDatabase(BaseExecutor.java:325)
-       at org.apache.ibatis.executor.BaseExecutor.query(BaseExecutor.java:156)
-       at 
org.apache.ibatis.executor.CachingExecutor.query(CachingExecutor.java:109)
-       at 
org.apache.ibatis.executor.CachingExecutor.query(CachingExecutor.java:89)
-       at 
org.apache.ibatis.session.defaults.DefaultSqlSession.selectList(DefaultSqlSession.java:147)
-       at 
org.apache.ibatis.session.defaults.DefaultSqlSession.selectList(DefaultSqlSession.java:140)
-       at sun.reflect.GeneratedMethodAccessor51.invoke(Unknown Source)
-       at 
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
-       at java.lang.reflect.Method.invoke(Method.java:498)
-       at 
org.mybatis.spring.SqlSessionTemplate$SqlSessionInterceptor.invoke(SqlSessionTemplate.java:426)
-       at com.sun.proxy.$Proxy81.selectList(Unknown Source)
-       at 
org.mybatis.spring.SqlSessionTemplate.selectList(SqlSessionTemplate.java:223)
-       at 
org.apache.ibatis.binding.MapperMethod.executeForMany(MapperMethod.java:147)
-       at org.apache.ibatis.binding.MapperMethod.execute(MapperMethod.java:80)
-       at 
org.apache.ibatis.binding.MapperProxy$PlainMethodInvoker.invoke(MapperProxy.java:152)
-       at org.apache.ibatis.binding.MapperProxy.invoke(MapperProxy.java:85)
-       at com.sun.proxy.$Proxy92.listEventTargetRunners(Unknown Source)
-       at 
org.apache.rocketmq.eventbridge.adapter.persistence.target.mybatis.repository.MybatisEventTargetRunnerRepository.listEventTargetRunners(MybatisEventTargetRunnerRepository.java:77)
-       at 
org.apache.rocketmq.eventbridge.adapter.persistence.target.mybatis.repository.MybatisEventTargetRunnerRepository$$FastClassBySpringCGLIB$$d4e0cf7.invoke(<generated>)
-       at 
org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:204)
-       at 
org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:746)
-       at 
org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
-       at 
org.springframework.dao.support.PersistenceExceptionTranslationInterceptor.invoke(PersistenceExceptionTranslationInterceptor.java:139)
-       at 
org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:185)
-       at 
org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:688)
-       at 
org.apache.rocketmq.eventbridge.adapter.persistence.target.mybatis.repository.MybatisEventTargetRunnerRepository$$EnhancerBySpringCGLIB$$ed257c41.listEventTargetRunners(<generated>)
-       at 
org.apache.rocketmq.eventbridge.adapter.runtime.service.TargetRunnerConfigOnDBObserver.getLatestTargetRunnerConfig(TargetRunnerConfigOnDBObserver.java:62)
-       at 
org.apache.rocketmq.eventbridge.adapter.runtime.service.AbstractTargetRunnerConfigObserver.diff(AbstractTargetRunnerConfigObserver.java:110)
-       at 
org.apache.rocketmq.eventbridge.adapter.runtime.service.TargetRunnerConfigOnDBObserver.lambda$addListen$0(TargetRunnerConfigOnDBObserver.java:82)
-       at 
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
-       at 
java.util.concurrent.FutureTask.runAndReset$$$capture(FutureTask.java:308)
-       at java.util.concurrent.FutureTask.runAndReset(FutureTask.java)
-       at 
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
-       at 
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
-       at 
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
-       at 
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
-       at java.lang.Thread.run(Thread.java:748)
-2023-08-15 20:23:33 jdbc[3414]: exception
-org.h2.jdbc.JdbcSQLException: Database is already closed (to disable automatic 
closing at VM shutdown, add ";DB_CLOSE_ON_EXIT=FALSE" to the db URL) [90121-197]
-       at org.h2.message.DbException.getJdbcSQLException(DbException.java:357)
-       at org.h2.message.DbException.get(DbException.java:179)
-       at org.h2.message.DbException.get(DbException.java:155)
-       at org.h2.message.DbException.get(DbException.java:144)
-       at org.h2.jdbc.JdbcConnection.checkClosed(JdbcConnection.java:1526)
-       at org.h2.jdbc.JdbcConnection.checkClosed(JdbcConnection.java:1502)
-       at org.h2.jdbc.JdbcConnection.createStatement(JdbcConnection.java:217)
-       at com.zaxxer.hikari.pool.PoolBase.isConnectionAlive(PoolBase.java:153)
-       at com.zaxxer.hikari.pool.HikariPool.getConnection(HikariPool.java:172)
-       at com.zaxxer.hikari.pool.HikariPool.getConnection(HikariPool.java:148)
-       at 
com.zaxxer.hikari.HikariDataSource.getConnection(HikariDataSource.java:100)
-       at 
org.springframework.jdbc.datasource.DataSourceUtils.fetchConnection(DataSourceUtils.java:151)
-       at 
org.springframework.jdbc.datasource.DataSourceUtils.doGetConnection(DataSourceUtils.java:115)
-       at 
org.springframework.jdbc.datasource.DataSourceUtils.getConnection(DataSourceUtils.java:78)
-       at 
org.mybatis.spring.transaction.SpringManagedTransaction.openConnection(SpringManagedTransaction.java:80)
-       at 
org.mybatis.spring.transaction.SpringManagedTransaction.getConnection(SpringManagedTransaction.java:67)
-       at 
org.apache.ibatis.executor.BaseExecutor.getConnection(BaseExecutor.java:337)
-       at 
org.apache.ibatis.executor.SimpleExecutor.prepareStatement(SimpleExecutor.java:86)
-       at 
org.apache.ibatis.executor.SimpleExecutor.doQuery(SimpleExecutor.java:62)
-       at 
org.apache.ibatis.executor.BaseExecutor.queryFromDatabase(BaseExecutor.java:325)
-       at org.apache.ibatis.executor.BaseExecutor.query(BaseExecutor.java:156)
-       at 
org.apache.ibatis.executor.CachingExecutor.query(CachingExecutor.java:109)
-       at 
org.apache.ibatis.executor.CachingExecutor.query(CachingExecutor.java:89)
-       at 
org.apache.ibatis.session.defaults.DefaultSqlSession.selectList(DefaultSqlSession.java:147)
-       at 
org.apache.ibatis.session.defaults.DefaultSqlSession.selectList(DefaultSqlSession.java:140)
-       at sun.reflect.GeneratedMethodAccessor51.invoke(Unknown Source)
-       at 
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
-       at java.lang.reflect.Method.invoke(Method.java:498)
-       at 
org.mybatis.spring.SqlSessionTemplate$SqlSessionInterceptor.invoke(SqlSessionTemplate.java:426)
-       at com.sun.proxy.$Proxy81.selectList(Unknown Source)
-       at 
org.mybatis.spring.SqlSessionTemplate.selectList(SqlSessionTemplate.java:223)
-       at 
org.apache.ibatis.binding.MapperMethod.executeForMany(MapperMethod.java:147)
-       at org.apache.ibatis.binding.MapperMethod.execute(MapperMethod.java:80)
-       at 
org.apache.ibatis.binding.MapperProxy$PlainMethodInvoker.invoke(MapperProxy.java:152)
-       at org.apache.ibatis.binding.MapperProxy.invoke(MapperProxy.java:85)
-       at com.sun.proxy.$Proxy92.listEventTargetRunners(Unknown Source)
-       at 
org.apache.rocketmq.eventbridge.adapter.persistence.target.mybatis.repository.MybatisEventTargetRunnerRepository.listEventTargetRunners(MybatisEventTargetRunnerRepository.java:77)
-       at 
org.apache.rocketmq.eventbridge.adapter.persistence.target.mybatis.repository.MybatisEventTargetRunnerRepository$$FastClassBySpringCGLIB$$d4e0cf7.invoke(<generated>)
-       at 
org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:204)
-       at 
org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:746)
-       at 
org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
-       at 
org.springframework.dao.support.PersistenceExceptionTranslationInterceptor.invoke(PersistenceExceptionTranslationInterceptor.java:139)
-       at 
org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:185)
-       at 
org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:688)
-       at 
org.apache.rocketmq.eventbridge.adapter.persistence.target.mybatis.repository.MybatisEventTargetRunnerRepository$$EnhancerBySpringCGLIB$$ed257c41.listEventTargetRunners(<generated>)
-       at 
org.apache.rocketmq.eventbridge.adapter.runtime.service.TargetRunnerConfigOnDBObserver.getLatestTargetRunnerConfig(TargetRunnerConfigOnDBObserver.java:62)
-       at 
org.apache.rocketmq.eventbridge.adapter.runtime.service.AbstractTargetRunnerConfigObserver.diff(AbstractTargetRunnerConfigObserver.java:110)
-       at 
org.apache.rocketmq.eventbridge.adapter.runtime.service.TargetRunnerConfigOnDBObserver.lambda$addListen$0(TargetRunnerConfigOnDBObserver.java:82)
-       at 
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
-       at 
java.util.concurrent.FutureTask.runAndReset$$$capture(FutureTask.java:308)
-       at java.util.concurrent.FutureTask.runAndReset(FutureTask.java)
-       at 
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
-       at 
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
-       at 
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
-       at 
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
-       at java.lang.Thread.run(Thread.java:748)
-2023-08-15 20:23:33 jdbc[3415]: exception
-org.h2.jdbc.JdbcSQLException: Database is already closed (to disable automatic 
closing at VM shutdown, add ";DB_CLOSE_ON_EXIT=FALSE" to the db URL) [90121-197]
-       at org.h2.message.DbException.getJdbcSQLException(DbException.java:357)
-       at org.h2.message.DbException.get(DbException.java:179)
-       at org.h2.message.DbException.get(DbException.java:155)
-       at org.h2.message.DbException.get(DbException.java:144)
-       at org.h2.jdbc.JdbcConnection.checkClosed(JdbcConnection.java:1526)
-       at org.h2.jdbc.JdbcConnection.checkClosed(JdbcConnection.java:1502)
-       at org.h2.jdbc.JdbcConnection.createStatement(JdbcConnection.java:217)
-       at com.zaxxer.hikari.pool.PoolBase.isConnectionAlive(PoolBase.java:153)
-       at com.zaxxer.hikari.pool.HikariPool.getConnection(HikariPool.java:172)
-       at com.zaxxer.hikari.pool.HikariPool.getConnection(HikariPool.java:148)
-       at 
com.zaxxer.hikari.HikariDataSource.getConnection(HikariDataSource.java:100)
-       at 
org.springframework.jdbc.datasource.DataSourceUtils.fetchConnection(DataSourceUtils.java:151)
-       at 
org.springframework.jdbc.datasource.DataSourceUtils.doGetConnection(DataSourceUtils.java:115)
-       at 
org.springframework.jdbc.datasource.DataSourceUtils.getConnection(DataSourceUtils.java:78)
-       at 
org.mybatis.spring.transaction.SpringManagedTransaction.openConnection(SpringManagedTransaction.java:80)
-       at 
org.mybatis.spring.transaction.SpringManagedTransaction.getConnection(SpringManagedTransaction.java:67)
-       at 
org.apache.ibatis.executor.BaseExecutor.getConnection(BaseExecutor.java:337)
-       at 
org.apache.ibatis.executor.SimpleExecutor.prepareStatement(SimpleExecutor.java:86)
-       at 
org.apache.ibatis.executor.SimpleExecutor.doQuery(SimpleExecutor.java:62)
-       at 
org.apache.ibatis.executor.BaseExecutor.queryFromDatabase(BaseExecutor.java:325)
-       at org.apache.ibatis.executor.BaseExecutor.query(BaseExecutor.java:156)
-       at 
org.apache.ibatis.executor.CachingExecutor.query(CachingExecutor.java:109)
-       at 
org.apache.ibatis.executor.CachingExecutor.query(CachingExecutor.java:89)
-       at 
org.apache.ibatis.session.defaults.DefaultSqlSession.selectList(DefaultSqlSession.java:147)
-       at 
org.apache.ibatis.session.defaults.DefaultSqlSession.selectList(DefaultSqlSession.java:140)
-       at sun.reflect.GeneratedMethodAccessor51.invoke(Unknown Source)
-       at 
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
-       at java.lang.reflect.Method.invoke(Method.java:498)
-       at 
org.mybatis.spring.SqlSessionTemplate$SqlSessionInterceptor.invoke(SqlSessionTemplate.java:426)
-       at com.sun.proxy.$Proxy81.selectList(Unknown Source)
-       at 
org.mybatis.spring.SqlSessionTemplate.selectList(SqlSessionTemplate.java:223)
-       at 
org.apache.ibatis.binding.MapperMethod.executeForMany(MapperMethod.java:147)
-       at org.apache.ibatis.binding.MapperMethod.execute(MapperMethod.java:80)
-       at 
org.apache.ibatis.binding.MapperProxy$PlainMethodInvoker.invoke(MapperProxy.java:152)
-       at org.apache.ibatis.binding.MapperProxy.invoke(MapperProxy.java:85)
-       at com.sun.proxy.$Proxy92.listEventTargetRunners(Unknown Source)
-       at 
org.apache.rocketmq.eventbridge.adapter.persistence.target.mybatis.repository.MybatisEventTargetRunnerRepository.listEventTargetRunners(MybatisEventTargetRunnerRepository.java:77)
-       at 
org.apache.rocketmq.eventbridge.adapter.persistence.target.mybatis.repository.MybatisEventTargetRunnerRepository$$FastClassBySpringCGLIB$$d4e0cf7.invoke(<generated>)
-       at 
org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:204)
-       at 
org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:746)
-       at 
org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
-       at 
org.springframework.dao.support.PersistenceExceptionTranslationInterceptor.invoke(PersistenceExceptionTranslationInterceptor.java:139)
-       at 
org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:185)
-       at 
org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:688)
-       at 
org.apache.rocketmq.eventbridge.adapter.persistence.target.mybatis.repository.MybatisEventTargetRunnerRepository$$EnhancerBySpringCGLIB$$ed257c41.listEventTargetRunners(<generated>)
-       at 
org.apache.rocketmq.eventbridge.adapter.runtime.service.TargetRunnerConfigOnDBObserver.getLatestTargetRunnerConfig(TargetRunnerConfigOnDBObserver.java:62)
-       at 
org.apache.rocketmq.eventbridge.adapter.runtime.service.AbstractTargetRunnerConfigObserver.diff(AbstractTargetRunnerConfigObserver.java:110)
-       at 
org.apache.rocketmq.eventbridge.adapter.runtime.service.TargetRunnerConfigOnDBObserver.lambda$addListen$0(TargetRunnerConfigOnDBObserver.java:82)
-       at 
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
-       at 
java.util.concurrent.FutureTask.runAndReset$$$capture(FutureTask.java:308)
-       at java.util.concurrent.FutureTask.runAndReset(FutureTask.java)
-       at 
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
-       at 
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
-       at 
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
-       at 
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
-       at java.lang.Thread.run(Thread.java:748)
-2023-08-15 20:23:33 jdbc[3416]: exception
-org.h2.jdbc.JdbcSQLException: Database is already closed (to disable automatic 
closing at VM shutdown, add ";DB_CLOSE_ON_EXIT=FALSE" to the db URL) [90121-197]
-       at org.h2.message.DbException.getJdbcSQLException(DbException.java:357)
-       at org.h2.message.DbException.get(DbException.java:179)
-       at org.h2.message.DbException.get(DbException.java:155)
-       at org.h2.message.DbException.get(DbException.java:144)
-       at org.h2.jdbc.JdbcConnection.checkClosed(JdbcConnection.java:1526)
-       at org.h2.jdbc.JdbcConnection.checkClosed(JdbcConnection.java:1502)
-       at org.h2.jdbc.JdbcConnection.createStatement(JdbcConnection.java:217)
-       at com.zaxxer.hikari.pool.PoolBase.isConnectionAlive(PoolBase.java:153)
-       at com.zaxxer.hikari.pool.HikariPool.getConnection(HikariPool.java:172)
-       at com.zaxxer.hikari.pool.HikariPool.getConnection(HikariPool.java:148)
-       at 
com.zaxxer.hikari.HikariDataSource.getConnection(HikariDataSource.java:100)
-       at 
org.springframework.jdbc.datasource.DataSourceUtils.fetchConnection(DataSourceUtils.java:151)
-       at 
org.springframework.jdbc.datasource.DataSourceUtils.doGetConnection(DataSourceUtils.java:115)
-       at 
org.springframework.jdbc.datasource.DataSourceUtils.getConnection(DataSourceUtils.java:78)
-       at 
org.mybatis.spring.transaction.SpringManagedTransaction.openConnection(SpringManagedTransaction.java:80)
-       at 
org.mybatis.spring.transaction.SpringManagedTransaction.getConnection(SpringManagedTransaction.java:67)
-       at 
org.apache.ibatis.executor.BaseExecutor.getConnection(BaseExecutor.java:337)
-       at 
org.apache.ibatis.executor.SimpleExecutor.prepareStatement(SimpleExecutor.java:86)
-       at 
org.apache.ibatis.executor.SimpleExecutor.doQuery(SimpleExecutor.java:62)
-       at 
org.apache.ibatis.executor.BaseExecutor.queryFromDatabase(BaseExecutor.java:325)
-       at org.apache.ibatis.executor.BaseExecutor.query(BaseExecutor.java:156)
-       at 
org.apache.ibatis.executor.CachingExecutor.query(CachingExecutor.java:109)
-       at 
org.apache.ibatis.executor.CachingExecutor.query(CachingExecutor.java:89)
-       at 
org.apache.ibatis.session.defaults.DefaultSqlSession.selectList(DefaultSqlSession.java:147)
-       at 
org.apache.ibatis.session.defaults.DefaultSqlSession.selectList(DefaultSqlSession.java:140)
-       at sun.reflect.GeneratedMethodAccessor51.invoke(Unknown Source)
-       at 
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
-       at java.lang.reflect.Method.invoke(Method.java:498)
-       at 
org.mybatis.spring.SqlSessionTemplate$SqlSessionInterceptor.invoke(SqlSessionTemplate.java:426)
-       at com.sun.proxy.$Proxy81.selectList(Unknown Source)
-       at 
org.mybatis.spring.SqlSessionTemplate.selectList(SqlSessionTemplate.java:223)
-       at 
org.apache.ibatis.binding.MapperMethod.executeForMany(MapperMethod.java:147)
-       at org.apache.ibatis.binding.MapperMethod.execute(MapperMethod.java:80)
-       at 
org.apache.ibatis.binding.MapperProxy$PlainMethodInvoker.invoke(MapperProxy.java:152)
-       at org.apache.ibatis.binding.MapperProxy.invoke(MapperProxy.java:85)
-       at com.sun.proxy.$Proxy92.listEventTargetRunners(Unknown Source)
-       at 
org.apache.rocketmq.eventbridge.adapter.persistence.target.mybatis.repository.MybatisEventTargetRunnerRepository.listEventTargetRunners(MybatisEventTargetRunnerRepository.java:77)
-       at 
org.apache.rocketmq.eventbridge.adapter.persistence.target.mybatis.repository.MybatisEventTargetRunnerRepository$$FastClassBySpringCGLIB$$d4e0cf7.invoke(<generated>)
-       at 
org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:204)
-       at 
org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:746)
-       at 
org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
-       at 
org.springframework.dao.support.PersistenceExceptionTranslationInterceptor.invoke(PersistenceExceptionTranslationInterceptor.java:139)
-       at 
org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:185)
-       at 
org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:688)
-       at 
org.apache.rocketmq.eventbridge.adapter.persistence.target.mybatis.repository.MybatisEventTargetRunnerRepository$$EnhancerBySpringCGLIB$$ed257c41.listEventTargetRunners(<generated>)
-       at 
org.apache.rocketmq.eventbridge.adapter.runtime.service.TargetRunnerConfigOnDBObserver.getLatestTargetRunnerConfig(TargetRunnerConfigOnDBObserver.java:62)
-       at 
org.apache.rocketmq.eventbridge.adapter.runtime.service.AbstractTargetRunnerConfigObserver.diff(AbstractTargetRunnerConfigObserver.java:110)
-       at 
org.apache.rocketmq.eventbridge.adapter.runtime.service.TargetRunnerConfigOnDBObserver.lambda$addListen$0(TargetRunnerConfigOnDBObserver.java:82)
-       at 
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
-       at 
java.util.concurrent.FutureTask.runAndReset$$$capture(FutureTask.java:308)
-       at java.util.concurrent.FutureTask.runAndReset(FutureTask.java)
-       at 
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
-       at 
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
-       at 
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
-       at 
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
-       at java.lang.Thread.run(Thread.java:748)
-2023-08-15 20:23:33 jdbc[3417]: exception
-org.h2.jdbc.JdbcSQLException: Database is already closed (to disable automatic 
closing at VM shutdown, add ";DB_CLOSE_ON_EXIT=FALSE" to the db URL) [90121-197]
-       at org.h2.message.DbException.getJdbcSQLException(DbException.java:357)
-       at org.h2.message.DbException.get(DbException.java:179)
-       at org.h2.message.DbException.get(DbException.java:155)
-       at org.h2.message.DbException.get(DbException.java:144)
-       at org.h2.jdbc.JdbcConnection.checkClosed(JdbcConnection.java:1526)
-       at org.h2.jdbc.JdbcConnection.checkClosed(JdbcConnection.java:1502)
-       at org.h2.jdbc.JdbcConnection.createStatement(JdbcConnection.java:217)
-       at com.zaxxer.hikari.pool.PoolBase.isConnectionAlive(PoolBase.java:153)
-       at com.zaxxer.hikari.pool.HikariPool.getConnection(HikariPool.java:172)
-       at com.zaxxer.hikari.pool.HikariPool.getConnection(HikariPool.java:148)
-       at 
com.zaxxer.hikari.HikariDataSource.getConnection(HikariDataSource.java:100)
-       at 
org.springframework.jdbc.datasource.DataSourceUtils.fetchConnection(DataSourceUtils.java:151)
-       at 
org.springframework.jdbc.datasource.DataSourceUtils.doGetConnection(DataSourceUtils.java:115)
-       at 
org.springframework.jdbc.datasource.DataSourceUtils.getConnection(DataSourceUtils.java:78)
-       at 
org.mybatis.spring.transaction.SpringManagedTransaction.openConnection(SpringManagedTransaction.java:80)
-       at 
org.mybatis.spring.transaction.SpringManagedTransaction.getConnection(SpringManagedTransaction.java:67)
-       at 
org.apache.ibatis.executor.BaseExecutor.getConnection(BaseExecutor.java:337)
-       at 
org.apache.ibatis.executor.SimpleExecutor.prepareStatement(SimpleExecutor.java:86)
-       at 
org.apache.ibatis.executor.SimpleExecutor.doQuery(SimpleExecutor.java:62)
-       at 
org.apache.ibatis.executor.BaseExecutor.queryFromDatabase(BaseExecutor.java:325)
-       at org.apache.ibatis.executor.BaseExecutor.query(BaseExecutor.java:156)
-       at 
org.apache.ibatis.executor.CachingExecutor.query(CachingExecutor.java:109)
-       at 
org.apache.ibatis.executor.CachingExecutor.query(CachingExecutor.java:89)
-       at 
org.apache.ibatis.session.defaults.DefaultSqlSession.selectList(DefaultSqlSession.java:147)
-       at 
org.apache.ibatis.session.defaults.DefaultSqlSession.selectList(DefaultSqlSession.java:140)
-       at sun.reflect.GeneratedMethodAccessor51.invoke(Unknown Source)
-       at 
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
-       at java.lang.reflect.Method.invoke(Method.java:498)
-       at 
org.mybatis.spring.SqlSessionTemplate$SqlSessionInterceptor.invoke(SqlSessionTemplate.java:426)
-       at com.sun.proxy.$Proxy81.selectList(Unknown Source)
-       at 
org.mybatis.spring.SqlSessionTemplate.selectList(SqlSessionTemplate.java:223)
-       at 
org.apache.ibatis.binding.MapperMethod.executeForMany(MapperMethod.java:147)
-       at org.apache.ibatis.binding.MapperMethod.execute(MapperMethod.java:80)
-       at 
org.apache.ibatis.binding.MapperProxy$PlainMethodInvoker.invoke(MapperProxy.java:152)
-       at org.apache.ibatis.binding.MapperProxy.invoke(MapperProxy.java:85)
-       at com.sun.proxy.$Proxy92.listEventTargetRunners(Unknown Source)
-       at 
org.apache.rocketmq.eventbridge.adapter.persistence.target.mybatis.repository.MybatisEventTargetRunnerRepository.listEventTargetRunners(MybatisEventTargetRunnerRepository.java:77)
-       at 
org.apache.rocketmq.eventbridge.adapter.persistence.target.mybatis.repository.MybatisEventTargetRunnerRepository$$FastClassBySpringCGLIB$$d4e0cf7.invoke(<generated>)
-       at 
org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:204)
-       at 
org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:746)
-       at 
org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
-       at 
org.springframework.dao.support.PersistenceExceptionTranslationInterceptor.invoke(PersistenceExceptionTranslationInterceptor.java:139)
-       at 
org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:185)
-       at 
org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:688)
-       at 
org.apache.rocketmq.eventbridge.adapter.persistence.target.mybatis.repository.MybatisEventTargetRunnerRepository$$EnhancerBySpringCGLIB$$ed257c41.listEventTargetRunners(<generated>)
-       at 
org.apache.rocketmq.eventbridge.adapter.runtime.service.TargetRunnerConfigOnDBObserver.getLatestTargetRunnerConfig(TargetRunnerConfigOnDBObserver.java:62)
-       at 
org.apache.rocketmq.eventbridge.adapter.runtime.service.AbstractTargetRunnerConfigObserver.diff(AbstractTargetRunnerConfigObserver.java:110)
-       at 
org.apache.rocketmq.eventbridge.adapter.runtime.service.TargetRunnerConfigOnDBObserver.lambda$addListen$0(TargetRunnerConfigOnDBObserver.java:82)
-       at 
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
-       at 
java.util.concurrent.FutureTask.runAndReset$$$capture(FutureTask.java:308)
-       at java.util.concurrent.FutureTask.runAndReset(FutureTask.java)
-       at 
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
-       at 
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
-       at 
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
-       at 
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
+       at org.h2.message.DbException.get(DbException.java:168)
+       at org.h2.value.ValueTimestamp.parse(ValueTimestamp.java:147)
+       at org.h2.value.Value.convertTo(Value.java:1071)
+       at org.h2.table.Column.convert(Column.java:177)
+       at org.h2.command.dml.Insert.insertRows(Insert.java:166)
+       at org.h2.command.dml.Insert.update(Insert.java:134)
+       at org.h2.command.CommandContainer.update(CommandContainer.java:102)
+       at org.h2.command.Command.executeUpdate(Command.java:261)
+       at 
org.h2.jdbc.JdbcPreparedStatement.executeUpdateInternal(JdbcPreparedStatement.java:199)
+       at 
org.h2.jdbc.JdbcPreparedStatement.executeUpdate(JdbcPreparedStatement.java:153)
+       at org.h2.result.UpdatableRow.insertRow(UpdatableRow.java:321)
+       at org.h2.jdbc.JdbcResultSet.insertRow(JdbcResultSet.java:2994)
+       at org.h2.server.web.WebApp.editResult(WebApp.java:1082)
+       at org.h2.server.web.WebApp.process(WebApp.java:227)
+       at org.h2.server.web.WebApp.processRequest(WebApp.java:170)
+       at org.h2.server.web.WebThread.process(WebThread.java:134)
+       at org.h2.server.web.WebThread.run(WebThread.java:90)
        at java.lang.Thread.run(Thread.java:748)
+Caused by: java.lang.StringIndexOutOfBoundsException: String index out of 
range: 0
+       at java.lang.String.charAt(String.java:658)
+       at org.h2.util.DateTimeUtils.parseDateValue(DateTimeUtils.java:337)
+       at org.h2.util.DateTimeUtils.parseTimestamp(DateTimeUtils.java:460)
+       at org.h2.value.ValueTimestamp.parse(ValueTimestamp.java:145)
+       ... 16 more
+2023-11-26 18:44:29 jdbc[8]: exception
+org.h2.jdbc.JdbcSQLException: Syntax error in SQL statement "SELECT * FROM 
EVENT_CLUSTER 
+
+
+
+
+ INSERT INTO[*] 
""EVENT_CLUSTER""(""ID"",""NAME"",""RESOURCES"",""REPLICA"",""IMAGE"",""MD5"",""GMT_CREATE"",""GMT_MODIFY"")VALUES(1,'demo','{\""cpu\"":
 
0.2,\""memory\"":0.5}',3,'registry.cn-beijing.cr.aliyuncs.com/eventbridge:20231115195431f55971','env=key',NOW(),NOW())
 "; SQL statement:
+SELECT * FROM EVENT_CLUSTER 
+
+
+
+
+ INSERT INTO 
"EVENT_CLUSTER"("ID","NAME","RESOURCES","REPLICA","IMAGE","MD5","GMT_CREATE","GMT_MODIFY")VALUES(1,'demo','{\"cpu\":
 
0.2,\"memory\":0.5}',3,'registry.cn-beijing.cr.aliyuncs.com/eventbridge:20231115195431f55971','env=key',now(),now())
 [42000-197]
+2023-11-26 18:44:55 jdbc[8]: exception
+org.h2.jdbc.JdbcSQLException: Syntax error in SQL statement "SELECT * FROM 
EVENT_CLUSTER 
+
+
+
+
+ INSERT INTO 
""EVENT_CLUSTER""(""ID"",""NAME"",""RESOURCES"",""REPLICA"",""IMAGE"",""MD5"",""GMT_CREATE"",""GMT_MODIFY"")VALUES(1,'demo','test,3,'registry.cn-beijing.cr.aliyuncs.com/eventbridge:20231115195431f55971','env=key[*]',now(),now());";
 SQL statement:
+SELECT * FROM EVENT_CLUSTER 
+
+
+
+
+ INSERT INTO 
"EVENT_CLUSTER"("ID","NAME","RESOURCES","REPLICA","IMAGE","MD5","GMT_CREATE","GMT_MODIFY")VALUES(1,'demo','test,3,'registry.cn-beijing.cr.aliyuncs.com/eventbridge:20231115195431f55971','env=key',now(),now());
 [42000-197]
diff --git a/scan.py b/scan.py
deleted file mode 100644
index e69de29..0000000
diff --git a/start/pom.xml b/start/pom.xml
index e1e1995..bfed07f 100644
--- a/start/pom.xml
+++ b/start/pom.xml
@@ -54,10 +54,10 @@
             <groupId>org.apache.rocketmq</groupId>
             <artifactId>rocketmq-eventbridge-infrastructure</artifactId>
         </dependency>
-        <dependency>
-            <groupId>org.apache.rocketmq</groupId>
-            <artifactId>rocketmq-test-demo</artifactId>
-        </dependency>
+<!--        <dependency>-->
+<!--            <groupId>org.apache.rocketmq</groupId>-->
+<!--            <artifactId>rocketmq-test-demo</artifactId>-->
+<!--        </dependency>-->
         <!-- Framework -->
         <dependency>
             <groupId>org.springframework.boot</groupId>
@@ -68,6 +68,11 @@
             <artifactId>spring-boot-starter-test</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            
<artifactId>rocketmq-eventbridge-adapter-runtime-manager</artifactId>
+            <groupId>org.apache.rocketmq</groupId>
+            <version>1.0.0</version>
+        </dependency>
         <!--Tools -->
         <dependency>
             <groupId>com.google.code.gson</groupId>
diff --git 
a/test/demo/src/main/java/org/apache/rocketmq/eventbridge/demo/DefaultDemo.java 
b/test/demo/src/main/java/org/apache/rocketmq/eventbridge/demo/DefaultDemo.java
index ce71c7a..bab366f 100644
--- 
a/test/demo/src/main/java/org/apache/rocketmq/eventbridge/demo/DefaultDemo.java
+++ 
b/test/demo/src/main/java/org/apache/rocketmq/eventbridge/demo/DefaultDemo.java
@@ -60,9 +60,9 @@ public class DefaultDemo {
     @PostConstruct
     public void initDemo() {
         log.info("init demo");
-        initEventBus();
-        initEventRule();
-        intEventTarget();
+//        initEventBus();
+//        initEventRule();
+//        intEventTarget();
 
     }

(rocketmq-eventbridge) branch feat/runtime-manager updated: feat:watch the cluster and worker.

Reply via email to