create()
+ .register("http", PlainConnectionSocketFactory.INSTANCE)
+ .register("https", new SSLConnectionSocketFactory(sslContext, NoopHostnameVerifier.INSTANCE))
+ .build()))
+ .build();
+ } else {
+ httpClient = HttpClientBuilder.create().setMaxConnPerRoute(50).setMaxConnTotal(100).build();
+ }
+
HttpGet httpGet = new HttpGet(downloadLink);
- CloseableHttpClient httpClient = HttpClientBuilder.create().setMaxConnPerRoute(50).setMaxConnTotal(100).build();
RequestConfig requestConfig = RequestConfig.custom()
.setConnectTimeout(TIMEOUT)
.setConnectionRequestTimeout(TIMEOUT)
@@ -187,6 +227,9 @@ public class BinlogDownloadQueue {
String name = tarArchiveEntry.getName();
File tarFile = new File(parentFile, name + ".tmp");
logger.info("start to download file " + tarFile.getName());
+ if (tarFile.exists()) {
+ tarFile.delete();
+ }
BufferedOutputStream bos = null;
try {
bos = new BufferedOutputStream(new FileOutputStream(tarFile));
@@ -204,6 +247,10 @@ public class BinlogDownloadQueue {
tais.close();
} else {
File file = new File(parentFile, fileName + ".tmp");
+ if (file.exists()) {
+ file.delete();
+ }
+
if (!file.isFile()) {
file.createNewFile();
}
@@ -245,17 +292,45 @@ public class BinlogDownloadQueue {
@Override
public void run() {
while (running) {
+ BinlogFile binlogFile = null;
try {
- BinlogFile binlogFile = downloadQueue.poll(5000, TimeUnit.MILLISECONDS);
+ binlogFile = downloadQueue.poll(5000, TimeUnit.MILLISECONDS);
if (binlogFile != null) {
- download(binlogFile);
+ int retry = 1;
+ while (true) {
+ try {
+ download(binlogFile);
+ break;
+ } catch (Throwable e) {
+ if (retry % 10 == 0) {
+ retry = retry + 1;
+ try {
+ logger.warn("download failed + " + binlogFile.toString() + "], retry : "
+ + retry, e);
+ // File errorFile = new File(destDir,
+ // "error.txt");
+ // FileWriter writer = new
+ // FileWriter(errorFile);
+ // writer.write(ExceptionUtils.getFullStackTrace(e));
+ // writer.flush();
+ // IOUtils.closeQuietly(writer);
+ LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(100 * retry));
+ } catch (Throwable e1) {
+ logger.error("write error failed", e1);
+ }
+ } else {
+ retry = retry + 1;
+ }
+ }
+ }
}
+
Runnable runnable = taskQueue.poll(5000, TimeUnit.MILLISECONDS);
if (runnable != null) {
runnable.run();
}
- } catch (Exception e) {
- e.printStackTrace();
+ } catch (Throwable e) {
+ logger.error("task process failed", e);
}
}
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/HttpHelper.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/HttpHelper.java
index 7b641ade..9ed865fe 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/HttpHelper.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/HttpHelper.java
@@ -155,6 +155,7 @@ public class HttpHelper {
.register("https", new SSLConnectionSocketFactory(sslContext, NoopHostnameVerifier.INSTANCE))
.build()))
.build();
+
// ---------------- 创建支持https 的client成功---------
URI uri = new URIBuilder(url).build();
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsBinlogEventParserProxy.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsBinlogEventParserProxy.java
index 88caca53..3c38ed0c 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsBinlogEventParserProxy.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsBinlogEventParserProxy.java
@@ -4,93 +4,109 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
+import org.apache.commons.lang.StringUtils;
+
import com.alibaba.otter.canal.parse.exception.PositionNotFoundException;
import com.alibaba.otter.canal.parse.inbound.ParserExceptionHandler;
import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser;
/**
+ * aliyun rds的binlog parser支持
+ *
+ *
+ * 注意点：aliyun的binlog会有定期清理并备份到oss上, 这里实现了一份自动下载oss+rds binlog的机制
+ *
+ *
* @author chengjin.lyf on 2018/7/20 上午10:52
* @since 1.0.25
*/
public class RdsBinlogEventParserProxy extends MysqlEventParser {
- private String rdsOpenApiUrl = "https://rds.aliyuncs.com/"; // openapi地址
- private String accesskey; // 云账号的ak
- private String secretkey; // 云账号sk
- private String instanceId; // rds实例id
- private Long startTime;
- private Long endTime;
- private String directory; // binlog
- // 目录
- private int batchSize = 4; // 最多下载的binlog文件数量
+ private String rdsOpenApiUrl = "https://rds.aliyuncs.com/"; // openapi地址
+ private String accesskey; // 云账号的ak
+ private String secretkey; // 云账号sk
+ private String instanceId; // rds实例id
+ private String directory; // binlog目录
+ private int batchFileSize = 4; // 最多下载的binlog文件数量
- private RdsLocalBinlogEventParser rdsBinlogEventParser = new RdsLocalBinlogEventParser();
- private ExecutorService executorService = Executors.newSingleThreadExecutor(new ThreadFactory() {
+ private RdsLocalBinlogEventParser rdsLocalBinlogEventParser = new RdsLocalBinlogEventParser();
+ private ExecutorService executorService = Executors.newSingleThreadExecutor(new ThreadFactory() {
- @Override
- public Thread newThread(Runnable r) {
- Thread t = new Thread(r, "rds-binlog-daemon-thread");
- t.setDaemon(true);
- return t;
- }
- });
+ @Override
+ public Thread newThread(Runnable r) {
+ Thread t = new Thread(r,
+ "rds-binlog-daemon-thread");
+ t.setDaemon(true);
+ return t;
+ }
+ });
@Override
public void start() {
- final ParserExceptionHandler targetHandler = this.getParserExceptionHandler();
- rdsBinlogEventParser.setLogPositionManager(this.getLogPositionManager());
- rdsBinlogEventParser.setDestination(destination);
- rdsBinlogEventParser.setAlarmHandler(this.getAlarmHandler());
- rdsBinlogEventParser.setConnectionCharset(this.connectionCharset);
- rdsBinlogEventParser.setConnectionCharsetNumber(this.connectionCharsetNumber);
- rdsBinlogEventParser.setEnableTsdb(this.enableTsdb);
- rdsBinlogEventParser.setEventBlackFilter(this.eventBlackFilter);
- rdsBinlogEventParser.setFilterQueryDcl(this.filterQueryDcl);
- rdsBinlogEventParser.setFilterQueryDdl(this.filterQueryDdl);
- rdsBinlogEventParser.setFilterQueryDml(this.filterQueryDml);
- rdsBinlogEventParser.setFilterRows(this.filterRows);
- rdsBinlogEventParser.setFilterTableError(this.filterTableError);
- rdsBinlogEventParser.setIsGTIDMode(this.isGTIDMode);
- rdsBinlogEventParser.setMasterInfo(this.masterInfo);
- rdsBinlogEventParser.setEventFilter(this.eventFilter);
- rdsBinlogEventParser.setMasterPosition(this.masterPosition);
- rdsBinlogEventParser.setTransactionSize(this.transactionSize);
- rdsBinlogEventParser.setUrl(this.rdsOpenApiUrl);
- rdsBinlogEventParser.setAccesskey(this.accesskey);
- rdsBinlogEventParser.setSecretkey(this.secretkey);
- rdsBinlogEventParser.setInstanceId(this.instanceId);
- rdsBinlogEventParser.setEventSink(eventSink);
- rdsBinlogEventParser.setDirectory(directory);
- rdsBinlogEventParser.setBatchSize(batchSize);
- rdsBinlogEventParser.setFinishListener(new RdsLocalBinlogEventParser.ParseFinishListener() {
-
- @Override
- public void onFinish() {
- executorService.execute(new Runnable() {
-
- @Override
- public void run() {
- rdsBinlogEventParser.stop();
- RdsBinlogEventParserProxy.this.start();
- }
- });
-
+ if (StringUtils.isNotEmpty(accesskey) && StringUtils.isNotEmpty(secretkey)
+ && StringUtils.isNotEmpty(instanceId)) {
+ final ParserExceptionHandler targetHandler = this.getParserExceptionHandler();
+ if (directory == null) {
+ directory = System.getProperty("java.io.tmpdir", "/tmp") + "/" + destination;
}
- });
- this.setParserExceptionHandler(new ParserExceptionHandler() {
+ rdsLocalBinlogEventParser.setLogPositionManager(this.getLogPositionManager());
+ rdsLocalBinlogEventParser.setDestination(destination);
+ rdsLocalBinlogEventParser.setAlarmHandler(this.getAlarmHandler());
+ rdsLocalBinlogEventParser.setConnectionCharset(this.connectionCharset);
+ rdsLocalBinlogEventParser.setConnectionCharsetNumber(this.connectionCharsetNumber);
+ rdsLocalBinlogEventParser.setEnableTsdb(this.enableTsdb);
+ rdsLocalBinlogEventParser.setEventBlackFilter(this.eventBlackFilter);
+ rdsLocalBinlogEventParser.setFilterQueryDcl(this.filterQueryDcl);
+ rdsLocalBinlogEventParser.setFilterQueryDdl(this.filterQueryDdl);
+ rdsLocalBinlogEventParser.setFilterQueryDml(this.filterQueryDml);
+ rdsLocalBinlogEventParser.setFilterRows(this.filterRows);
+ rdsLocalBinlogEventParser.setFilterTableError(this.filterTableError);
+ // rdsLocalBinlogEventParser.setIsGTIDMode(this.isGTIDMode);
+ rdsLocalBinlogEventParser.setMasterInfo(this.masterInfo);
+ rdsLocalBinlogEventParser.setEventFilter(this.eventFilter);
+ rdsLocalBinlogEventParser.setMasterPosition(this.masterPosition);
+ rdsLocalBinlogEventParser.setTransactionSize(this.transactionSize);
+ rdsLocalBinlogEventParser.setUrl(this.rdsOpenApiUrl);
+ rdsLocalBinlogEventParser.setAccesskey(this.accesskey);
+ rdsLocalBinlogEventParser.setSecretkey(this.secretkey);
+ rdsLocalBinlogEventParser.setInstanceId(this.instanceId);
+ rdsLocalBinlogEventParser.setEventSink(eventSink);
+ rdsLocalBinlogEventParser.setDirectory(directory);
+ rdsLocalBinlogEventParser.setBatchFileSize(batchFileSize);
+ rdsLocalBinlogEventParser.setParallel(this.parallel);
+ rdsLocalBinlogEventParser.setParallelBufferSize(this.parallelBufferSize);
+ rdsLocalBinlogEventParser.setParallelThreadSize(this.parallelThreadSize);
+ rdsLocalBinlogEventParser.setFinishListener(new RdsLocalBinlogEventParser.ParseFinishListener() {
+
+ @Override
+ public void onFinish() {
+ executorService.execute(new Runnable() {
+
+ @Override
+ public void run() {
+ rdsLocalBinlogEventParser.stop();
+ RdsBinlogEventParserProxy.this.start();
+ }
+ });
- @Override
- public void handle(Throwable e) {
- handleMysqlParserException(e);
- if (targetHandler != null) {
- targetHandler.handle(e);
}
- }
- });
+ });
+ this.setParserExceptionHandler(new ParserExceptionHandler() {
+
+ @Override
+ public void handle(Throwable e) {
+ handleMysqlParserException(e);
+ if (targetHandler != null) {
+ targetHandler.handle(e);
+ }
+ }
+ });
+ }
+
super.start();
}
- public void handleMysqlParserException(Throwable throwable) {
+ private void handleMysqlParserException(Throwable throwable) {
if (throwable instanceof PositionNotFoundException) {
logger.info("remove rds not found position, try download rds binlog!");
executorService.execute(new Runnable() {
@@ -101,12 +117,12 @@ public class RdsBinlogEventParserProxy extends MysqlEventParser {
logger.info("stop mysql parser!");
RdsBinlogEventParserProxy rdsBinlogEventParserProxy = RdsBinlogEventParserProxy.this;
long serverId = rdsBinlogEventParserProxy.getServerId();
- rdsBinlogEventParser.setServerId(serverId);
+ rdsLocalBinlogEventParser.setServerId(serverId);
rdsBinlogEventParserProxy.stop();
logger.info("start rds mysql binlog parser!");
- rdsBinlogEventParser.start();
- } catch (Exception e) {
- e.printStackTrace();
+ rdsLocalBinlogEventParser.start();
+ } catch (Throwable e) {
+ logger.info("handle exception failed", e);
}
}
});
@@ -143,7 +159,8 @@ public class RdsBinlogEventParserProxy extends MysqlEventParser {
this.directory = directory;
}
- public void setBatchSize(int batchSize) {
- this.batchSize = batchSize;
+ public void setBatchFileSize(int batchFileSize) {
+ this.batchFileSize = batchFileSize;
}
+
}
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsBinlogOpenApi.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsBinlogOpenApi.java
index 356a3520..2e79c536 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsBinlogOpenApi.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsBinlogOpenApi.java
@@ -1,51 +1,20 @@
package com.alibaba.otter.canal.parse.inbound.mysql.rds;
-import io.netty.handler.codec.http.HttpResponseStatus;
-
-import java.io.BufferedOutputStream;
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.UnsupportedEncodingException;
import java.net.URI;
import java.net.URISyntaxException;
-import java.net.URLEncoder;
-import java.text.SimpleDateFormat;
import java.util.Collections;
import java.util.Date;
-import java.util.HashMap;
import java.util.List;
-import java.util.Map;
-import java.util.TimeZone;
-import java.util.TreeMap;
-import java.util.UUID;
-import javax.crypto.Mac;
-import javax.crypto.SecretKey;
-import javax.crypto.spec.SecretKeySpec;
-
-import org.apache.commons.codec.binary.Base64;
-import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
-import org.apache.commons.compress.archivers.tar.TarArchiveInputStream;
-import org.apache.commons.io.FileUtils;
-import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.StringUtils;
-import org.apache.http.HttpResponse;
-import org.apache.http.client.config.RequestConfig;
-import org.apache.http.client.methods.HttpGet;
-import org.apache.http.impl.client.CloseableHttpClient;
-import org.apache.http.impl.client.HttpClientBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.alibaba.fastjson.JSON;
-import com.alibaba.fastjson.JSONArray;
-import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.parse.inbound.mysql.rds.data.BinlogFile;
import com.alibaba.otter.canal.parse.inbound.mysql.rds.data.DescribeBinlogFileResult;
+import com.alibaba.otter.canal.parse.inbound.mysql.rds.data.RdsBackupPolicy;
import com.alibaba.otter.canal.parse.inbound.mysql.rds.data.RdsItem;
+import com.alibaba.otter.canal.parse.inbound.mysql.rds.request.DescribeBackupPolicyRequest;
import com.alibaba.otter.canal.parse.inbound.mysql.rds.request.DescribeBinlogFilesRequest;
/**
@@ -54,13 +23,7 @@ import com.alibaba.otter.canal.parse.inbound.mysql.rds.request.DescribeBinlogFil
*/
public class RdsBinlogOpenApi {
- protected static final Logger logger = LoggerFactory.getLogger(RdsBinlogOpenApi.class);
- private static final String ISO8601_DATE_FORMAT = "yyyy-MM-dd'T'HH:mm:ss'Z'";
- private static final int TIMEOUT = 10000;
- private static final String ENCODING = "UTF-8";
- private static final String MAC_NAME = "HmacSHA1";
- private static final String API_VERSION = "2014-08-15";
- private static final String SIGNATURE_VERSION = "1.0";
+ protected static final Logger logger = LoggerFactory.getLogger(RdsBinlogOpenApi.class);
public static List listBinlogFiles(String url, String ak, String sk, String dbInstanceId,
Date startTime, Date endTime) {
@@ -106,279 +69,32 @@ public class RdsBinlogOpenApi {
return Collections.EMPTY_LIST;
}
- public static void downloadBinlogFiles(String url, String ak, String sk, String dbInstanceId, Date startTime,
- Date endTime, File destDir) throws Throwable {
- int pageSize = 100;
- int pageNumber = 0;
- int pageRecordCount = 1;
- String hostInstanceID = null;
- while (pageRecordCount > 0 && pageRecordCount <= pageSize) {
- pageNumber += 1;
- String result = describeBinlogFiles(url, ak, sk, dbInstanceId, startTime, endTime, pageSize, pageNumber);
- JSONObject jsobObj = JSON.parseObject(result);
- pageRecordCount = jsobObj.getInteger("PageRecordCount");
-
- if (pageRecordCount > 0) {
- FileUtils.forceMkdir(destDir);
- File hostIdFile = new File(destDir, "hostId");
- if (hostIdFile.exists()) {
- List lines = IOUtils.readLines(new FileInputStream(hostIdFile));
- hostInstanceID = StringUtils.join(lines, "\n");
- }
-
- String itemStr = jsobObj.getString("Items");
- JSONObject binLogFileObj = JSONObject.parseObject(itemStr);
- JSONArray items = binLogFileObj.getJSONArray("BinLogFile");
- if (items == null || items.isEmpty()) {
- continue;
- }
- for (int i = 0; i < items.size(); i++) {
- JSONObject item = (JSONObject) items.get(i);
- String oneHostInstanceID = item.getString("HostInstanceID");
- if (hostInstanceID == null) {
- hostInstanceID = oneHostInstanceID;
- FileOutputStream hostIdFileOut = null;
- try {
- hostIdFileOut = new FileOutputStream(hostIdFile);
- hostIdFileOut.write(oneHostInstanceID.getBytes());
- hostIdFileOut.flush();
- } finally {
- IOUtils.closeQuietly(hostIdFileOut);
- }
- }
-
- if (hostInstanceID.equals(oneHostInstanceID)) { // 只选择一个host下载
- String downloadLink = item.getString("DownloadLink");
- String fileName = StringUtils.substringBetween(downloadLink, "mysql-bin.", ".tar");
- if (StringUtils.isNotEmpty(fileName)) {
- File currentFile = new File(destDir, "mysql-bin." + fileName);
- if (currentFile.isFile() && currentFile.exists()) {
- // 检查一下文件是否存在，存在就就没必要下载了
- continue;
- }
- }
-
- HttpGet httpGet = new HttpGet(downloadLink);
- CloseableHttpClient httpClient = HttpClientBuilder.create()
- .setMaxConnPerRoute(50)
- .setMaxConnTotal(100)
- .build();
- RequestConfig requestConfig = RequestConfig.custom()
- .setConnectTimeout(TIMEOUT)
- .setConnectionRequestTimeout(TIMEOUT)
- .setSocketTimeout(TIMEOUT)
- .build();
- httpGet.setConfig(requestConfig);
- HttpResponse response = httpClient.execute(httpGet);
- int statusCode = response.getStatusLine().getStatusCode();
- if (statusCode != HttpResponseStatus.OK.code()) {
- throw new RuntimeException("download failed , url:" + downloadLink + " , statusCode:"
- + statusCode);
- }
- saveFile(destDir, response);
- }
- }
+ public static RdsBackupPolicy queryBinlogBackupPolicy(String url, String ak, String sk, String dbInstanceId) {
+ DescribeBackupPolicyRequest request = new DescribeBackupPolicyRequest();
+ if (StringUtils.isNotEmpty(url)) {
+ try {
+ URI uri = new URI(url);
+ request.setEndPoint(uri.getHost());
+ } catch (URISyntaxException e) {
+ logger.error("resolve url host failed, will use default rds endpoint!");
}
}
- }
-
- private static void saveFile(File parentFile, HttpResponse response) throws IOException {
- InputStream is = response.getEntity().getContent();
- long totalSize = Long.parseLong(response.getFirstHeader("Content-Length").getValue());
- String fileName = response.getFirstHeader("Content-Disposition").getValue();
- fileName = StringUtils.substringAfter(fileName, "filename=");
- boolean isTar = StringUtils.endsWith(fileName, ".tar");
- FileUtils.forceMkdir(parentFile);
- FileOutputStream fos = null;
- try {
- if (isTar) {
- TarArchiveInputStream tais = new TarArchiveInputStream(is);
- TarArchiveEntry tarArchiveEntry = null;
- while ((tarArchiveEntry = tais.getNextTarEntry()) != null) {
- String name = tarArchiveEntry.getName();
- File tarFile = new File(parentFile, name);
- logger.info("start to download file " + tarFile.getName());
- BufferedOutputStream bos = null;
- try {
- bos = new BufferedOutputStream(new FileOutputStream(tarFile));
- int read = -1;
- byte[] buffer = new byte[1024];
- while ((read = tais.read(buffer)) != -1) {
- bos.write(buffer, 0, read);
- }
- logger.info("download file " + tarFile.getName() + " end!");
- } finally {
- IOUtils.closeQuietly(bos);
- }
- }
- tais.close();
- } else {
- File file = new File(parentFile, fileName);
- if (!file.isFile()) {
- file.createNewFile();
+ request.setRdsInstanceId(dbInstanceId);
+ request.setAccessKeyId(ak);
+ request.setAccessKeySecret(sk);
+ int retryTime = 3;
+ while (true) {
+ try {
+ return request.doAction();
+ } catch (Exception e) {
+ if (retryTime-- <= 0) {
+ throw new RuntimeException(e);
}
try {
- fos = new FileOutputStream(file);
- byte[] buffer = new byte[1024];
- int len;
- long copySize = 0;
- long nextPrintProgress = 0;
- logger.info("start to download file " + file.getName());
- while ((len = is.read(buffer)) != -1) {
- fos.write(buffer, 0, len);
- copySize += len;
- long progress = copySize * 100 / totalSize;
- if (progress >= nextPrintProgress) {
- logger.info("download " + file.getName() + " progress : " + progress
- + "% , download size : " + copySize + ", total size : " + totalSize);
- nextPrintProgress += 10;
- }
- }
- logger.info("download file " + file.getName() + " end!");
- fos.flush();
- } finally {
- IOUtils.closeQuietly(fos);
+ Thread.sleep(100L);
+ } catch (InterruptedException e1) {
}
}
- } finally {
- IOUtils.closeQuietly(fos);
}
}
-
- public static String describeBinlogFiles(String url, String ak, String sk, String dbInstanceId, Date startTime,
- Date endTime, int pageSize, int pageNumber) throws Exception {
- Map paramMap = new HashMap();
- paramMap.put("Action", "DescribeBinlogFiles");
- paramMap.put("DBInstanceId", dbInstanceId); // rds实例id
- paramMap.put("StartTime", formatIso8601Date(startTime));
- paramMap.put("EndTime", formatIso8601Date(endTime));
- paramMap.put("PageSize", String.valueOf(pageSize));
- paramMap.put("PageNumber", String.valueOf(pageNumber));
- return doRequest(url, paramMap, ak, sk);
- }
-
- private static String doRequest(String domin, Map param, String ak, String sk) throws Exception {
- param.put("AccessKeyId", ak);
- param.put("SignatureMethod", "HMAC-SHA1");
- param.put("SignatureVersion", SIGNATURE_VERSION);
- param.put("Version", API_VERSION);
- param.put("SignatureNonce", UUID.randomUUID().toString());
- param.put("Format", "JSON");
- param.put("Timestamp", formatIso8601Date(new Date()));
- String signStr = generate("POST", param, sk);
- param.put("Signature", signStr);
- String request = concatQueryString(param);
- String url = domin + "?" + request;
- String result = HttpHelper.post(url, null, Collections.EMPTY_MAP, TIMEOUT);
- return result;
- }
-
- public static String concatQueryString(Map parameters) throws UnsupportedEncodingException {
- if (null == parameters) {
- return null;
- }
- StringBuilder urlBuilder = new StringBuilder("");
- for (Map.Entry entry : parameters.entrySet()) {
- String key = entry.getKey();
- String val = entry.getValue();
- urlBuilder.append(encode(key));
- if (val != null) {
- urlBuilder.append("=").append(encode(val));
- }
- urlBuilder.append("&");
- }
- int strIndex = urlBuilder.length();
- if (parameters.size() > 0) {
- urlBuilder.deleteCharAt(strIndex - 1);
- }
- return urlBuilder.toString();
- }
-
- public static String encode(String value) throws UnsupportedEncodingException {
- return URLEncoder.encode(value, "UTF-8");
- }
-
- private static String formatIso8601Date(Date date) {
- SimpleDateFormat df = new SimpleDateFormat(ISO8601_DATE_FORMAT);
- df.setTimeZone(TimeZone.getTimeZone("GMT"));
- return df.format(date);
- }
-
- /**
- * 使用 HMAC-SHA1 签名方法对对encryptText进行签名
- *
- * @param encryptText 被签名的字符串
- * @param encryptKey 密钥
- * @return
- * @throws Exception
- */
- public static byte[] HmacSHA1Encrypt(String encryptText, String encryptKey) throws Exception {
- byte[] data = encryptKey.getBytes(ENCODING);
- // 根据给定的字节数组构造一个密钥,第二参数指定一个密钥算法的名称
- SecretKey secretKey = new SecretKeySpec(data, MAC_NAME);
- // 生成一个指定 Mac 算法 的 Mac 对象
- Mac mac = Mac.getInstance(MAC_NAME);
- // 用给定密钥初始化 Mac 对象
- mac.init(secretKey);
-
- byte[] text = encryptText.getBytes(ENCODING);
- // 完成 Mac 操作
- return mac.doFinal(text);
- }
-
- private static String base64(byte input[]) throws UnsupportedEncodingException {
- return new String(Base64.encodeBase64(input), ENCODING);
- }
-
- /** 对参数名称和参数值进行URL编码 **/
- public static String generate(String method, Map parameter, String accessKeySecret)
- throws Exception {
- String signString = generateSignString(method, parameter);
- byte[] signBytes = HmacSHA1Encrypt(signString, accessKeySecret + "&");
- String signature = base64(signBytes);
- if ("POST".equals(method)) {
- return signature;
- }
- return URLEncoder.encode(signature, "UTF-8");
- }
-
- private static String generateQueryString(TreeMap treeMap) {
- StringBuilder canonicalizedQueryString = new StringBuilder();
- boolean first = true;
- for (String key : treeMap.navigableKeySet()) {
- String value = treeMap.get(key);
- if (!first) {
- canonicalizedQueryString.append("&");
- }
- first = false;
- canonicalizedQueryString.append(percentEncode(key)).append("=").append(percentEncode(value));
- }
- return canonicalizedQueryString.toString();
- }
-
- public static String generateSignString(String httpMethod, Map parameter) throws IOException {
- TreeMap sortParameter = new TreeMap();
- sortParameter.putAll(parameter);
- String canonicalizedQueryString = generateQueryString(sortParameter);
- if (null == httpMethod) {
- throw new RuntimeException("httpMethod can not be empty");
- }
- /** 构造待签名的字符串* */
- StringBuilder stringToSign = new StringBuilder();
- stringToSign.append(httpMethod).append("&");
- stringToSign.append(percentEncode("/")).append("&");
- stringToSign.append(percentEncode(canonicalizedQueryString));
- return stringToSign.toString();
- }
-
- public static String percentEncode(String value) {
- try {
- return value == null ? null : URLEncoder.encode(value, ENCODING)
- .replaceAll("\\+", "%20")
- .replaceAll("\\*", "%2A")
- .replaceAll("%7E", "~");
- } catch (Exception e) {
- }
- return "";
- }
}
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsLocalBinlogEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsLocalBinlogEventParser.java
index 05a1935e..aac14d99 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsLocalBinlogEventParser.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsLocalBinlogEventParser.java
@@ -28,15 +28,15 @@ import com.alibaba.otter.canal.protocol.position.LogPosition;
*/
public class RdsLocalBinlogEventParser extends LocalBinlogEventParser implements CanalEventParser, LocalBinLogConnection.FileParserListener {
- private String url = "https://rds.aliyuncs.com/"; // openapi地址
- private String accesskey; // 云账号的ak
- private String secretkey; // 云账号sk
- private String instanceId; // rds实例id
+ private String url; // openapi地址
+ private String accesskey; // 云账号的ak
+ private String secretkey; // 云账号sk
+ private String instanceId; // rds实例id
private Long startTime;
private Long endTime;
private BinlogDownloadQueue binlogDownloadQueue;
private ParseFinishListener finishListener;
- private int batchSize;
+ private int batchFileSize;
public RdsLocalBinlogEventParser(){
}
@@ -65,10 +65,9 @@ public class RdsLocalBinlogEventParser extends LocalBinlogEventParser implements
instanceId,
new Date(startTime),
new Date(endTime));
- binlogDownloadQueue = new BinlogDownloadQueue(binlogFiles, batchSize, directory);
+ binlogDownloadQueue = new BinlogDownloadQueue(binlogFiles, batchFileSize, directory);
binlogDownloadQueue.silenceDownload();
needWait = true;
- parallel = false;
// try to download one file,use to test server id
binlogDownloadQueue.tryOne();
} catch (Throwable e) {
@@ -91,10 +90,12 @@ public class RdsLocalBinlogEventParser extends LocalBinlogEventParser implements
binlogDownloadQueue.notifyNotMatch();
try {
binlogDownloadQueue.cleanDir();
+ binlogDownloadQueue.tryOne();
binlogDownloadQueue.prepare();
- } catch (Exception e) {
+ } catch (Throwable e) {
throw new RuntimeException(e);
}
+
try {
binlogDownloadQueue.execute(new Runnable() {
@@ -163,8 +164,10 @@ public class RdsLocalBinlogEventParser extends LocalBinlogEventParser implements
}
// 处理下logManager位点问题
LogPosition logPosition = logPositionManager.getLatestIndexBy(destination);
- EntryPosition position = logPosition.getPostion();
- if (position != null) {
+ Long timestamp = 0L;
+ if (logPosition != null && logPosition.getPostion() != null) {
+ timestamp = logPosition.getPostion().getTimestamp();
+ EntryPosition position = logPosition.getPostion();
LogPosition newLogPosition = new LogPosition();
String journalName = position.getJournalName();
int sepIdx = journalName.indexOf(".");
@@ -181,9 +184,13 @@ public class RdsLocalBinlogEventParser extends LocalBinlogEventParser implements
}
if (binlogDownloadQueue.isLastFile(fileName)) {
- logger.info("all file parse complete, switch to mysql parser!");
+ logger.warn("last file : " + fileName + " , timestamp : " + timestamp
+ + " , all file parse complete, switch to mysql parser!");
finishListener.onFinish();
return;
+ } else {
+ logger.warn("parse local binlog file : " + fileName + " , timestamp : " + timestamp
+ + " , try the next binlog !");
}
binlogDownloadQueue.prepare();
} catch (Exception e) {
@@ -207,7 +214,7 @@ public class RdsLocalBinlogEventParser extends LocalBinlogEventParser implements
void onFinish();
}
- public void setBatchSize(int batchSize) {
- this.batchSize = batchSize;
+ public void setBatchFileSize(int batchFileSize) {
+ this.batchFileSize = batchFileSize;
}
}
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/BinlogFile.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/BinlogFile.java
index e6b354d7..bee87521 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/BinlogFile.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/BinlogFile.java
@@ -6,7 +6,7 @@ package com.alibaba.otter.canal.parse.inbound.mysql.rds.data;
*/
public class BinlogFile {
- private Long FileSize;
+ private Long FileSize;
private String LogBeginTime;
private String LogEndTime;
private String DownloadLink;
@@ -69,4 +69,12 @@ public class BinlogFile {
public void setFileName(String fileName) {
this.fileName = fileName;
}
+
+ @Override
+ public String toString() {
+ return "BinlogFile [FileSize=" + FileSize + ", LogBeginTime=" + LogBeginTime + ", LogEndTime=" + LogEndTime
+ + ", DownloadLink=" + DownloadLink + ", HostInstanceID=" + HostInstanceID + ", LinkExpiredTime="
+ + LinkExpiredTime + ", fileName=" + fileName + "]";
+ }
+
}
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/DescribeBinlogFileResult.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/DescribeBinlogFileResult.java
index d8d54e44..d5b6820d 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/DescribeBinlogFileResult.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/DescribeBinlogFileResult.java
@@ -5,12 +5,13 @@ package com.alibaba.otter.canal.parse.inbound.mysql.rds.data;
* @since 1.0.25
*/
public class DescribeBinlogFileResult {
+
private RdsItem Items;
- private long PageNumber;
- private long TotalRecordCount;
- private long TotalFileSize;
- private String RequestId;
- private long PageRecordCount;
+ private long PageNumber;
+ private long TotalRecordCount;
+ private long TotalFileSize;
+ private String RequestId;
+ private long PageRecordCount;
public RdsItem getItems() {
return Items;
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/RdsBackupPolicy.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/RdsBackupPolicy.java
index b6ca7c53..e007671a 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/RdsBackupPolicy.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/RdsBackupPolicy.java
@@ -9,15 +9,16 @@ public class RdsBackupPolicy {
/**
* 数据备份保留天数（7到730天）。
*/
- private String BackupRetentionPeriod;
+ private String BackupRetentionPeriod;
/**
* 数据备份时间，格式：HH:mmZ- HH:mm Z。
*/
- private String PreferredBackupTime;
+ private String PreferredBackupTime;
/**
- * 数据备份周期。Monday：周一；Tuesday：周二；Wednesday：周三；Thursday：周四；Friday：周五；Saturday：周六；Sunday：周日。
+ * 数据备份周期。Monday：周一；Tuesday：周二；Wednesday：周三；Thursday：周四；Friday：周五；Saturday：
+ * 周六；Sunday：周日。
*/
- private String PreferredBackupPeriod;
+ private String PreferredBackupPeriod;
/**
* 日志备份状态。Enable：开启；Disabled：关闭。
*/
@@ -25,7 +26,7 @@ public class RdsBackupPolicy {
/**
* 日志备份保留天数（7到730天）。
*/
- private int LogBackupRetentionPeriod;
+ private int LogBackupRetentionPeriod;
public String getBackupRetentionPeriod() {
return BackupRetentionPeriod;
@@ -66,4 +67,12 @@ public class RdsBackupPolicy {
public void setLogBackupRetentionPeriod(int logBackupRetentionPeriod) {
LogBackupRetentionPeriod = logBackupRetentionPeriod;
}
+
+ @Override
+ public String toString() {
+ return "RdsBackupPolicy [BackupRetentionPeriod=" + BackupRetentionPeriod + ", PreferredBackupTime="
+ + PreferredBackupTime + ", PreferredBackupPeriod=" + PreferredBackupPeriod + ", BackupLog=" + BackupLog
+ + ", LogBackupRetentionPeriod=" + LogBackupRetentionPeriod + "]";
+ }
+
}
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/RdsItem.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/RdsItem.java
index a2cfaffe..81c484e3 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/RdsItem.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/RdsItem.java
@@ -7,6 +7,7 @@ import java.util.List;
* @since 1.0.25
*/
public class RdsItem {
+
private List BinLogFile;
public List getBinLogFile() {
@@ -16,4 +17,10 @@ public class RdsItem {
public void setBinLogFile(List binLogFile) {
BinLogFile = binLogFile;
}
+
+ @Override
+ public String toString() {
+ return "RdsItem [BinLogFile=" + BinLogFile + "]";
+ }
+
}
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/request/AbstractRequest.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/request/AbstractRequest.java
index 976968f6..f43523b5 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/request/AbstractRequest.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/request/AbstractRequest.java
@@ -1,12 +1,18 @@
package com.alibaba.otter.canal.parse.inbound.mysql.rds.request;
+import io.netty.handler.codec.http.HttpResponseStatus;
+
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.URLEncoder;
import java.security.cert.CertificateException;
import java.security.cert.X509Certificate;
import java.text.SimpleDateFormat;
-import java.util.*;
+import java.util.Date;
+import java.util.Map;
+import java.util.TimeZone;
+import java.util.TreeMap;
+import java.util.UUID;
import java.util.concurrent.TimeUnit;
import javax.crypto.Mac;
@@ -32,8 +38,6 @@ import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
import org.apache.http.ssl.SSLContexts;
import org.apache.http.util.EntityUtils;
-import io.netty.handler.codec.http.HttpResponseStatus;
-
/**
* @author chengjin.lyf on 2018/8/7 下午2:26
* @since 1.0.25
@@ -49,42 +53,38 @@ public abstract class AbstractRequest {
*/
private static final String MAC_NAME = "HmacSHA1";
- private String accessKeyId;
+ private String accessKeyId;
- private String accessKeySecret;
+ private String accessKeySecret;
/**
- * api 版本
- *
+ * api 版本
*/
- private String version;
+ private String version;
- private String endPoint = "rds.aliyuncs.com";
+ private String endPoint = "rds.aliyuncs.com";
- private String protocol = "http";
+ private String protocol = "http";
public void setProtocol(String protocol) {
this.protocol = protocol;
}
- private int timeout = (int) TimeUnit.MINUTES.toMillis(1);
-
+ private int timeout = (int) TimeUnit.MINUTES.toMillis(1);
private Map treeMap = new TreeMap();
- public void putQueryString(String name, String value){
- if (StringUtils.isBlank(name) || StringUtils.isBlank(value)){
+ public void putQueryString(String name, String value) {
+ if (StringUtils.isBlank(name) || StringUtils.isBlank(value)) {
return;
}
treeMap.put(name, value);
}
-
public void setVersion(String version) {
this.version = version;
}
-
public void setEndPoint(String endPoint) {
this.endPoint = endPoint;
}
@@ -169,7 +169,7 @@ public abstract class AbstractRequest {
p.put("Format", "JSON");
p.put("Version", version);
p.put("AccessKeyId", accessKeyId);
- p.put("SignatureMethod", "HMAC-SHA1"); //此处不能用变量 MAC_NAME
+ p.put("SignatureMethod", "HMAC-SHA1"); // 此处不能用变量 MAC_NAME
p.put("Timestamp", formatUTCTZ(new Date()));
p.put("SignatureVersion", "1.0");
p.put("SignatureNonce", UUID.randomUUID().toString());
@@ -193,6 +193,7 @@ public abstract class AbstractRequest {
* @return
* @throws IOException
*/
+ @SuppressWarnings("deprecation")
private final HttpResponse executeHttpRequest(HttpGet getMethod, String host) throws Exception {
SSLContext sslContext = SSLContexts.custom().loadTrustMaterial(null, new TrustStrategy() {
@@ -202,24 +203,24 @@ public abstract class AbstractRequest {
}
}).build();
SSLConnectionSocketFactory sslsf = new SSLConnectionSocketFactory(sslContext,
- new String[] { "TLSv1" },
- null,
- SSLConnectionSocketFactory.ALLOW_ALL_HOSTNAME_VERIFIER);
+ new String[] { "TLSv1" },
+ null,
+ SSLConnectionSocketFactory.ALLOW_ALL_HOSTNAME_VERIFIER);
Registry registry = RegistryBuilder.create()
- .register("http", PlainConnectionSocketFactory.INSTANCE)
- .register("https", sslsf)
- .build();
+ .register("http", PlainConnectionSocketFactory.INSTANCE)
+ .register("https", sslsf)
+ .build();
HttpClientConnectionManager httpClientConnectionManager = new PoolingHttpClientConnectionManager(registry);
CloseableHttpClient httpClient = HttpClientBuilder.create()
- .setMaxConnPerRoute(50)
- .setMaxConnTotal(100)
- .setConnectionManager(httpClientConnectionManager)
- .build();
+ .setMaxConnPerRoute(50)
+ .setMaxConnTotal(100)
+ .setConnectionManager(httpClientConnectionManager)
+ .build();
RequestConfig requestConfig = RequestConfig.custom()
- .setConnectTimeout(timeout)
- .setConnectionRequestTimeout(timeout)
- .setSocketTimeout(timeout)
- .build();
+ .setConnectTimeout(timeout)
+ .setConnectionRequestTimeout(timeout)
+ .setSocketTimeout(timeout)
+ .build();
getMethod.setConfig(requestConfig);
HttpResponse response = httpClient.execute(getMethod);
int statusCode = response.getStatusLine().getStatusCode();
@@ -232,14 +233,14 @@ public abstract class AbstractRequest {
protected abstract T processResult(HttpResponse response) throws Exception;
- protected void processBefore(){
+ protected void processBefore() {
}
- public final T doAction() throws Exception {
+ public final T doAction() throws Exception {
processBefore();
String requestStr = makeRequestString(treeMap);
- HttpGet httpGet = new HttpGet(protocol + "://" +endPoint + "?" + requestStr);
+ HttpGet httpGet = new HttpGet(protocol + "://" + endPoint + "?" + requestStr);
HttpResponse response = executeHttpRequest(httpGet, endPoint);
if (response.getStatusLine().getStatusCode() != HttpStatus.SC_OK) {
String result = EntityUtils.toString(response.getEntity());
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/request/DescribeBackupPolicyRequest.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/request/DescribeBackupPolicyRequest.java
index 0fbdfc47..cf51961a 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/request/DescribeBackupPolicyRequest.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/request/DescribeBackupPolicyRequest.java
@@ -1,27 +1,26 @@
package com.alibaba.otter.canal.parse.inbound.mysql.rds.request;
-import com.alibaba.otter.canal.parse.inbound.mysql.rds.data.RdsBackupPolicy;
import org.apache.http.HttpResponse;
import org.apache.http.util.EntityUtils;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
+import com.alibaba.otter.canal.parse.inbound.mysql.rds.data.RdsBackupPolicy;
/**
* rds 备份策略查询
+ *
* @author chengjin.lyf on 2018/8/7 下午3:41
* @since 1.0.25
*/
public class DescribeBackupPolicyRequest extends AbstractRequest {
-
- public DescribeBackupPolicyRequest() {
+ public DescribeBackupPolicyRequest(){
setVersion("2014-08-15");
putQueryString("Action", "DescribeBackupPolicy");
}
-
public void setRdsInstanceId(String rdsInstanceId) {
putQueryString("DBInstanceId", rdsInstanceId);
}
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/request/DescribeBinlogFilesRequest.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/request/DescribeBinlogFilesRequest.java
index 1c91c255..8aa90569 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/request/DescribeBinlogFilesRequest.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/request/DescribeBinlogFilesRequest.java
@@ -2,12 +2,12 @@ package com.alibaba.otter.canal.parse.inbound.mysql.rds.request;
import java.util.Date;
-import com.alibaba.otter.canal.parse.inbound.mysql.rds.data.DescribeBinlogFileResult;
import org.apache.http.HttpResponse;
import org.apache.http.util.EntityUtils;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
+import com.alibaba.otter.canal.parse.inbound.mysql.rds.data.DescribeBinlogFileResult;
/**
* @author chengjin.lyf on 2018/8/7 下午3:41
@@ -15,11 +15,9 @@ import com.alibaba.fastjson.TypeReference;
*/
public class DescribeBinlogFilesRequest extends AbstractRequest {
-
- public DescribeBinlogFilesRequest() {
+ public DescribeBinlogFilesRequest(){
setVersion("2014-08-15");
putQueryString("Action", "DescribeBinlogFiles");
-
}
public void setRdsInstanceId(String rdsInstanceId) {
@@ -35,11 +33,11 @@ public class DescribeBinlogFilesRequest extends AbstractRequest() {
- });
+ DescribeBinlogFileResult describeBinlogFileResult = JSONObject.parseObject(result,
+ new TypeReference() {
+ });
return describeBinlogFileResult;
}
}
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java
index 9a886336..904b55a9 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java
@@ -68,7 +68,8 @@ public class MemoryTableMeta implements TableMetaTSDB {
try {
// druid暂时flush privileges语法解析有问题
- if (!StringUtils.startsWithIgnoreCase(StringUtils.trim(ddl), "flush")) {
+ if (!StringUtils.startsWithIgnoreCase(StringUtils.trim(ddl), "flush")
+ && !StringUtils.startsWithIgnoreCase(StringUtils.trim(ddl), "grant")) {
repository.console(ddl);
}
} catch (Throwable e) {
diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/RdsBinlogEventParserProxyTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/RdsBinlogEventParserProxyTest.java
index 986d4d6a..546ee58c 100644
--- a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/RdsBinlogEventParserProxyTest.java
+++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/RdsBinlogEventParserProxyTest.java
@@ -8,6 +8,7 @@ import java.util.concurrent.atomic.AtomicLong;
import org.junit.Assert;
import org.junit.Test;
+import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter;
import com.alibaba.otter.canal.parse.helper.TimeoutChecker;
import com.alibaba.otter.canal.parse.inbound.mysql.rds.RdsBinlogEventParserProxy;
import com.alibaba.otter.canal.parse.index.AbstractLogPositionManager;
@@ -28,10 +29,9 @@ public class RdsBinlogEventParserProxyTest {
private static final String MYSQL_ADDRESS = "";
private static final String USERNAME = "";
private static final String PASSWORD = "";
- public static final String DBNAME = "";
- public static final String TBNAME = "";
- public static final String DDL = "";
-
+ public static final String DBNAME = "";
+ public static final String TBNAME = "";
+ public static final String DDL = "";
@Test
public void test_timestamp() throws InterruptedException {
@@ -41,7 +41,7 @@ public class RdsBinlogEventParserProxyTest {
final RdsBinlogEventParserProxy controller = new RdsBinlogEventParserProxy();
Calendar calendar = Calendar.getInstance();
- calendar.add(Calendar.DAY_OF_YEAR, -1);
+ calendar.add(Calendar.HOUR_OF_DAY, -24 * 4);
final EntryPosition defaultPosition = buildPosition(null, null, calendar.getTimeInMillis());
controller.setSlaveId(3344L);
controller.setDetectingEnable(false);
@@ -51,13 +51,15 @@ public class RdsBinlogEventParserProxyTest {
controller.setInstanceId("");
controller.setAccesskey("");
controller.setSecretkey("");
- controller.setBatchSize(4);
-// controller.setRdsOpenApiUrl("https://rds.aliyuncs.com/");
+ controller.setDirectory("/tmp/binlog");
+ controller.setEventBlackFilter(new AviaterRegexFilter("mysql\\.*"));
+ controller.setFilterTableError(true);
+ controller.setBatchFileSize(4);
controller.setEventSink(new AbstractCanalEventSinkTest>() {
@Override
public boolean sink(List entrys, InetSocketAddress remoteAddress, String destination)
- throws CanalSinkException {
+ throws CanalSinkException {
for (CanalEntry.Entry entry : entrys) {
if (entry.getEntryType() != CanalEntry.EntryType.HEARTBEAT) {
entryCount.incrementAndGet();
@@ -79,8 +81,8 @@ public class RdsBinlogEventParserProxyTest {
controller.setLogPositionManager(new AbstractLogPositionManager() {
private LogPosition logPosition;
+
public void persistLogPosition(String destination, LogPosition logPosition) {
- System.out.println(logPosition);
this.logPosition = logPosition;
}
@@ -105,7 +107,6 @@ public class RdsBinlogEventParserProxyTest {
Assert.assertTrue(entryPosition.getTimestamp() <= defaultPosition.getTimestamp());
}
-
// ======================== helper method =======================
private EntryPosition buildPosition(String binlogFile, Long offest, Long timestamp) {
diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/RdsBinlogOpenApiTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/RdsBinlogOpenApiTest.java
index 596847ae..25a14284 100644
--- a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/RdsBinlogOpenApiTest.java
+++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/RdsBinlogOpenApiTest.java
@@ -1,11 +1,14 @@
package com.alibaba.otter.canal.parse.inbound.mysql;
-import java.io.File;
import java.util.Date;
+import java.util.List;
import org.apache.commons.lang.time.DateUtils;
+import org.junit.Test;
import com.alibaba.otter.canal.parse.inbound.mysql.rds.RdsBinlogOpenApi;
+import com.alibaba.otter.canal.parse.inbound.mysql.rds.data.BinlogFile;
+import com.alibaba.otter.canal.parse.inbound.mysql.rds.data.RdsBackupPolicy;
/**
* @author agapple 2017年10月15日 下午2:14:34
@@ -13,15 +16,19 @@ import com.alibaba.otter.canal.parse.inbound.mysql.rds.RdsBinlogOpenApi;
*/
public class RdsBinlogOpenApiTest {
+ @Test
public void testSimple() throws Throwable {
- Date startTime = DateUtils.parseDate("2017-10-13 20:56:58", new String[] { "yyyy-MM-dd HH:mm:ss" });
- Date endTime = DateUtils.parseDate("2017-10-14 02:57:59", new String[] { "yyyy-MM-dd HH:mm:ss" });
- RdsBinlogOpenApi.downloadBinlogFiles("https://rds.aliyuncs.com/",
- "",
- "",
- "rm-bp180v4mfjnm157es",
- startTime,
- endTime,
- new File("/tmp/binlog/"));
+ Date startTime = DateUtils.parseDate("2018-08-10 12:00:00", new String[] { "yyyy-MM-dd HH:mm:ss" });
+ Date endTime = DateUtils.parseDate("2018-08-11 12:00:00", new String[] { "yyyy-MM-dd HH:mm:ss" });
+ String url = "https://rds.aliyuncs.com/";
+ String ak = "";
+ String sk = "";
+ String dbInstanceId = "";
+
+ RdsBackupPolicy backupPolicy = RdsBinlogOpenApi.queryBinlogBackupPolicy(url, ak, sk, dbInstanceId);
+ System.out.println(backupPolicy);
+
+ List binlogFiles = RdsBinlogOpenApi.listBinlogFiles(url, ak, sk, dbInstanceId, startTime, endTime);
+ System.out.println(binlogFiles);
}
}