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

Browse Source
This commit is contained in:
mcy
2018-12-23 18:24:04 +08:00
parent 0469d28c05
commit b5697a366a
1 changed files with 9 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java
View File

@@ -30,16 +30,16 @@ import com.google.common.collect.Lists;
*/ */
public class KafkaCanalConnector implements CanalMQConnector { public class KafkaCanalConnector implements CanalMQConnector {
private KafkaConsumer<String, Message> kafkaConsumer; protected KafkaConsumer<String, Message> kafkaConsumer;
private KafkaConsumer<String, String> kafkaConsumer2; // 用于扁平message的数据消费 protected KafkaConsumer<String, String> kafkaConsumer2; // 用于扁平message的数据消费
private String topic; protected String topic;
private Integer partition; protected Integer partition;
private Properties properties; protected Properties properties;
private volatile boolean connected = false; protected volatile boolean connected = false;
private volatile boolean running = false; protected volatile boolean running = false;
private boolean flatMessage; protected boolean flatMessage;
private Map<Integer, Long> currentOffsets = new HashMap<>(); private Map<Integer, Long> currentOffsets = new HashMap<>();
public KafkaCanalConnector(String servers, String topic, Integer partition, String groupId, Integer batchSize, public KafkaCanalConnector(String servers, String topic, Integer partition, String groupId, Integer batchSize,
boolean flatMessage){ boolean flatMessage){