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

Merge pull request #1 from alibaba/master

Browse Source 
merge
This commit is contained in:
wangsaner
2019-03-28 10:37:24 +08:00
committed by GitHub
parent 739963acd0 97640b717b
commit d6ef66a281
2 changed files with 41 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
driver/src/test/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnectorTest.java
View File

@@ -3,26 +3,51 @@ package com.alibaba.otter.canal.parse.driver.mysql;
import java.io.IOException;
import java.net.InetSocketAddress;
import org.junit.Assert;
import org.junit.Test;
import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ResultSetPacket;
public class MysqlConnectorTest {
public static void main(String args[]) {
MysqlConnector connector = new MysqlConnector(new InetSocketAddress("yddb01.mysql.database.chinacloudapi.cn",
3306), "ps-admin01@yddb01", "1qaz3edc");
@Test
public void testQuery() {
MysqlConnector connector = new MysqlConnector(new InetSocketAddress("127.0.0.1", 3306), "xxxxx", "xxxxx");
try {
connector.connect();
MysqlQueryExecutor executor = new MysqlQueryExecutor(connector);
ResultSetPacket result = executor.query("show variables like '%char%';");
System.out.println(result);
result = executor.query("select * from test.test1");
System.out.println(result);
} catch (IOException e) {
e.printStackTrace();
Assert.fail(e.getMessage());
} finally {
try {
connector.disconnect();
} catch (IOException e) {
Assert.fail(e.getMessage());
}
}
}
// @Test
public void testUpdate() {
MysqlConnector connector = new MysqlConnector(new InetSocketAddress("127.0.0.1", 3306), "xxxxx", "xxxxx");
try {
connector.connect();
MysqlUpdateExecutor executor = new MysqlUpdateExecutor(connector);
executor.update("insert into test.test2(id,name,score,text_value) values(null,'中文1',10,'中文2')");
} catch (IOException e) {
Assert.fail(e.getMessage());
} finally {
try {
connector.disconnect();
} catch (IOException e) {
Assert.fail(e.getMessage());
}
}
}
}

24
server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java
View File

@@ -139,7 +139,7 @@ public class CanalKafkaProducer implements CanalMQProducer {
private void send(MQProperties.CanalDestination canalDestination, String topicName, Message message)
throws Exception {
if (!kafkaProperties.getFlatMessage()) {
List<ProducerRecord<String, Message>> records = new ArrayList<>();
List<ProducerRecord<String, Message>> records = new ArrayList<ProducerRecord<String, Message>>();
if (canalDestination.getPartitionHash() != null && !canalDestination.getPartitionHash().isEmpty()) {
Message[] messages = MQMessageUtils.messagePartition(message,
canalDestination.getPartitionsNum(),
@@ -148,23 +148,23 @@ public class CanalKafkaProducer implements CanalMQProducer {
for (int i = 0; i < length; i++) {
Message messagePartition = messages[i];
if (messagePartition != null) {
records.add(new ProducerRecord<>(topicName, i, null, messagePartition));
records.add(new ProducerRecord<String, Message>(topicName, i, null, messagePartition));
}
}
} else {
final int partition = canalDestination.getPartition() != null ? canalDestination.getPartition() : 0;
records.add(new ProducerRecord<>(topicName, partition, null, message));
records.add(new ProducerRecord<String, Message>(topicName, partition, null, message));
}
if (!records.isEmpty()) {
for (ProducerRecord<String, Message> record : records) {
producer.send(record).get();
}
if (logger.isDebugEnabled()) {
logger.debug("Send message to kafka topic: [{}], packet: {}", topicName, message.toString());
}
}
if (!records.isEmpty()) {
for (ProducerRecord<String, Message> record : records) {
producer.send(record).get();
}
if (logger.isDebugEnabled()) {
logger.debug("Send message to kafka topic: [{}], packet: {}", topicName, message.toString());
}
}
} else {
// 发送扁平数据json
List<FlatMessage> flatMessages = MQMessageUtils.messageConverter(message);