withlin / canal-go

Alibaba mysql database binlog incremental subscription & consumer components Canal's golang client[阿里巴巴mysql数据库binlog的增量订阅&消费组件 Canal 的 go 客户端 ] https://github.com/alibaba/canal

canal-go

canal-go's Issues

canal-go能监听多个destination吗，能不能写个样例

canal-go能监听多个destination吗，能不能写个样例

建议代码中有错误不要直接panic

返回error，由调用方决定如何处理，Go触发panic和Java抛异常是不一样的。

Decode RawEntries模式时，raw没有赋值为true

根据java的客户端代码来理解lazyParseEntry 模式时需要设置raw为true。
java代码

                      if (lazyParseEntry) {
                            // byteString
                            result.setRawEntries(messages.getMessagesList());
                            result.setRaw(true);
                        }

参考：https://github.com/alibaba/canal/blob/master/client/src/main/java/com/alibaba/otter/canal/client/CanalMessageDeserializer.java#L35
目前canal-go代码

                if lazyParseEntry {
			message.RawEntries = messages.Messages
		}

https://github.com/withlin/canal-go/blob/master/protocol/Message.go#L62

获取不到详细的内容

获取不到具体的内容，在sample中直接打印fmt.Println(message.Entries)，获取到的信息是
[{version:1 logfileName:"mysql-bin.000002" logfileOffset:23611 serverId:1 serverenCode:"UTF-8" executeTime:1541055476000 sourceType:MYSQL eventLength:72 0xc42001243c [32 189 45] {} [] 0} {version:1 logfileName:"mysql-bin.000002" logfileOffset:24257 serverId:1 serverenCode:"UTF-8" executeTime:1541055476000 sourceType:MYSQL eventLength:27 0xc42001246c [18 6 51 57 52 56 55 51] {} [] 0}]

