chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

fixed format

Browse Source
This commit is contained in:
七锋
2018-06-14 19:25:34 +08:00
parent dca1b20043
commit a4255a6377
38 changed files with 11024 additions and 11392 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
client/src/main/java/com/alibaba/otter/canal/client/CanalConnector.java
View File

@@ -152,6 +152,7 @@ public interface CanalConnector {
/**
* 中断的阻塞，用于优雅停止client
*
* @throws CanalClientException
*/
void stopRunning() throws CanalClientException;

1
common/src/main/java/com/alibaba/otter/canal/common/zookeeper/running/ServerRunningMonitor.java
View File

@@ -4,7 +4,6 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import com.alibaba.otter.canal.common.CanalException;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.exception.ZkException;
import org.I0Itec.zkclient.exception.ZkInterruptedException;

4
deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java
View File

@@ -434,13 +434,13 @@ public class CanalController {
}
// 启动网络接口
if(canalServer!=null) {
if (canalServer != null) {
canalServer.start();
}
}
public void stop() throws Throwable {
if(canalServer!=null) {
if (canalServer != null) {
canalServer.stop();
}

1
deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java
View File

@@ -59,6 +59,7 @@ public class CanalLauncher {
private static void setGlobalUncaughtExceptionHandler() {
Thread.setDefaultUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
@Override
public void uncaughtException(Thread t, Throwable e) {
logger.error("UnCaughtException", e);

4
driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/GTIDSet.java
View File

@@ -3,8 +3,7 @@ package com.alibaba.otter.canal.parse.driver.mysql.packets;
import java.io.IOException;
/**
* Created by hiwjd on 2018/4/23.
* hiwjd0@gmail.com
* Created by hiwjd on 2018/4/23. hiwjd0@gmail.com
*/
public interface GTIDSet {
@@ -17,6 +16,7 @@ public interface GTIDSet {
/**
* 更新当前实例
*
* @param str
* @throws Exception
*/

83
driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/MysqlGTIDSet.java
View File

@@ -1,14 +1,14 @@
package com.alibaba.otter.canal.parse.driver.mysql.packets;
import com.alibaba.otter.canal.parse.driver.mysql.utils.ByteHelper;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.*;
import java.util.HashMap;
import java.util.Map;
import com.alibaba.otter.canal.parse.driver.mysql.utils.ByteHelper;
/**
* Created by hiwjd on 2018/4/23.
* hiwjd0@gmail.com
* Created by hiwjd on 2018/4/23. hiwjd0@gmail.com
*/
public class MysqlGTIDSet implements GTIDSet {
@@ -56,60 +56,25 @@ public class MysqlGTIDSet implements GTIDSet {
}
/**
* 解析如下格式的字符串为MysqlGTIDSet:
*
* 726757ad-4455-11e8-ae04-0242ac110002:1 =>
* MysqlGTIDSet{
* sets: {
* 726757ad-4455-11e8-ae04-0242ac110002: UUIDSet{
* SID: 726757ad-4455-11e8-ae04-0242ac110002,
* intervals: [{start:1, stop:2}]
* }
* }
* }
*
* 726757ad-4455-11e8-ae04-0242ac110002:1-3 =>
* MysqlGTIDSet{
* sets: {
* 726757ad-4455-11e8-ae04-0242ac110002: UUIDSet{
* SID: 726757ad-4455-11e8-ae04-0242ac110002,
* intervals: [{start:1, stop:4}]
* }
* }
* }
*
* 726757ad-4455-11e8-ae04-0242ac110002:1-3:4 =>
* MysqlGTIDSet{
* sets: {
* 726757ad-4455-11e8-ae04-0242ac110002: UUIDSet{
* SID: 726757ad-4455-11e8-ae04-0242ac110002,
* intervals: [{start:1, stop:5}]
* }
* }
* }
*
* 726757ad-4455-11e8-ae04-0242ac110002:1-3:7-9 =>
* MysqlGTIDSet{
* sets: {
* 726757ad-4455-11e8-ae04-0242ac110002: UUIDSet{
* SID: 726757ad-4455-11e8-ae04-0242ac110002,
* intervals: [{start:1, stop:4}, {start:7, stop: 10}]
* }
* }
* }
*
* 726757ad-4455-11e8-ae04-0242ac110002:1-3,726757ad-4455-11e8-ae04-0242ac110003:4 =>
* MysqlGTIDSet{
* sets: {
* 726757ad-4455-11e8-ae04-0242ac110002: UUIDSet{
* SID: 726757ad-4455-11e8-ae04-0242ac110002,
* intervals: [{start:1, stop:4}]
* },
* 726757ad-4455-11e8-ae04-0242ac110003: UUIDSet{
* SID: 726757ad-4455-11e8-ae04-0242ac110002,
* intervals: [{start:4, stop:5}]
* }
* }
* 解析如下格式的字符串为MysqlGTIDSet: 726757ad-4455-11e8-ae04-0242ac110002:1 =>
* MysqlGTIDSet{ sets: { 726757ad-4455-11e8-ae04-0242ac110002: UUIDSet{ SID:
* 726757ad-4455-11e8-ae04-0242ac110002, intervals: [{start:1, stop:2}] } }
* } 726757ad-4455-11e8-ae04-0242ac110002:1-3 => MysqlGTIDSet{ sets: {
* 726757ad-4455-11e8-ae04-0242ac110002: UUIDSet{ SID:
* 726757ad-4455-11e8-ae04-0242ac110002, intervals: [{start:1, stop:4}] } }
* } 726757ad-4455-11e8-ae04-0242ac110002:1-3:4 => MysqlGTIDSet{ sets: {
* 726757ad-4455-11e8-ae04-0242ac110002: UUIDSet{ SID:
* 726757ad-4455-11e8-ae04-0242ac110002, intervals: [{start:1, stop:5}] } }
* } 726757ad-4455-11e8-ae04-0242ac110002:1-3:7-9 => MysqlGTIDSet{ sets: {
* 726757ad-4455-11e8-ae04-0242ac110002: UUIDSet{ SID:
* 726757ad-4455-11e8-ae04-0242ac110002, intervals: [{start:1, stop:4},
* {start:7, stop: 10}] } } }
* 726757ad-4455-11e8-ae04-0242ac110002:1-3,726757
* ad-4455-11e8-ae04-0242ac110003:4 => MysqlGTIDSet{ sets: {
* 726757ad-4455-11e8-ae04-0242ac110002: UUIDSet{ SID:
* 726757ad-4455-11e8-ae04-0242ac110002, intervals: [{start:1, stop:4}] },
* 726757ad-4455-11e8-ae04-0242ac110003: UUIDSet{ SID:
* 726757ad-4455-11e8-ae04-0242ac110002, intervals: [{start:4, stop:5}] } }
* }
*
* @param gtidData

50
driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/UUIDSet.java
View File

@@ -1,7 +1,5 @@
package com.alibaba.otter.canal.parse.driver.mysql.packets;
import com.alibaba.otter.canal.parse.driver.mysql.utils.ByteHelper;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
@@ -10,9 +8,10 @@ import java.util.Collections;
import java.util.List;
import java.util.UUID;
import com.alibaba.otter.canal.parse.driver.mysql.utils.ByteHelper;
/**
* Created by hiwjd on 2018/4/23.
* hiwjd0@gmail.com
* Created by hiwjd on 2018/4/23. hiwjd0@gmail.com
*/
public class UUIDSet {
@@ -54,6 +53,7 @@ public class UUIDSet {
}
public static class Interval implements Comparable<Interval> {
public long start;
public long stop;
@@ -85,19 +85,15 @@ public class UUIDSet {
}
/**
* 解析如下格式字符串为UUIDSet:
*
* 726757ad-4455-11e8-ae04-0242ac110002:1 =>
* UUIDSet{SID: 726757ad-4455-11e8-ae04-0242ac110002, intervals: [{start:1, stop:2}]}
*
* 726757ad-4455-11e8-ae04-0242ac110002:1-3 =>
* UUIDSet{SID: 726757ad-4455-11e8-ae04-0242ac110002, intervals: [{start:1, stop:4}]}
*
* 726757ad-4455-11e8-ae04-0242ac110002:1-3:4
* UUIDSet{SID: 726757ad-4455-11e8-ae04-0242ac110002, intervals: [{start:1, stop:5}]}
*
* 726757ad-4455-11e8-ae04-0242ac110002:1-3:7-9
* UUIDSet{SID: 726757ad-4455-11e8-ae04-0242ac110002, intervals: [{start:1, stop:4}, {start:7, stop:10}]}
* 解析如下格式字符串为UUIDSet: 726757ad-4455-11e8-ae04-0242ac110002:1 => UUIDSet{SID:
* 726757ad-4455-11e8-ae04-0242ac110002, intervals: [{start:1, stop:2}]}
* 726757ad-4455-11e8-ae04-0242ac110002:1-3 => UUIDSet{SID:
* 726757ad-4455-11e8-ae04-0242ac110002, intervals: [{start:1, stop:4}]}
* 726757ad-4455-11e8-ae04-0242ac110002:1-3:4 UUIDSet{SID:
* 726757ad-4455-11e8-ae04-0242ac110002, intervals: [{start:1, stop:5}]}
* 726757ad-4455-11e8-ae04-0242ac110002:1-3:7-9 UUIDSet{SID:
* 726757ad-4455-11e8-ae04-0242ac110002, intervals: [{start:1, stop:4},
* {start:7, stop:10}]}
*
* @param str
* @return
@@ -110,7 +106,7 @@ public class UUIDSet {
}
List<Interval> intervals = new ArrayList<Interval>();
for (int i=1; i<ss.length; i++) {
for (int i = 1; i < ss.length; i++) {
intervals.add(parseInterval(ss[i]));
}
@@ -134,7 +130,7 @@ public class UUIDSet {
sb.append(":");
sb.append(interval.start);
sb.append("-");
sb.append(interval.stop-1);
sb.append(interval.stop - 1);
}
}
@@ -142,12 +138,8 @@ public class UUIDSet {
}
/**
* 解析如下格式字符串为Interval:
*
* 1 => Interval{start:1, stop:2}
* 1-3 => Interval{start:1, stop:4}
*
* 注意！字符串格式表达时[n,m]是两侧都包含的，Interval表达时[n,m)右侧开
* 解析如下格式字符串为Interval: 1 => Interval{start:1, stop:2} 1-3 =>
* Interval{start:1, stop:4} 注意！字符串格式表达时[n,m]是两侧都包含的，Interval表达时[n,m)右侧开
*
* @param str
* @return
@@ -173,8 +165,8 @@ public class UUIDSet {
}
/**
* 把{start,stop}连续的合并掉:
* [{start:1, stop:4},{start:4, stop:5}] => [{start:1, stop:5}]
* 把{start,stop}连续的合并掉: [{start:1, stop:4},{start:4, stop:5}] => [{start:1,
* stop:5}]
*
* @param intervals
* @return
@@ -183,11 +175,11 @@ public class UUIDSet {
List<Interval> combined = new ArrayList<Interval>();
Collections.sort(intervals);
int len = intervals.size();
for (int i=0; i<len; i++) {
for (int i = 0; i < len; i++) {
combined.add(intervals.get(i));
int j;
for (j=i+1; j<len; j++) {
for (j = i + 1; j < len; j++) {
if (intervals.get(i).stop >= intervals.get(j).start) {
intervals.get(i).stop = intervals.get(j).stop;
} else {

12
driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/client/BinlogDumpGTIDCommandPacket.java
View File

@@ -1,16 +1,14 @@
package com.alibaba.otter.canal.parse.driver.mysql.packets.client;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import com.alibaba.otter.canal.parse.driver.mysql.packets.CommandPacket;
import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet;
import com.alibaba.otter.canal.parse.driver.mysql.utils.ByteHelper;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
/**
* Created by hiwjd on 2018/4/24.
* hiwjd0@gmail.com
*
* Created by hiwjd on 2018/4/24. hiwjd0@gmail.com
* https://dev.mysql.com/doc/internals/en/com-binlog-dump-gtid.html
*/
public class BinlogDumpGTIDCommandPacket extends CommandPacket {
@@ -22,7 +20,7 @@ public class BinlogDumpGTIDCommandPacket extends CommandPacket {
public long slaveServerId;
public GTIDSet gtidSet;
public BinlogDumpGTIDCommandPacket() {
public BinlogDumpGTIDCommandPacket(){
setCommand((byte) 0x1e);
}

72
driver/src/test/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlGTIDSetTest.java
View File

@@ -1,17 +1,23 @@
package com.alibaba.otter.canal.parse.driver.mysql;
import static org.junit.Assert.assertEquals;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import org.junit.Test;
import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet;
import com.alibaba.otter.canal.parse.driver.mysql.packets.MysqlGTIDSet;
import com.alibaba.otter.canal.parse.driver.mysql.packets.UUIDSet;
import org.junit.Test;
import static org.junit.Assert.*;
import java.io.IOException;
import java.util.*;
/**
* Created by hiwjd on 2018/4/25.
* hiwjd0@gmail.com
* Created by hiwjd on 2018/4/25. hiwjd0@gmail.com
*/
public class MysqlGTIDSetTest {
@@ -20,14 +26,12 @@ public class MysqlGTIDSetTest {
GTIDSet gtidSet = MysqlGTIDSet.parse("726757ad-4455-11e8-ae04-0242ac110002:1");
byte[] bytes = gtidSet.encode();
byte[] expected = {
0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x72, 0x67, 0x57, (byte)0xad,
0x44, 0x55, 0x11, (byte)0xe8, (byte)0xae, 0x04, 0x02, 0x42, (byte)0xac, 0x11, 0x00, 0x02,
0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x02, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00
};
byte[] expected = { 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x72, 0x67, 0x57, (byte) 0xad, 0x44, 0x55,
0x11, (byte) 0xe8, (byte) 0xae, 0x04, 0x02, 0x42, (byte) 0xac, 0x11, 0x00, 0x02, 0x01, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00 };
for (int i=0; i<bytes.length; i++) {
for (int i = 0; i < bytes.length; i++) {
assertEquals(expected[i], bytes[i]);
}
}
@@ -35,31 +39,17 @@ public class MysqlGTIDSetTest {
@Test
public void testParse() {
Map<String, MysqlGTIDSet> cases = new HashMap<String, MysqlGTIDSet>(5);
cases.put(
"726757ad-4455-11e8-ae04-0242ac110002:1",
buildForTest(new Material("726757ad-4455-11e8-ae04-0242ac110002", 1, 2))
);
cases.put(
"726757ad-4455-11e8-ae04-0242ac110002:1-3",
buildForTest(new Material("726757ad-4455-11e8-ae04-0242ac110002", 1, 4))
);
cases.put(
"726757ad-4455-11e8-ae04-0242ac110002:1-3:4",
buildForTest(new Material("726757ad-4455-11e8-ae04-0242ac110002", 1, 5))
);
cases.put(
"726757ad-4455-11e8-ae04-0242ac110002:1-3:7-9",
buildForTest(new Material("726757ad-4455-11e8-ae04-0242ac110002", 1, 4, 7, 10))
);
cases.put(
"726757ad-4455-11e8-ae04-0242ac110002:1-3,726757ad-4455-11e8-ae04-0242ac110003:4",
buildForTest(
Arrays.asList(
new Material("726757ad-4455-11e8-ae04-0242ac110002", 1, 4),
new Material("726757ad-4455-11e8-ae04-0242ac110003", 4, 5)
)
)
);
cases.put("726757ad-4455-11e8-ae04-0242ac110002:1",
buildForTest(new Material("726757ad-4455-11e8-ae04-0242ac110002", 1, 2)));
cases.put("726757ad-4455-11e8-ae04-0242ac110002:1-3",
buildForTest(new Material("726757ad-4455-11e8-ae04-0242ac110002", 1, 4)));
cases.put("726757ad-4455-11e8-ae04-0242ac110002:1-3:4",
buildForTest(new Material("726757ad-4455-11e8-ae04-0242ac110002", 1, 5)));
cases.put("726757ad-4455-11e8-ae04-0242ac110002:1-3:7-9",
buildForTest(new Material("726757ad-4455-11e8-ae04-0242ac110002", 1, 4, 7, 10)));
cases.put("726757ad-4455-11e8-ae04-0242ac110002:1-3,726757ad-4455-11e8-ae04-0242ac110003:4",
buildForTest(Arrays.asList(new Material("726757ad-4455-11e8-ae04-0242ac110002", 1, 4),
new Material("726757ad-4455-11e8-ae04-0242ac110003", 4, 5))));
for (Map.Entry<String, MysqlGTIDSet> entry : cases.entrySet()) {
MysqlGTIDSet expected = entry.getValue();
@@ -71,7 +61,7 @@ public class MysqlGTIDSetTest {
private static class Material {
public Material(String uuid, long start, long stop) {
public Material(String uuid, long start, long stop){
this.uuid = uuid;
this.start = start;
this.stop = stop;
@@ -79,7 +69,7 @@ public class MysqlGTIDSetTest {
this.stop1 = 0;
}
public Material(String uuid, long start, long stop, long start1, long stop1) {
public Material(String uuid, long start, long stop, long start1, long stop1){
this.uuid = uuid;
this.start = start;
this.stop = stop;

10
driver/src/test/java/com/alibaba/otter/canal/parse/driver/mysql/UUIDSetTest.java
View File

@@ -1,16 +1,16 @@
package com.alibaba.otter.canal.parse.driver.mysql;
import com.alibaba.otter.canal.parse.driver.mysql.packets.UUIDSet;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
import java.util.HashMap;
import java.util.Map;
import static org.junit.Assert.*;
import org.junit.Test;
import com.alibaba.otter.canal.parse.driver.mysql.packets.UUIDSet;
/**
* Created by hiwjd on 2018/4/26.
* hiwjd0@gmail.com
* Created by hiwjd on 2018/4/26. hiwjd0@gmail.com
*/
public class UUIDSetTest {

1
example/src/main/java/com/alibaba/otter/canal/example/AbstractCanalClientTest.java
View File

@@ -13,7 +13,6 @@ import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.impl.ClusterCanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;

24
kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java
View File

@@ -1,14 +1,15 @@
package com.alibaba.otter.canal.kafka.client;
import com.alibaba.otter.canal.protocol.Message;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
import com.alibaba.otter.canal.protocol.Message;
/**
* canal kafka 数据操作客户端
@@ -24,10 +25,9 @@ public class KafkaCanalConnector {
private Integer partition;
private Properties properties;
public KafkaCanalConnector(String servers, String topic, Integer partition, String groupId) {
public KafkaCanalConnector(String servers, String topic, Integer partition, String groupId){
this.topic = topic;
this.partition = partition;
@@ -36,10 +36,11 @@ public class KafkaCanalConnector {
properties.put("group.id", groupId);
properties.put("enable.auto.commit", false);
properties.put("auto.commit.interval.ms", "1000");
properties.put("auto.offset.reset", "latest"); //earliest //如果没有offset则从最后的offset开始读
properties.put("request.timeout.ms", "40000"); //必须大于session.timeout.ms的设置
properties.put("session.timeout.ms", "30000"); //默认为30秒
properties.put("max.poll.records", "1"); //所以一次只取一条数据
properties.put("auto.offset.reset", "latest"); // earliest
// //如果没有offset则从最后的offset开始读
properties.put("request.timeout.ms", "40000"); // 必须大于session.timeout.ms的设置
properties.put("session.timeout.ms", "30000"); // 默认为30秒
properties.put("max.poll.records", "1"); // 所以一次只取一条数据
properties.put("key.deserializer", StringDeserializer.class.getName());
properties.put("value.deserializer", MessageDeserializer.class.getName());
}
@@ -110,8 +111,7 @@ public class KafkaCanalConnector {
* @return
*/
public Message getWithoutAck(Long timeout, TimeUnit unit) {
ConsumerRecords<String, Message> records =
kafkaConsumer.poll(unit.toMillis(timeout)); //基于配置，最多只能poll到一条数据
ConsumerRecords<String, Message> records = kafkaConsumer.poll(unit.toMillis(timeout)); // 基于配置，最多只能poll到一条数据
if (!records.isEmpty()) {
return records.iterator().next().value();

1
kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnectors.java
View File

@@ -7,6 +7,7 @@ package com.alibaba.otter.canal.kafka.client;
* @version 1.0.0
*/
public class KafkaCanalConnectors {
/**
* 创建kafka客户端链接
*

15
kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/MessageDeserializer.java
View File

@@ -1,15 +1,16 @@
package com.alibaba.otter.canal.kafka.client;
import java.util.Map;
import org.apache.kafka.common.serialization.Deserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.CanalPacket;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.exception.CanalClientException;
import com.google.protobuf.ByteString;
import org.apache.kafka.common.serialization.Deserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Map;
/**
* Kafka Message类的反序列化
@@ -18,6 +19,7 @@ import java.util.Map;
* @version 1.0.0
*/
public class MessageDeserializer implements Deserializer<Message> {
private static Logger logger = LoggerFactory.getLogger(MessageDeserializer.class);
@Override
@@ -27,8 +29,7 @@ public class MessageDeserializer implements Deserializer<Message> {
@Override
public Message deserialize(String topic, byte[] data) {
try {
if (data == null)
return null;
if (data == null) return null;
else {
CanalPacket.Packet p = CanalPacket.Packet.parseFrom(data);
switch (p.getType()) {

25
kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/CanalKafkaClientExample.java
View File

@@ -1,14 +1,15 @@
package com.alibaba.otter.canal.kafka.client.running;
import com.alibaba.otter.canal.kafka.client.KafkaCanalConnector;
import com.alibaba.otter.canal.kafka.client.KafkaCanalConnectors;
import com.alibaba.otter.canal.protocol.Message;
import java.util.concurrent.TimeUnit;
import org.apache.kafka.common.errors.WakeupException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;
import java.util.concurrent.TimeUnit;
import com.alibaba.otter.canal.kafka.client.KafkaCanalConnector;
import com.alibaba.otter.canal.kafka.client.KafkaCanalConnectors;
import com.alibaba.otter.canal.protocol.Message;
/**
* Kafka client example
@@ -17,6 +18,7 @@ import java.util.concurrent.TimeUnit;
* @version 1.0.0
*/
public class CanalKafkaClientExample {
protected final static Logger logger = LoggerFactory.getLogger(CanalKafkaClientExample.class);
private KafkaCanalConnector connector;
@@ -26,19 +28,22 @@ public class CanalKafkaClientExample {
private Thread thread = null;
private Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() {
public void uncaughtException(Thread t, Throwable e) {
logger.error("parse events has an error", e);
}
};
public CanalKafkaClientExample(String servers, String topic, Integer partition, String groupId) {
public CanalKafkaClientExample(String servers, String topic, Integer partition, String groupId){
connector = KafkaCanalConnectors.newKafkaConnector(servers, topic, partition, groupId);
}
public static void main(String[] args) {
try {
final CanalKafkaClientExample kafkaCanalClientExample = new CanalKafkaClientExample(AbstractKafkaTest.servers,
AbstractKafkaTest.topic, AbstractKafkaTest.partition, AbstractKafkaTest.groupId);
AbstractKafkaTest.topic,
AbstractKafkaTest.partition,
AbstractKafkaTest.groupId);
logger.info("## start the kafka consumer: {}-{}", AbstractKafkaTest.topic, AbstractKafkaTest.groupId);
kafkaCanalClientExample.start();
logger.info("## the canal kafka consumer is running now ......");
@@ -56,7 +61,8 @@ public class CanalKafkaClientExample {
}
});
while (running) ;
while (running)
;
} catch (Throwable e) {
logger.error("## Something goes wrong when starting up the kafka consumer:", e);
System.exit(0);
@@ -91,12 +97,13 @@ public class CanalKafkaClientExample {
}
private void process() {
while (!running) ;
while (!running)
;
try {
connector.subscribe();
while (running) {
try {
Message message = connector.getWithoutAck(1L, TimeUnit.SECONDS); //获取message
Message message = connector.getWithoutAck(1L, TimeUnit.SECONDS); // 获取message
if (message == null) {
continue;
}

17
kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaClientRunningTest.java
View File

@@ -1,16 +1,17 @@
package com.alibaba.otter.canal.kafka.client.running;
import com.alibaba.otter.canal.kafka.client.KafkaCanalConnector;
import com.alibaba.otter.canal.kafka.client.KafkaCanalConnectors;
import com.alibaba.otter.canal.protocol.Message;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.apache.kafka.common.errors.WakeupException;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import com.alibaba.otter.canal.kafka.client.KafkaCanalConnector;
import com.alibaba.otter.canal.kafka.client.KafkaCanalConnectors;
import com.alibaba.otter.canal.protocol.Message;
/**
* Kafka consumer获取Message的测试例子
@@ -19,6 +20,7 @@ import java.util.concurrent.TimeUnit;
* @version 1.0.0
*/
public class KafkaClientRunningTest extends AbstractKafkaTest {
private Logger logger = LoggerFactory.getLogger(KafkaClientRunningTest.class);
private boolean running = true;
@@ -30,6 +32,7 @@ public class KafkaClientRunningTest extends AbstractKafkaTest {
final KafkaCanalConnector connector = KafkaCanalConnectors.newKafkaConnector(servers, topic, partition, groupId);
executor.submit(new Runnable() {
@Override
public void run() {
connector.subscribe();
@@ -41,7 +44,7 @@ public class KafkaClientRunningTest extends AbstractKafkaTest {
}
connector.ack();
} catch (WakeupException e) {
//ignore
// ignore
}
}
connector.unsubscribe();

2
kafka/src/main/java/com/alibaba/otter/canal/kafka/CanalLauncher.java
View File

@@ -1,6 +1,5 @@
package com.alibaba.otter.canal.kafka;
import com.alibaba.otter.canal.kafka.producer.CanalKafkaStarter;
/**
@@ -13,7 +12,6 @@ public class CanalLauncher {
public static void main(String[] args) {
CanalServerStarter.init();
CanalKafkaStarter.init();
}
}

12
kafka/src/main/java/com/alibaba/otter/canal/kafka/CanalServerStarter.java
View File

@@ -1,12 +1,13 @@
package com.alibaba.otter.canal.kafka;
import com.alibaba.otter.canal.deployer.CanalController;
import java.io.FileInputStream;
import java.util.Properties;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.FileInputStream;
import java.util.Properties;
import com.alibaba.otter.canal.deployer.CanalController;
/**
* canal server 启动类
@@ -15,9 +16,9 @@ import java.util.Properties;
* @version 1.0.0
*/
public class CanalServerStarter {
private static final String CLASSPATH_URL_PREFIX = "classpath:";
private static final Logger logger = LoggerFactory.getLogger(CanalServerStarter.class);
private static boolean running = false;
public static void init() {
@@ -61,12 +62,13 @@ public class CanalServerStarter {
}
}
public static boolean isRunning(){
public static boolean isRunning() {
return running;
}
private static void setGlobalUncaughtExceptionHandler() {
Thread.setDefaultUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
@Override
public void uncaughtException(Thread t, Throwable e) {
logger.error("UnCaughtException", e);

14
kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java
View File

@@ -1,8 +1,7 @@
package com.alibaba.otter.canal.kafka.producer;
import com.alibaba.otter.canal.kafka.producer.KafkaProperties.Topic;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
@@ -10,7 +9,9 @@ import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Properties;
import com.alibaba.otter.canal.kafka.producer.KafkaProperties.Topic;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
/**
* kafka producer 主操作类
@@ -19,6 +20,7 @@ import java.util.Properties;
* @version 1.0.0
*/
public class CanalKafkaProducer {
private static final Logger logger = LoggerFactory.getLogger(CanalKafkaProducer.class);
private Producer<String, Message> producer;
@@ -51,7 +53,8 @@ public class CanalKafkaProducer {
boolean valid = false;
if (message != null && !message.getEntries().isEmpty()) {
for (CanalEntry.Entry entry : message.getEntries()) {
if (entry.getEntryType() != CanalEntry.EntryType.TRANSACTIONBEGIN && entry.getEntryType() != CanalEntry.EntryType.TRANSACTIONEND) {
if (entry.getEntryType() != CanalEntry.EntryType.TRANSACTIONBEGIN
&& entry.getEntryType() != CanalEntry.EntryType.TRANSACTIONEND) {
valid = true;
break;
}
@@ -61,7 +64,6 @@ public class CanalKafkaProducer {
return;
}
ProducerRecord<String, Message> record;
if (topic.getPartition() != null) {
record = new ProducerRecord<String, Message>(topic.getTopic(), topic.getPartition(), null, message);

42
kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java
View File

@@ -1,20 +1,21 @@
package com.alibaba.otter.canal.kafka.producer;
import java.io.FileInputStream;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.yaml.snakeyaml.Yaml;
import com.alibaba.otter.canal.kafka.CanalServerStarter;
import com.alibaba.otter.canal.kafka.producer.KafkaProperties.CanalDestination;
import com.alibaba.otter.canal.kafka.producer.KafkaProperties.Topic;
import com.alibaba.otter.canal.protocol.ClientIdentity;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.yaml.snakeyaml.Yaml;
import java.io.FileInputStream;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* kafka 启动类
@@ -23,6 +24,7 @@ import java.util.concurrent.Executors;
* @version 1.0.0
*/
public class CanalKafkaStarter {
private static final String CLASSPATH_URL_PREFIX = "classpath:";
private static final Logger logger = LoggerFactory.getLogger(CanalKafkaStarter.class);
@@ -36,23 +38,22 @@ public class CanalKafkaStarter {
public static void init() {
try {
logger.info("## load kafka configurations");
String conf = System.getProperty("kafka.conf", "classpath:kafka.yml");
if (conf.startsWith(CLASSPATH_URL_PREFIX)) {
conf = StringUtils.substringAfter(conf, CLASSPATH_URL_PREFIX);
kafkaProperties = new Yaml().loadAs(CanalKafkaStarter.class.getClassLoader().getResourceAsStream(conf), KafkaProperties.class);
kafkaProperties = new Yaml().loadAs(CanalKafkaStarter.class.getClassLoader().getResourceAsStream(conf),
KafkaProperties.class);
} else {
kafkaProperties = new Yaml().loadAs(new FileInputStream(conf), KafkaProperties.class);
}
//初始化 kafka producer
// 初始化 kafka producer
canalKafkaProducer = new CanalKafkaProducer();
canalKafkaProducer.init(kafkaProperties);
//对应每个instance启动一个worker线程
// 对应每个instance启动一个worker线程
List<CanalDestination> destinations = kafkaProperties.getCanalDestinations();
executorService = Executors.newFixedThreadPool(destinations.size());
@@ -60,6 +61,7 @@ public class CanalKafkaStarter {
logger.info("## start the kafka workers.");
for (final CanalDestination destination : destinations) {
executorService.execute(new Runnable() {
@Override
public void run() {
worker(destination);
@@ -69,6 +71,7 @@ public class CanalKafkaStarter {
running = true;
logger.info("## the kafka workers is running now ......");
Runtime.getRuntime().addShutdownHook(new Thread() {
public void run() {
try {
logger.info("## stop the kafka workers");
@@ -90,10 +93,11 @@ public class CanalKafkaStarter {
}
}
private static void worker(CanalDestination destination) {
while (!running) ;
while (!CanalServerStarter.isRunning()) ; //等待server启动完成
while (!running)
;
while (!CanalServerStarter.isRunning())
; // 等待server启动完成
logger.info("## start the canal consumer: {}.", destination.getCanalDestination());
CanalServerWithEmbedded server = CanalServerWithEmbedded.instance();
ClientIdentity clientIdentity = new ClientIdentity(destination.getCanalDestination(), (short) 1001, "");
@@ -103,7 +107,7 @@ public class CanalKafkaStarter {
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
//ignore
// ignore
}
continue;
}
@@ -123,7 +127,7 @@ public class CanalKafkaStarter {
destination.getTopics().add(topic);
}
for (Topic topic : destination.getTopics()) {
canalKafkaProducer.send(topic, message); //发送message到所有topic
canalKafkaProducer.send(topic, message); // 发送message到所有topic
}
}

3
kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/KafkaProperties.java
View File

@@ -12,6 +12,7 @@ import java.util.Set;
* @version 1.0.0
*/
public class KafkaProperties {
private String servers = "localhost:6667";
private int retries = 0;
private int batchSize = 16384;
@@ -23,6 +24,7 @@ public class KafkaProperties {
private List<CanalDestination> canalDestinations = new ArrayList<CanalDestination>();
public static class CanalDestination {
private String canalDestination;
private String topic;
private Integer partition;
@@ -62,6 +64,7 @@ public class KafkaProperties {
}
public static class Topic {
private String topic;
private Integer partition;

13
kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/MessageSerializer.java
View File

@@ -1,13 +1,14 @@
package com.alibaba.otter.canal.kafka.producer;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.CanalPacket;
import com.alibaba.otter.canal.protocol.Message;
import java.util.Map;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Serializer;
import org.springframework.util.CollectionUtils;
import java.util.Map;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.CanalPacket;
import com.alibaba.otter.canal.protocol.Message;
/**
* Kafka Message类的序列化
@@ -16,6 +17,7 @@ import java.util.Map;
* @version 1.0.0
*/
public class MessageSerializer implements Serializer<Message> {
@Override
public void configure(Map<String, ?> configs, boolean isKey) {
}
@@ -23,8 +25,7 @@ public class MessageSerializer implements Serializer<Message> {
@Override
public byte[] serialize(String topic, Message data) {
try {
if (data == null)
return null;
if (data == null) return null;
else {
CanalPacket.Messages.Builder messageBuilder = CanalPacket.Messages.newBuilder();
if (data.getId() != -1 && !CollectionUtils.isEmpty(data.getEntries())) {

3
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java
View File

@@ -221,7 +221,8 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
if (isGTIDMode()) {
erosaConnection.dump(MysqlGTIDSet.parse(startPosition.getGtid()), sinkHandler);
} else {
if (StringUtils.isEmpty(startPosition.getJournalName()) && startPosition.getTimestamp() != null) {
if (StringUtils.isEmpty(startPosition.getJournalName())
&& startPosition.getTimestamp() != null) {
erosaConnection.dump(startPosition.getTimestamp(), sinkHandler);
} else {
erosaConnection.dump(startPosition.getJournalName(),

4
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/ErosaConnection.java
View File

@@ -1,9 +1,9 @@
package com.alibaba.otter.canal.parse.inbound;
import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet;
import java.io.IOException;
import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet;
/**
* 通用的Erosa的链接接口, 用于一般化处理mysql/oracle的解析过程
*

2
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinLogConnection.java
View File

@@ -4,12 +4,12 @@ import java.io.File;
import java.io.IOException;
import java.util.List;
import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet;
import org.apache.commons.lang.NotImplementedException;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet;
import com.alibaba.otter.canal.parse.inbound.ErosaConnection;
import com.alibaba.otter.canal.parse.inbound.SinkFunction;
import com.alibaba.otter.canal.parse.inbound.mysql.local.BinLogFileQueue;

4
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java
View File

@@ -8,8 +8,6 @@ import java.nio.charset.Charset;
import java.util.List;
import java.util.concurrent.TimeUnit;
import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet;
import com.alibaba.otter.canal.parse.driver.mysql.packets.client.BinlogDumpGTIDCommandPacket;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -17,8 +15,10 @@ import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.parse.driver.mysql.MysqlConnector;
import com.alibaba.otter.canal.parse.driver.mysql.MysqlQueryExecutor;
import com.alibaba.otter.canal.parse.driver.mysql.MysqlUpdateExecutor;
import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet;
import com.alibaba.otter.canal.parse.driver.mysql.packets.HeaderPacket;
import com.alibaba.otter.canal.parse.driver.mysql.packets.client.BinlogDumpCommandPacket;
import com.alibaba.otter.canal.parse.driver.mysql.packets.client.BinlogDumpGTIDCommandPacket;
import com.alibaba.otter.canal.parse.driver.mysql.packets.client.RegisterSlaveCommandPacket;
import com.alibaba.otter.canal.parse.driver.mysql.packets.client.SemiAckCommandPacket;
import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ErrorPacket;

1
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaHistoryDO.java
View File

@@ -87,7 +87,6 @@ public class MetaHistoryDO {
this.useSchema = useSchema;
}
public String getExtra() {
return extra;
}

2
pom.xml
View File

@@ -256,7 +256,7 @@
<dependency>
<groupId>com.alibaba.fastsql</groupId>
<artifactId>fastsql</artifactId>
<version>2.0.0_preview_186</version>
<version>2.0.0_preview_371</version>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>

3134
protocol/src/main/java/com/alibaba/otter/canal/protocol/CanalEntry.java
View File

File diff suppressed because it is too large Load Diff

1
protocol/src/main/java/com/alibaba/otter/canal/protocol/position/EntryPosition.java
View File

@@ -1,6 +1,5 @@
package com.alibaba.otter.canal.protocol.position;
/**
* 数据库对象的唯一标示
*

6
server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java
View File

@@ -221,7 +221,8 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
if (CollectionUtils.isEmpty(events.getEvents())) {
logger.debug("get successfully, clientId:{} batchSize:{} but result is null",
clientIdentity.getClientId(), batchSize);
clientIdentity.getClientId(),
batchSize);
return new Message(-1, new ArrayList<Entry>()); // 返回空包，避免生成batchId，浪费性能
} else {
// 记录到流式信息
@@ -299,7 +300,8 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
if (CollectionUtils.isEmpty(events.getEvents())) {
logger.debug("getWithoutAck successfully, clientId:{} batchSize:{} but result is null",
clientIdentity.getClientId(), batchSize);
clientIdentity.getClientId(),
batchSize);
return new Message(-1, new ArrayList<Entry>()); // 返回空包，避免生成batchId，浪费性能
} else {
// 记录到流式信息

8
sink/src/main/java/com/alibaba/otter/canal/sink/entry/group/TimelineTransactionBarrier.java
View File

@@ -62,9 +62,9 @@ public class TimelineTransactionBarrier extends TimelineBarrier {
public void clear(Event event) {
super.clear(event);
//应该先判断2，再判断是否是事务尾，因为事务尾也可以导致txState的状态为2
//如果先判断事务尾，那么2的状态可能永远没机会被修改了，系统出现死锁
//CanalSinkException被注释的代码是不是可以放开？？我们内部使用的时候已经放开了，从代码逻辑的分析上以及实践效果来看，应该抛异常
// 应该先判断2，再判断是否是事务尾，因为事务尾也可以导致txState的状态为2
// 如果先判断事务尾，那么2的状态可能永远没机会被修改了，系统出现死锁
// CanalSinkException被注释的代码是不是可以放开？？我们内部使用的时候已经放开了，从代码逻辑的分析上以及实践效果来看，应该抛异常
if (txState.intValue() == 2) {// 非事务中
boolean result = txState.compareAndSet(2, 0);
if (result == false) {
@@ -92,7 +92,7 @@ public class TimelineTransactionBarrier extends TimelineBarrier {
return true; // 事务允许通过
}
} else if (txState.compareAndSet(0, 2)) { // 非事务保护中
//当基于zk-cursor启动的时候，拿到的第一个Event是TransactionEnd
// 当基于zk-cursor启动的时候，拿到的第一个Event是TransactionEnd
return true; // DDL/DCL/TransactionEnd允许通过
}
}