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

测试调整

Browse Source
This commit is contained in:
mcy
2019-01-21 09:56:35 +08:00
parent 851637611f
commit 197ca67c8b
1 changed files with 5 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -44,12 +44,16 @@ public class CanalKafkaProducer implements CanalMQProducer {
properties.put("bootstrap.servers", kafkaProperties.getServers()); properties.put("bootstrap.servers", kafkaProperties.getServers());
properties.put("acks", kafkaProperties.getAcks()); properties.put("acks", kafkaProperties.getAcks());
properties.put("compression.type", kafkaProperties.getCompressionType()); properties.put("compression.type", kafkaProperties.getCompressionType());
properties.put("retries", kafkaProperties.getRetries());
properties.put("batch.size", kafkaProperties.getBatchSize()); properties.put("batch.size", kafkaProperties.getBatchSize());
properties.put("linger.ms", kafkaProperties.getLingerMs()); properties.put("linger.ms", kafkaProperties.getLingerMs());
properties.put("max.request.size", kafkaProperties.getMaxRequestSize()); properties.put("max.request.size", kafkaProperties.getMaxRequestSize());
properties.put("buffer.memory", kafkaProperties.getBufferMemory()); properties.put("buffer.memory", kafkaProperties.getBufferMemory());
properties.put("key.serializer", StringSerializer.class.getName()); properties.put("key.serializer", StringSerializer.class.getName());
if(kafkaProperties.getTransaction()){
properties.put("transactional.id", "canal-transactional-id");
} else {
properties.put("retries", kafkaProperties.getRetries());
}
if (!kafkaProperties.getFlatMessage()) { if (!kafkaProperties.getFlatMessage()) {
properties.put("value.serializer", MessageSerializer.class.getName()); properties.put("value.serializer", MessageSerializer.class.getName());
producer = new KafkaProducer<String, Message>(properties); producer = new KafkaProducer<String, Message>(properties);