相同的canal instance，用java能获取到内容（ 客户端用的是https://github.com/liukelin/canal_mysql_nosql_sync/tree/master/canal-client ），在java连接成功后会有一个log
2018-11-01 15:06:39.566 [New I/O server worker #1-2] INFO c.a.otter.canal.instance.core.AbstractCanalInstance - subscribe filter change to ...

此时再断开java客户端，运行go客户端，就可以获取到信息，但是重启canal instance后又不行

canal-go如何进行数据库全量同步，adapter可以通过curl，这种client通过什么全量同步？

canal-go如何进行数据库全量同步，adapter可以通过curl，这种client通过什么全量同步？

无法获取详细变更内容

情况描述

在完成了简单配置后，运行sample中的main函数，无法获取到update、delete等操作等变化记录，只能得到如下信息
&{3 [{version:1 logfileName:"mysql-bin.000056" logfileOffset:73667 serverId:1682989439 serverenCode:"UTF-8" executeTime:1573480770000 sourceType:MYSQL eventLength:68 0xc000018830 [32 144 79] {} [] 0} {version:1 logfileName:"mysql-bin.000056" logfileOffset:73837 serverId:1682989439 serverenCode:"UTF-8" executeTime:1573480770000 sourceType:MYSQL eventLength:31 0xc000018860 [18 5 52 53 49 56 55] {} [] 0}] false }
TRANSACTIONBEGIN
TRANSACTIONEND

##配置文件如下

• conf/example/example.conf， 只做了如下修改：

canal.instance.master.address=xx.xx.xx.xx:3306

• conf/canal.properties， 只做了如下修改：

canal.instance.parser.parallel = false

数据库varbinary的字段中文会乱码

数据库varbinary的字段中文会乱码
尝试用ISO-8859-1解码后也乱码，请问下该如何解决？

无法监听canal数据

最新版本1.09无法监听到非cluster下canal的数据，simples下的main.go运行后没有数据。。。
canal-deployer 版本是1.1.4， MySQL是阿里云RDS，5.6

go: github.com/samuel/[email protected]: unknown revision c4fab1ac1bec

$ go mod vendor
go: finding github.com/samuel/go-zookeeper v0.0.0-20180130194729-c4fab1ac1bec
go: finding github.com/golang/protobuf v1.2.0
go: github.com/golang/[email protected]: unknown revision v1.2.0
go: github.com/samuel/[email protected]: unknown revision c4fab1ac1bec
go: error loading module requirements

有没有go module可以直接使用的方法，可以完善一下README吗

有没有go module可以直接使用的方法
可以完善一下README吗，我看到历史版本的README，不知道为什么现在删除了？能更新一下吗

ClusterCanalConnector 多个实例监听多个 Destination 只有一个能竞争到

你好，我们在用 canal-go 来消费多个 Destination 的 binlog 做数据同步，但是发现多个实例只有一个能够竞争到并且连接到 canal-server。

看代码实现 doConnect 的时候，zk 的 path 都是 /canal-consumer，如果启动了多个实例，要去监听不同的 Destination，看起来只有一个实例能够竞争到去 Connect。

修改了一下用 "/canal-consumer/" + destination 拼接成 zk 的 path 就可以了。不知道这里这样实现会不会有坑。

const 	(
	runningFlag = byte(0)
	notRunningFlag = byte(0)
	path = "/canal-consumer"
)

func NewClusterCanalConnector(canalNode *CanalClusterNode, username string, password string, destination string,
	soTimeOut int32, idleTimeOut int32) (*ClusterCanalConnector,error) {

	destinationPath := fmt.Sprintf("%s/%s", path, destination)

	err := checkRootPath(canalNode.zkClient, destinationPath)
	if err != nil {
		return nil, err
	}

	currentSequence, err := createEphemeralSequence(canalNode.zkClient, destinationPath)
	if err != nil {
		return nil,err
	}

	cluster := &ClusterCanalConnector{
		canalNode:   canalNode,
		username:    username,
		password:    password,
		destination: destination,
		soTimeOut:   soTimeOut,
		idleTimeOut: idleTimeOut,
		RetryTimes:  0,
		currentSequence:currentSequence,
		zkVersion:	 0,
		Path:		 destinationPath,
	}

	return cluster, nil
}

canal 1.1.5 连接启动失败

panic: runtime error: slice bounds out of range [:10004] with capacity 240 [recovered]
panic: runtime error: slice bounds out of range [:10004] with capacity 240

goroutine 19 [running]:
github.com/samuel/go-zookeeper/zk.decodePacket.func1(0xc00022de68)
D:/GoPath/pkg/mod/github.com/samuel/[email protected]/zk/structs.go:398 +0xfe
panic(0xc65aa0, 0xc000416000)
C:/Program Files/Go/src/runtime/panic.go:969 +0x1c7
github.com/samuel/go-zookeeper/zk.decodePacketValue(0xc000400010, 0x0, 0xf0, 0xc13760, 0xc00040a010, 0x197, 0x197, 0x0, 0x0)
D:/GoPath/pkg/mod/github.com/samuel/[email protected]/zk/structs.go:474 +0xa4a
github.com/samuel/go-zookeeper/zk.decodePacketValue(0xc000400000, 0x10, 0x100, 0xc08e60, 0xc00040a000, 0x16, 0x10, 0x100, 0xc00022de28)
D:/GoPath/pkg/mod/github.com/samuel/[email protected]/zk/structs.go:433 +0x471
github.com/samuel/go-zookeeper/zk.decodePacket(0xc000400000, 0x10, 0x100, 0xc08e60, 0xc00040a000, 0x0, 0x0, 0x0)
D:/GoPath/pkg/mod/github.com/samuel/[email protected]/zk/structs.go:407 +0x178
github.com/samuel/go-zookeeper/zk.(*Conn).authenticate(0xc000224420, 0x0, 0x0)
D:/GoPath/pkg/mod/github.com/samuel/[email protected]/zk/conn.go:726 +0x45b
github.com/samuel/go-zookeeper/zk.(*Conn).loop(0xc000224420)
D:/GoPath/pkg/mod/github.com/samuel/[email protected]/zk/conn.go:486 +0x65
github.com/samuel/go-zookeeper/zk.Connect.func1(0xc000224420)
D:/GoPath/pkg/mod/github.com/samuel/[email protected]/zk/conn.go:223 +0x36
created by github.com/samuel/go-zookeeper/zk.Connect
D:/GoPath/pkg/mod/github.com/samuel/[email protected]/zk/conn.go:222 +0x4b8

客户端不支持canal 1.1.2

对canal 1.1.0 解析正常，升级canal 1.1.2 后，获取不到数据，无报错
是哪里需要修改一下么？

消费数据中断

我按照Demo连接了测试的canal ，但发现数据成功消费了一部分后，就不再消费这是怎么回事（实际是有数据更新的）

整理文档

as title

golang

UnSubscribe方法根本就会真正取消订阅

好多地方的c.Running的判断完全无意义， 且真正需要运行的逻辑根本没执行到

关于数据吞吐的一点疑问，消费者client跟不上生产者binlog,会出现什么情况？

如果不接入消息队列的话，client消费跟不上数据库产生的速度，会产生什么影响吗？
比如连接了10个数据库，只有一个客户端。
数据库在不断的产生数据，客户端却没有消费，会出现什么情况？

多个canal-go连接同一个canal server后挂掉

使用的是 client.NewSimpleCanalConnector()，是否只能起一个客户端？
起多个相同的客户端后，全部客户段都挂了。
Panicing something goes wrong with reason:something goes wrong with channel:[id: 0x5609bb8b, /127.0.0.1:49897 =\u003e /127.0.0.1:11111], exception=com.alibaba.otter.canal.server.exception.CanalServerException: ack error , clientId:1001 batchId:18 is not exist , please check

Panicing something goes wrong with reason:something goes wrong with channel:[id: 0x6e1eb5f4, /127.0.0.1:50066 =\u003e /127.0.0.1:11111], exception=com.alibaba.otter.canal.meta.exception.CanalMetaManagerException: batchId:21 is not the firstly:20

我的逻辑是先获取了数据，处理后再应答，但多个客户端的情况下不按顺序应答会有问题，请问如何解决？

批量丢失数据

你好，这边在使用canal-go客户端的时候。发现在触发批量更新的是，客户端不能接收到所有的数据，debug发现，在p.body中数据是全的，但是通过proto.Unmarshal(p.Body, messages) 解析后，到messages就不正确了。

有中文乱码问题

canal-go链接canal后接收到message，进行解析，出现中文乱码。
请问如何解决？

release tag is imcompatible with official semantic versioning standard

Hi, @withlin thanks for your job. but right now the release tag is not compatible with semantic versioning standard. Would you update for that in next release ?

不知是不是BUG

场景：
1）所有组件均正常运行中。。
2）canal-go开始读取数据，读了一会，提示没有数据，然后很长一段时间，比如1分钟，仍然提示没有数据，但是当使用CTRL＋C终止客户端后，再重运行，突然会有一堆数据出来。

3）此情况，应该可以重现，不知是否BUG？

文件名多了一个点

原文件 /protocol/Position/log_identity..go，导致 go get 报错。应该是 /protocol/Position/log_identity.go。

canal-go 不支持1.1.4版本的Canal server！！！

我下載了最新版的Canal server， 版本号是1.1.4， 然后用php版的canal client连接做测试， 都能正常使用， 后来准备用Go版本的canal client， 一连接， 发现怎么改数据库， 都是提示:===没有数据了===, 这我就郁闷了， 再详细看你们的文档， 是用docker环境来run mysql和canal server. 我就尝试下， 发现docker环境用的canal server居然是1.1.0版本的， 然后发现居然能用了。。。go版本的client什么时候能支持 1.1.4 版本的 canal server???

客户端报错 something goes wrong with reason:something goes wrong with channel

错误信息

panic: something goes wrong with reason:something goes wrong with channel:[id: 0x7f903ed0, /172.25.56.90:52211 => /10.179.25.89:11111], exception=com.alibaba.otter.canal.server.exception.CanalServerException: ClientIdentity:ClientIdentity[destination=example,clientId=1001,filter=<null>] should subscribe first


goroutine 1 [running]:
github.com/CanalClient/canal-go/client.(*SimpleCanalConnector).receiveMessages(0xc0000f3f00, 0x2e)
	/Users/test/go/src/github.com/CanalClient/canal-go/client/SimpleCanalConnector.go:272 +0x838
github.com/CanalClient/canal-go/client.(*SimpleCanalConnector).GetWithOutAck(0xc0000f3f00, 0x3e8, 0x0, 0xc0000f3e14, 0x0)
	/Users/test/go/src/github.com/CanalClient/canal-go/client/SimpleCanalConnector.go:191 +0x2ec
github.com/CanalClient/canal-go/client.(*SimpleCanalConnector).Get(0xc0000f3f00, 0x3e8, 0x0, 0x0, 0x0)
	/Users/test/go/src/github.com/CanalClient/canal-go/client/SimpleCanalConnector.go:196 +0x47
main.main()
	/Users/test/go/src/server_cache/public/canal_client.go:21 +0x13a

Process finished with exit code 2

代码

func main()  {
	connector := client.NewSimpleCanalConnector("mc1.cn", 11111, "", "", "example", 60000, 60*60*1000)
	connector.Connect()

	fmt.Println("连接成功")
	for {
		message := connector.Get(1000, nil, nil)
		batchId := message.Id
		if batchId == -1 || len(message.Entries) <= 0 {
			time.Sleep(300 * time.Millisecond)
			fmt.Println("===没有数据了===")
			continue
		}

		printEntry(message.Entries)
	}
}

使用java客户端测试可以正常的获取binlog日志解析，环境是 centeros6.7+mysql5.7+go1.11

有计划支持集群吗？

并发开多个连接不同服务端报错（authentication:packet type=CLIENTAUTHENTICATION is NOT supported!）

time="2021-11-04T10:35:40+08:00" level=info msg="something goes wrong when doing authentication:packet type=CLIENTAUTHENTICATION is NOT supported!" src=ncc
time="2021-11-04T10:35:40+08:00" level=info msg="" src=default

文档写的堪忧

as title

protobuf命名冲突警告

你好，运行时会提示命名冲突

WARNING: proto: file "EntryProtocol.proto" has a name conflict over com.alibaba.otter.canal.protocol.HEARTBEAT
A future release will panic on registration conflicts. See:
https://developers.google.com/protocol-buffers/docs/reference/go/faq#namespace-conflict

增加mutex.Unlock

https://github.com/CanalClient/canal-go/blob/3bb26226c4ba38ce6f904d41b6b223e7fdcd4fa1/client/simple_canal_connector.go#L428

返回前需要 mutex.Unlock()

好像不能解析 create table 事件

好像不能解析 create table 事件

配置访问账号和密码后，总是返回EOF错误

在canal-server中配置了如下的访问控制方式

# canal instance user/passwd
canal.user = canal
canal.passwd = E3619321C1A937C46A0D8BD1DAC39F93B27D4458

使用canal-go的代码

connector := client.NewSimpleCanalConnector("192.168.199.17", 11111, "canal", "canal", "example", 60000, 60*60*1000)
err:=connector.Connect()

err总是为EOF错误

但在使用Java客户端时，能正常连接 https://github.com/alibaba/canal/tree/master/example

进程内不支持创建多个客户端

客户端操作的连接用全局变量，当需要在一个进程内创建多个实例连接时，会导致交互错乱

没法获取到变更的数据信息，只能获取到EntryType_TRANSACTIONBEGIN和EntryType_TRANSACTIONEND

1. go的结果
   
2. java的结果
   

debug后发现只能获取到事务开始和事务结束的entry

canal-server只改过ip之类的信息，使用的集群模式

mysql5.7.38配置文件需要调整一下才能抓取到变更

怎样只订阅一张表的curd

比如我只想知道 d_ec_user.t_user 表更新，应该怎么做？

客户端解码问题

cancal用的是canal/canal-server: v1.1.0镜像，客户端一直没有成功
遇到两个问题
第一个：在get的时候解码报错了，偶现
Fatal error: proto: Messages: illegal tag 0 (wire type 0)exit status 1

第二个：如果get成功了，就一定会报这个错误
panic: runtime error: invalid memory address or nil pointer dereference
[signal SIGSEGV: segmentation violation code=0x1 addr=0x0 pc=0x5222d2]

goroutine 1 [running]:
github.com/CanalClient/canal-go/protocol.(*RowChange).Reset(0x0)
/home/wenyuanhao/goworkspace/src/github.com/CanalClient/canal-go/protocol/EntryProtocol.pb.go:917 +0x22
github.com/CanalSharp/canal-go/vendor/github.com/golang/protobuf/proto.Unmarshal(0xc420016240, 0x17, 0x20, 0x6481c0, 0x0, 0x0, 0x68)
/home/wenyuanhao/goworkspace/src/github.com/CanalSharp/canal-go/vendor/github.com/golang/protobuf/proto/decode.go:336 +0x3b
main.printEntry(0xc42017c000, 0x1b0, 0x266)
/home/wenyuanhao/goworkspace/src/github.com/CanalSharp/canal-go/samples/main.go:41 +0x149
main.main()
/home/wenyuanhao/goworkspace/src/github.com/CanalSharp/canal-go/samples/main.go:29 +0x1d9
exit status 2

dial tcp 172.217.160.81:443: i/o timeout

`
go/src/github.com/withlin/canal-go/samples$ go build main.go

go: github.com/gogo/[email protected]: Get https://proxy.golang.org/github.com/gogo/protobuf/@v/v1.3.1.mod: dial tcp 172.217.160.81:443: i/o timeout
`

请问这个可以怎么解决

release版本

为何没有release？使用的时候无法锁定版本

mysql协议解析问题

代码位置：/client/simple_canal_connector.go

func readNextPacket() ([]byte, error) {
.....
        rdr := bufio.NewReader(conn)
	data := make([]byte, 0, 4*1024)
	n, err := io.ReadFull(rdr, data[:4])
	if err != nil {
		return nil, err
	}
	data = data[:n]
	dataLen := binary.BigEndian.Uint32(data)
	if uint64(dataLen) > uint64(cap(data)) {
		data = make([]byte, 0, dataLen)
	}
	n, err = io.ReadFull(rdr, data[:dataLen])
	if err != nil {
		return nil, err
	}
	data = data[:n]
	return data, nil
}

mysql的协议格式中，前四个字节位 |data_len| sequence_id |
但是代码里的处理是直接把前四个字节当作数据的长度，是有问题吧？还是我理解错了？

请问是不是没有实现KafkaCanalConnector?

我看Java那边有，这边好像没有。

使用nohup go run xx.go 2&>1 /dev/null 时，运行一段时间CPU飙升100% 返回的都是EOF

获取ps进程ID后kill掉，发现没用，后台使用top命令发现CPU占用100%的main函数，kill掉后才算是结束掉
测试过程中不使用nohup 没发现什么问题
求教

内部改成监听的形式，而不是轮训

as title

go get error

go: github.com/CanalSharp/[email protected]: parsing go.mod: unexpected module path "github.com/CanalClient/canal-go"

处理ack数据时为什么需要panic？

3a7ee23#diff-c4b62398375776a7a64942ec90c82994L347

	case PacketType_ACK:
		err := proto.Unmarshal(p.Body, ack)
		if err != nil {
			return nil, err
		}
		panic(errors.New(fmt.Sprintf("something goes wrong with reason:%s", ack.GetErrorMessage())))

代码如上，解析后如果err为空为什么需要panic呢？

doConnect() 为什么要//等待成为最小的节点，最小的节点去运行

这样导致我无法启动两个go-client订阅同一个zk

拆分SimpleCanalConnector.receiveMessages 函数

这个函数做了两件事情，
1 从canal获取数据。
2 解析获取到的数据。

将解析数据放入proto下面会更好一些。也有利于代码的复用

unsupported version at this client

pb版本冲突之后, 更新v1.1.1的包之后, 出现这种问题, 是不是 docker的image cannal-server 也要配套的 v.1.1.1 版本?

编码问题