bithaolee/canal-python

alibaba canal 客户端（Python3 版本）

canal-python's Issues

漏了protobuf 包

demo有个bug，

format_data['before']=format_data['after'] = dict() 改为
format_data['before'], format_data['after'] = dict(), dict()

我在查看修改数据库记录前后端的变化，发现修改前后字段的内容都一样，吓我一跳。呵呵。

check_valid账号密码是哪里设置的，有什么作用

def check_valid(self, username=b'', password=b''):

update有时候同步不到

测试10次update，有时候只能同步到1，2次，有时候能同步10次

version1.1.4 subscribe中的client_id 问题

不同的client_id， 不能做到分别消费对应的数据。
比如：client_id = b"1001" 消费了数据
client_id = b"1002"就无法继续消费数据了

demo 几个调整建议

1、 # 原代码连等式会造成before 和 after的值全部都是after的值，需分别创建字典。
format_data['before'] = dict()
format_data['after'] = dict()
'''
format_data['before'] = format_data['after'] = dict()
'''
2、 if event_type == EntryProtocol_pb2.EventType.DELETE:
for column in row.beforeColumns:
# 原代码只能记录记录的最后一个字段，需做一下更改，这样可以显示所有字段
format_data[column.name] = column.value
'''
format_data = {
column.name: column.value
}
'''
elif event_type == EntryProtocol_pb2.EventType.INSERT:
for column in row.afterColumns:
# 原代码只能记录记录的最后一个字段，需做一下更改，这样可以显示所有字段
format_data[column.name] = column.value
'''
format_data = {
column.name: column.value
}
'''

ack with error protobuf object, （ACK 的包写错了对象）

源代码：https://github.com/haozi3156666/canal-python/blob/68ecec0795ef712c0c00d001fccb39215cfbc3dc/canal/client.py#L116

        elif packet.type == CanalProtocol_pb2.PacketType.ACK:
            ack = CanalProtocol_pb2.PacketType.Ack()
            ack.MergeFromString(packet.body)
            if ack.error_code > 0:
                raise Exception('get data error. error code:%s, error message:%s' % (ack.error_code, ack.error_message))

CanalProtocol_pb2.PacketType.Ack()是一个枚举？即使是枚举也不能这么写吧。此处ACK回包，应该是返回 CanalProtocol_pb2.Ack()

        elif packet.type == CanalProtocol_pb2.PacketType.ACK:
            ack = CanalProtocol_pb2.Ack()
            ack.MergeFromString(packet.body)
            if ack.error_code > 0:
                raise Exception('get data error. error code:%s, error message:%s' % (ack.error_code, ack.error_message))

否则，运行有可能报错，至于为何不是必现的错误，我也不解，不知是否是bytes流的问题。

client demo 错误

/canal/client.py
id(format_data['before']) == id(format_data['after']) is true

should be : format_data['before'],format_data['after'] = dict(), dict()

canal python client HA

canal的standby节点启动canal python client时报错，主节点可以正常启动并消费。
Exception: Failed to subscribe. error code:400, error message:something goes wrong with channel:[id: 0x781f97b1, /192.168.44.129:52834 => /192.168.44.129:11111], exception=com.alibaba.otter.canal.server.exception.CanalServerException: destination:example should start first

canal 连接代码是不是没有处理update

{'db': 'pex', 'table': 'a', 'event_type': 2, 'data': {'before': {'id': '2', 'xa': '2'}, 'after': {'id': '2', 'xa': '2'}}}，
after & before 一模一样

canal-python示例代码问题反馈

https://github.com/bithaolee/canal-python 最后的代码示例 问题修改反馈

1. 只打印了最后一个字段 format_data每次都重新赋值而不是追加

format_data = dict()
if event_type == EntryProtocol_pb2.EventType.DELETE:
    for column in row.beforeColumns:
        format_data = {
            column.name: column.value
        }
elif event_type == EntryProtocol_pb2.EventType.INSERT:
    for column in row.afterColumns:
        format_data = {
            column.name: column.value
        }

==>

format_data = dict()
if event_type == EntryProtocol_pb2.EventType.DELETE:
    for column in row.beforeColumns:
        format_data[column.name] = column.value                    
elif event_type == EntryProtocol_pb2.EventType.INSERT:
    for column in row.afterColumns:
        format_data[column.name] = column.value

2. before和after内容一样

format_data['before'] = format_data['after'] = dict()
==>
format_data['before'] = dict()
format_data['after'] = dict()

按照示例 python3.7

这个库不支持连接Zookeeper

client.subscribe中filter写复杂的正则不起作用

client.subscribe(client_id=b'1001', destination=b'example', filter=b'.*\.(name1|name2)')
filter中的没起作用

连接失败

完整日志：

