diff --git a/.gitignore b/.gitignore index 5373a3de..dabbc0de 100644 --- a/.gitignore +++ b/.gitignore @@ -21,4 +21,4 @@ client-adapter/example/ canal-admin/canal-admin-ui/dist canal-admin/canal-admin-ui/node -canal-admin/canal-admin-ui/node-modules +canal-admin/canal-admin-ui/node_modules diff --git a/canal-admin/canal-admin-server/pom.xml b/canal-admin/canal-admin-server/pom.xml index be47435d..165b1cf5 100644 --- a/canal-admin/canal-admin-server/pom.xml +++ b/canal-admin/canal-admin-server/pom.xml @@ -21,19 +21,13 @@ spring-boot-starter-jdbc - - org.javalite - activejdbc - mysql mysql-connector-java - commons-dbutils commons-dbutils - 1.6 diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/common/exception/ConfigurationException.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/common/exception/ConfigurationException.java new file mode 100644 index 00000000..617bffc8 --- /dev/null +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/common/exception/ConfigurationException.java @@ -0,0 +1,25 @@ +/* + * 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 com.alibaba.otter.canal.admin.common.exception; + +public class ConfigurationException extends RuntimeException{ + + public ConfigurationException(String message) { + super(message); + } +} diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/common/exception/ParamValidationException.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/common/exception/ParamValidationException.java new file mode 100644 index 00000000..96d2edab --- /dev/null +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/common/exception/ParamValidationException.java @@ -0,0 +1,31 @@ +/* + * 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 com.alibaba.otter.canal.admin.common.exception; + +import org.springframework.http.HttpStatus; +import org.springframework.web.bind.annotation.ResponseStatus; + +/** + * Parameter validation failure exception + */ +@ResponseStatus(value = HttpStatus.BAD_REQUEST) +public class ParamValidationException extends SystemException { + + public ParamValidationException(String message) { + super(message); + } +} diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/common/exception/PermissionDeniedException.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/common/exception/PermissionDeniedException.java new file mode 100644 index 00000000..d7d1148f --- /dev/null +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/common/exception/PermissionDeniedException.java @@ -0,0 +1,39 @@ +/* + * 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 com.alibaba.otter.canal.admin.common.exception; + +import org.springframework.http.HttpStatus; +import org.springframework.web.bind.annotation.ResponseStatus; + +/** + * Permission denied exception + */ +@ResponseStatus(value = HttpStatus.UNAUTHORIZED) +public class PermissionDeniedException extends RuntimeException { + + public PermissionDeniedException(String message) { + super(message); + } + + public PermissionDeniedException(String message, Throwable cause) { + super(message, cause); + } + + protected PermissionDeniedException() { + super(); + } +} diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/common/exception/ResourceNotFoundException.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/common/exception/ResourceNotFoundException.java new file mode 100644 index 00000000..4423966e --- /dev/null +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/common/exception/ResourceNotFoundException.java @@ -0,0 +1,43 @@ +/* + * 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 com.alibaba.otter.canal.admin.common.exception; + +import org.springframework.http.HttpStatus; +import org.springframework.web.bind.annotation.ResponseStatus; + +/** + * Resource not found exception + */ +@ResponseStatus(value = HttpStatus.NOT_FOUND) +public class ResourceNotFoundException extends SystemException { + + public ResourceNotFoundException() { + } + + public ResourceNotFoundException(String message) { + super(message); + } + + public ResourceNotFoundException(String message, Throwable cause) { + super(message, cause); + } + + public ResourceNotFoundException(Throwable cause) { + super(cause); + } + +} diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/common/exception/ServiceException.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/common/exception/ServiceException.java new file mode 100644 index 00000000..2b598555 --- /dev/null +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/common/exception/ServiceException.java @@ -0,0 +1,47 @@ +/* + * 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 com.alibaba.otter.canal.admin.common.exception; + +import org.springframework.http.HttpStatus; +import org.springframework.web.bind.annotation.ResponseStatus; + +/** + * Service Logic Exception + */ +@ResponseStatus(value = HttpStatus.SERVICE_UNAVAILABLE) +public class ServiceException extends RuntimeException { + + public ServiceException() { + } + + public ServiceException(String message) { + super(message); + } + + public ServiceException(String message, Throwable cause) { + super(message, cause); + } + + public ServiceException(Throwable cause) { + super(cause); + } + + public ServiceException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) { + super(message, cause, enableSuppression, writableStackTrace); + } + +} diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/common/exception/SystemException.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/common/exception/SystemException.java new file mode 100644 index 00000000..4507dbfa --- /dev/null +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/common/exception/SystemException.java @@ -0,0 +1,43 @@ +/* + * 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 com.alibaba.otter.canal.admin.common.exception; + +import org.springframework.http.HttpStatus; +import org.springframework.web.bind.annotation.ResponseStatus; + +/** + * System Exception + */ +@ResponseStatus(value = HttpStatus.INTERNAL_SERVER_ERROR) +public class SystemException extends RuntimeException { + + public SystemException() { + super(); + } + + public SystemException(String message) { + super(message); + } + + public SystemException(String message, Throwable cause) { + super(message, cause); + } + + public SystemException(Throwable cause) { + super(cause); + } +} diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/common/exception/VersionValidationException.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/common/exception/VersionValidationException.java new file mode 100644 index 00000000..f02c8cbc --- /dev/null +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/common/exception/VersionValidationException.java @@ -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. + */ + +package com.alibaba.otter.canal.admin.common.exception; + + +import org.springframework.http.HttpStatus; +import org.springframework.web.bind.annotation.ResponseStatus; + +@ResponseStatus(value = HttpStatus.BAD_REQUEST) +public class VersionValidationException extends SystemException{ + + public VersionValidationException(String message) { + super(message); + } +} diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/controller/CanalConfigController.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/controller/CanalConfigController.java new file mode 100644 index 00000000..f56cecad --- /dev/null +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/controller/CanalConfigController.java @@ -0,0 +1,23 @@ +package com.alibaba.otter.canal.admin.controller; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestMethod; +import org.springframework.web.bind.annotation.RestController; + +import com.alibaba.otter.canal.admin.model.CanalConfig; +import com.alibaba.otter.canal.admin.service.CanalConfigService; + +@RestController +@RequestMapping("/api/{env}/config") +public class CanalConfigController { + + @Autowired + CanalConfigService canalConfigService; + + @RequestMapping(value = "/canal", method = RequestMethod.GET) + public CanalConfig canalConfig(@PathVariable String env) { + return canalConfigService.getCanalConfig(); + } +} diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/dao/CanalConfigDao.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/dao/CanalConfigDao.java index 24ceb05d..71f177c0 100644 --- a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/dao/CanalConfigDao.java +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/dao/CanalConfigDao.java @@ -6,35 +6,30 @@ import org.apache.commons.dbutils.handlers.BeanHandler; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Repository; -import javax.annotation.PostConstruct; import java.sql.SQLException; import java.util.Date; @Repository public class CanalConfigDao { + @Autowired private QueryRunner runner; - @PostConstruct - public void init() { + public CanalConfig findById(Long id) { try { - CanalConfig canalConfig = findById(1L); - canalConfig = canalConfig; - canalConfig.setContent(canalConfig.getContent()+" xxx"); - updateContent(canalConfig); + String sql = "select id,name,content,modified_time as modifiedTime" + " from canal_config where id=?"; + return runner.query(sql, new BeanHandler<>(CanalConfig.class), id); } catch (SQLException e) { - e.printStackTrace(); + throw new RuntimeException(e.getMessage(), e); } } - public CanalConfig findById(Long id) throws SQLException { - String sql = "select id,name,content,modified_time as modifiedTime" + - " from canal_config where id=?"; - return runner.query(sql, new BeanHandler<>(CanalConfig.class), id); - } - - public void updateContent(CanalConfig canalConfig) throws SQLException { - String sql = "update canal_config set content=?, modified_time=? where id=?"; - runner.update(sql, canalConfig.getContent(), new Date(), canalConfig.getId()); + public int updateContent(CanalConfig canalConfig){ + try { + String sql = "update canal_config set content=?, modified_time=? where id=?"; + return runner.update(sql, canalConfig.getContent(), new Date(), canalConfig.getId()); + } catch (SQLException e) { + throw new RuntimeException(e.getMessage(), e); + } } } diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/model/CanalConfig.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/model/CanalConfig.java index a3855de3..489bdc1b 100644 --- a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/model/CanalConfig.java +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/model/CanalConfig.java @@ -1,16 +1,13 @@ package com.alibaba.otter.canal.admin.model; -import org.javalite.activejdbc.Model; -import org.javalite.activejdbc.annotations.Table; - -import java.sql.Timestamp; import java.util.Date; public class CanalConfig { - private Long id; + + private Long id; private String name; private String content; - private Date modifiedTime; + private Date modifiedTime; public Long getId() { return id; diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/model/Person.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/model/Person.java deleted file mode 100644 index 496a1643..00000000 --- a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/model/Person.java +++ /dev/null @@ -1,7 +0,0 @@ -package com.alibaba.otter.canal.admin.model; - -import org.javalite.activejdbc.Model; - -public class Person extends Model { - -} \ No newline at end of file diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/CanalConfigService.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/CanalConfigService.java new file mode 100644 index 00000000..92c69705 --- /dev/null +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/CanalConfigService.java @@ -0,0 +1,10 @@ +package com.alibaba.otter.canal.admin.service; + +import com.alibaba.otter.canal.admin.model.CanalConfig; + +public interface CanalConfigService { + + CanalConfig getCanalConfig(); + + CanalConfig getAdapterConfig(); +} diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/impl/CanalConfigServiceImpl.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/impl/CanalConfigServiceImpl.java new file mode 100644 index 00000000..3af022e2 --- /dev/null +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/impl/CanalConfigServiceImpl.java @@ -0,0 +1,27 @@ +package com.alibaba.otter.canal.admin.service.impl; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Service; + +import com.alibaba.otter.canal.admin.dao.CanalConfigDao; +import com.alibaba.otter.canal.admin.model.CanalConfig; +import com.alibaba.otter.canal.admin.service.CanalConfigService; + +@Service +public class CanalConfigServiceImpl implements CanalConfigService { + + @Autowired + CanalConfigDao canalConfigDao; + + public CanalConfig getCanalConfig() { + return canalConfigDao.findById(1L); + } + + public CanalConfig getAdapterConfig() { + return canalConfigDao.findById(2L); + } + + public void updateContent(CanalConfig canalConfig) { + canalConfigDao.findById(canalConfig.getId()); + } +} diff --git a/canal-admin/pom.xml b/canal-admin/pom.xml index b997eafe..440c7df7 100644 --- a/canal-admin/pom.xml +++ b/canal-admin/pom.xml @@ -33,16 +33,16 @@ import - - org.javalite - activejdbc - 2.3 - mysql mysql-connector-java 5.1.40 + + commons-dbutils + commons-dbutils + 1.6 + diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java index de512923..b86ea503 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java @@ -6,6 +6,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import com.alibaba.fastjson.JSON; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -151,6 +152,8 @@ public abstract class AbstractCanalAdapterWorker { if (flatMessage) { // batch write writeOut((List) messages); + //FIXME xxx + messages.forEach((message -> System.out.println(JSON.toJSONString(message)))); } else { for (final Object message : messages) { writeOut((Message) message); diff --git a/client-adapter/launcher/src/main/resources/application.yml b/client-adapter/launcher/src/main/resources/application.yml index 14bc3e6e..7c455118 100644 --- a/client-adapter/launcher/src/main/resources/application.yml +++ b/client-adapter/launcher/src/main/resources/application.yml @@ -7,10 +7,10 @@ spring: default-property-inclusion: non_null canal.conf: - mode: tcp # kafka rocketMQ - canalServerHost: 127.0.0.1:11111 + mode: kafka # kafka rocketMQ +# canalServerHost: 127.0.0.1:11111 # zookeeperHosts: slave1:2181 -# mqServers: 127.0.0.1:9092 #or rocketmq nameservers + mqServers: 127.0.0.1:9092 #or rocketmq # flatMessage: true batchSize: 500 syncBatchSize: 1000 @@ -18,11 +18,6 @@ canal.conf: timeout: accessKey: secretKey: -# enableMessageTrace: -# accessChannel: -# customizedTraceTopic: -# namespace: - # srcDataSources: # defaultDS: # url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index 106005f1..e01b8028 100644 --- a/deployer/src/main/resources/canal.properties +++ b/deployer/src/main/resources/canal.properties @@ -13,7 +13,7 @@ canal.zkServers = canal.zookeeper.flush.period = 1000 canal.withoutNetty = false # tcp, kafka, RocketMQ -canal.serverMode = tcp +canal.serverMode = kafka # flush meta cursor/parse position to file canal.file.data.dir = ${canal.conf.dir} canal.file.flush.period = 1000 @@ -67,7 +67,7 @@ canal.instance.parser.parallel = true canal.instance.parser.parallelBufferSize = 256 # table meta tsdb info -canal.instance.tsdb.enable = true +canal.instance.tsdb.enable = false canal.instance.tsdb.dir = ${canal.file.data.dir:../conf}/${canal.instance.destination:} canal.instance.tsdb.url = jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL; canal.instance.tsdb.dbUsername = canal @@ -104,7 +104,7 @@ canal.instance.global.spring.xml = classpath:spring/file-instance.xml ################################################## ######### MQ ############# ################################################## -canal.mq.servers = 127.0.0.1:6667 +canal.mq.servers = 127.0.0.1:9092 canal.mq.retries = 0 canal.mq.batchSize = 16384 canal.mq.maxRequestSize = 1048576 diff --git a/deployer/src/main/resources/example/instance.properties b/deployer/src/main/resources/example/instance.properties index fda81838..bbacf00d 100644 --- a/deployer/src/main/resources/example/instance.properties +++ b/deployer/src/main/resources/example/instance.properties @@ -18,7 +18,7 @@ canal.instance.rds.secretkey= canal.instance.rds.instanceId= # table meta tsdb info -canal.instance.tsdb.enable=true +canal.instance.tsdb.enable=false #canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb #canal.instance.tsdb.dbUsername=canal #canal.instance.tsdb.dbPassword=canal @@ -30,8 +30,8 @@ canal.instance.tsdb.enable=true #canal.instance.standby.gtid= # username/password -canal.instance.dbUsername=canal -canal.instance.dbPassword=canal +canal.instance.dbUsername=root +canal.instance.dbPassword=121212 canal.instance.connectionCharset = UTF-8 # enable druid Decrypt database password canal.instance.enableDruid=false