binlogreader <==>kafkasink ,filter过滤无效，报Caused by: java.lang.IllegalArgumentException: user has not select privilege on [ "xxx." ]

{
"job": {
"content": [{
"reader": {
"name": "binlogreader",
"parameter": {
"cat": "insert,update,delete",
"filter": "canal\\.test.*",
"host": "192.168.1.211",
"jdbcUrl": "jdbc:mysql://192.168.1.211:3309/xxx?characterEncoding=UTF-8&useSSL=false",
"password": "root",
"pavingData": false,
"port": 3309,
"splitUpdate": false,
"table": [""],
"username": "root"
}
},
"writer": {
"name": "kafkasink",
"parameter": {
"producerSettings": {
"bootstrap.servers": "192.168.1.211:9092",
"auto.commit.enable": "false"
},
"tableFields": [],
"topic": "test-topica"
}
}
}],
"setting": {
"restore": {
"isRestore": true,
"isStream": true
},
"speed": {
"readerChannel": 1,
"writerChannel": 1
}
}
}
}