connected to 127.0.0.1:11111
Traceback (most recent call last):
  File "canal_client.py", line 10, in <module>
    client.check_valid(username=b'canal', password=b'canal')
  File "/usr/local/lib/python3.6/site-packages/canal/client.py", line 37, in check_valid
    data = self.connector.read_next_packet()
  File "/usr/local/lib/python3.6/site-packages/canal/connector.py", line 41, in read_next_packet
    data = self.read(self.packet_len)
  File "/usr/local/lib/python3.6/site-packages/canal/connector.py", line 29, in read
    raise Exception('TSocket: Could not read bytes from server')
Exception: TSocket: Could not read bytes from server

希望能针对canal1.4的新增的acl校验优化一下

canal1.4如果使用了canal-admin，即使在canal_local.properties 里不配置用户密码，客户端访问canal-server时也会要求校验，但是canal-python缺少这个功能，无法通过验证

你好，请问如果出现有消息没有同步到，没有消费那么下次怎么取到未消费的消息？

你好，看你的例子里，消息回执都没有写，写的也是可选。
请问如果出现有消息没有同步到，没有消费那么下次怎么取到未消费的消息？

attribute 'options' of 'google.protobuf.pyext._message.FileDescriptor' objects is not readable

这个报错有看过吗

不能区分BEGIN ,COMMIT

In [48]: EntryProtocol_pb2.EventType.items()                                                                                                                                                                                                                    
Out[48]: 
[('EVENTTYPECOMPATIBLEPROTO2', 0),
 ('INSERT', 1),
 ('UPDATE', 2),
 ('DELETE', 3),
 ('CREATE', 4),
 ('ALTER', 5),
 ('ERASE', 6),
 ('QUERY', 7),
 ('TRUNCATE', 8),
 ('RENAME', 9),
 ('CINDEX', 10),
 ('DINDEX', 11),
 ('GTID', 12),
 ('XACOMMIT', 13),
 ('XAROLLBACK', 14),
 ('MHEARTBEAT', 15)]

BEGIN和COMMIT的header.eventType都是0, 也就是EntryProtocol_pb2.EventType.Name(0)='EVENTTYPECOMPATIBLEPROTO2'
无法区分begin和commit.

还有一个问题是发现经常获取不到COMMIT, 除非COMMIT比事务中最后一个语句晚6秒执行

event_type为delete、insert时format_data不完全

第28行、第33行 row.beforeColumns、row.afterColumns中包含多个字段的修改
下面遍历语句中对format_data的操作是 format_data = {column.name: column.value} ，这样format_data仅仅会保留row.afterColumns中最后一个字段的信息
建议改成format_data.setdefault(column.name, column.value)

同步blob的字符串中文乱码

客户端中，不能同步处理好mysql源表的blog类型的字符串，会有乱码的发生。乱码只产生于 中文字符，英文字符没有问题。
canal版本为1.1.14
canal-python pip客户端为0.4版本

复原的步骤为：

create table test_blog ( id int not null primary key auto_increment, detail blob not null );

insert into test_blog (detail) values("测试");

此时同步这条数据就是乱码。

建立与Canal的连接代码bug

if event_type == EntryProtocol_pb2.EventType.DELETE:
for column in row.beforeColumns:
format_data = {
column.name: column.value
}
elif event_type == EntryProtocol_pb2.EventType.INSERT:
for column in row.afterColumns:
format_data = {
column.name: column.value
}

format_data = {column.name: column.value} 只会保留最后一个column的数据，
应该是format_data[column.name] = column.value

重启server端之后会重复消费

直接运行example.py，重启canal的server端之后，上次消费完ack之后的数据还会再次被消费

example.py中第41行存在BUG

format_data['before'] = format_data['after'] = dict()这句表示format_data['before']与format_data['after'] 指向的是同一个地址，修改format_data['before']时format_data['after'] 也会被修改，最终导致无法在mysql row模式下同时取到update前后的数据取值。
建议把第41行修改为：
format_data['before'] = dict()
format_data['after'] = dict()

something goes wrong when doing authentication. error code:400, error message:auth failed for user:

版本：canal 1.1.4
client.check_valid(username=b'', password=b'') 这里的username 和 password我不清楚应该填写哪个。

经过测试，如果canal server用户非local方式启动是可以正常运行，local方式启动的会有问题。

client.get()报错

Traceback (most recent call last):
File "/Users/admin/Desktop/project/dataCollection/canal.py", line 13, in
message = client.get()
File "/Users/admin/Desktop/project/dataCollection/canal/client.py", line 85, in get
message = self.get_without_ack(size)
File "/Users/admin/Desktop/project/dataCollection/canal/client.py", line 119, in get_without_ack
ack = CanalProtocol_pb2.PacketType.Ack()
File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/site-packages/google/protobuf/internal/enum_type_wrapper.py", line 95, in getattr
raise AttributeError
AttributeError