diff --git a/kafka-client/pom.xml b/kafka-client/pom.xml
index 0efd13d4..264cfa0c 100644
--- a/kafka-client/pom.xml
+++ b/kafka-client/pom.xml
@@ -7,6 +7,7 @@
canal
com.alibaba.otter
1.0.26-SNAPSHOT
+ ../pom.xml
com.alibaba.otter
canal.kafka.client
@@ -17,7 +18,7 @@
com.alibaba.otter
- canal.client
+ canal.protocol
${project.version}
diff --git a/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java b/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java
index fc749fec..e3a68934 100644
--- a/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java
+++ b/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java
@@ -1,6 +1,5 @@
package com.alibaba.otter.canal.kafka.client;
-import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.exception.CanalClientException;
import org.apache.kafka.clients.consumer.ConsumerRecords;
@@ -20,7 +19,7 @@ import java.util.concurrent.TimeUnit;
* @author machengyuan @ 2018-6-12
* @version 1.0.0
*/
-public class KafkaCanalConnector implements CanalConnector {
+public class KafkaCanalConnector {
private KafkaConsumer kafkaConsumer;
@@ -39,23 +38,31 @@ public class KafkaCanalConnector implements CanalConnector {
properties.put("bootstrap.servers", servers);
properties.put("group.id", groupId);
properties.put("enable.auto.commit", false);
- properties.put("auto.commit.interval.ms", 1000);
+ properties.put("auto.commit.interval.ms", "1000");
properties.put("auto.offset.reset", "latest"); //earliest //如果没有offset则从最后的offset开始读
- properties.put("request.timeout.ms", 600000);
- properties.put("offsets.commit.timeout.ms", 300000);
- properties.put("session.timeout.ms", 30000);
- properties.put("max.poll.records", 1); //一次只取一条message
+ properties.put("request.timeout.ms", "360000"); //必须大于session.timeout.ms的设置
+ properties.put("session.timeout.ms", "300000"); //默认为5分钟
+ properties.put("max.poll.records", "1"); //所以一次只取一条数据
properties.put("key.deserializer", StringDeserializer.class.getName());
properties.put("value.deserializer", MessageDeserializer.class.getName());
}
- @Override
- public void connect() throws CanalClientException {
- kafkaConsumer = new KafkaConsumer(properties);
+ /**
+ * 重新设置sessionTime
+ *
+ * @param timeout
+ * @param unit
+ */
+ public void setSessionTimeout(Long timeout, TimeUnit unit) {
+ long t = unit.toMillis(timeout);
+ properties.put("request.timeout.ms", String.valueOf(t + 60000));
+ properties.put("session.timeout.ms", String.valueOf(t));
}
- @Override
- public void disconnect() throws CanalClientException {
+ /**
+ * 释放链接
+ */
+ public void disconnect() {
if (kafkaConsumer != null) {
try {
kafkaConsumer.close();
@@ -65,17 +72,17 @@ public class KafkaCanalConnector implements CanalConnector {
}
}
- @Override
- public boolean checkValid() throws CanalClientException {
- return true;
- }
-
- @Override
- public void subscribe(String filter) throws CanalClientException {
+ /**
+ * 订阅topic，不能重复订阅
+ */
+ public void subscribe() {
try {
- if (kafkaConsumer == null) {
- throw new CanalClientException("connect the kafka first before subscribe");
+ if (kafkaConsumer != null) {
+ throw new CanalClientException("resubscribe error");
}
+
+ kafkaConsumer = new KafkaConsumer(properties);
+
if (partition == null) {
kafkaConsumer.subscribe(Collections.singletonList(topic));
} else {
@@ -85,86 +92,69 @@ public class KafkaCanalConnector implements CanalConnector {
}
} catch (WakeupException e) {
closeByWakeupException(e);
- } catch (Exception e) {
- throw new CanalClientException(e);
}
}
- @Override
- public void subscribe() throws CanalClientException {
- subscribe(null);
- }
-
- @Override
- public void unsubscribe() throws CanalClientException {
+ /**
+ * 取消订阅
+ */
+ public void unsubscribe() {
try {
kafkaConsumer.unsubscribe();
} catch (WakeupException e) {
closeByWakeupException(e);
- } catch (Exception e) {
- throw new CanalClientException(e);
}
}
- @Override
- public Message get(int batchSize) throws CanalClientException {
- return get(batchSize, 100L, TimeUnit.MILLISECONDS);
+ /**
+ * 获取数据，自动进行确认
+ *
+ * @return
+ */
+ public Message get() {
+ return get(100L, TimeUnit.MILLISECONDS);
}
- @Override
- public Message get(int batchSize, Long timeout, TimeUnit unit) throws CanalClientException {
- Message message = getWithoutAck(batchSize, timeout, unit);
- this.ack(1);
+ public Message get(Long timeout, TimeUnit unit) {
+ Message message = getWithoutAck(timeout, unit);
+ this.ack();
return message;
}
- @Override
- public Message getWithoutAck(int batchSize) throws CanalClientException {
- return getWithoutAck(batchSize, 100L, TimeUnit.MILLISECONDS);
+ public Message getWithoutAck() {
+ return getWithoutAck(100L, TimeUnit.MILLISECONDS);
}
- @Override
- public Message getWithoutAck(int batchSize, Long timeout, TimeUnit unit) throws CanalClientException {
+ /**
+ * 获取数据，不进行确认，等待处理完成手工确认
+ *
+ * @return
+ */
+ public Message getWithoutAck(Long timeout, TimeUnit unit) {
try {
- ConsumerRecords records = kafkaConsumer.poll(unit.toMillis(timeout)); //基于配置，一次最多只能poll到一条Msg
+ ConsumerRecords records =
+ kafkaConsumer.poll(unit.toMillis(timeout)); //基于配置，最多只能poll到一条数据
if (!records.isEmpty()) {
return records.iterator().next().value();
}
} catch (WakeupException e) {
closeByWakeupException(e);
- } catch (Exception e) {
- throw new CanalClientException(e);
}
return null;
}
- @Override
- public void ack(long batchId) throws CanalClientException {
+ /**
+ * 提交offset，如果超过 session.timeout.ms 设置的时间没有ack则会抛出异常，ack失败
+ */
+ public void ack() {
try {
kafkaConsumer.commitSync();
} catch (WakeupException e) {
closeByWakeupException(e);
- } catch (Exception e) {
- throw new CanalClientException(e);
}
}
- @Override
- public void rollback(long batchId) throws CanalClientException {
-
- }
-
- @Override
- public void rollback() throws CanalClientException {
-
- }
-
- @Override
- public void stopRunning() throws CanalClientException {
-
- }
-
private void closeByWakeupException(WakeupException e) {
kafkaConsumer.close();
throw e;
diff --git a/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnectors.java b/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnectors.java
index f0d4dea6..eb5f248c 100644
--- a/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnectors.java
+++ b/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnectors.java
@@ -1,7 +1,5 @@
package com.alibaba.otter.canal.kafka.client;
-import com.alibaba.otter.canal.client.CanalConnector;
-
public class KafkaCanalConnectors {
/**
* 创建kafka客户端链接
@@ -12,7 +10,7 @@ public class KafkaCanalConnectors {
* @param groupId
* @return
*/
- public static CanalConnector newKafkaConnector(String servers, String topic, Integer partition, String groupId) {
+ public static KafkaCanalConnector newKafkaConnector(String servers, String topic, Integer partition, String groupId) {
return new KafkaCanalConnector(servers, topic, partition, groupId);
}
@@ -24,7 +22,7 @@ public class KafkaCanalConnectors {
* @param groupId
* @return
*/
- public static CanalConnector newKafkaConnector(String servers, String topic, String groupId) {
+ public static KafkaCanalConnector newKafkaConnector(String servers, String topic, String groupId) {
return new KafkaCanalConnector(servers, topic, null, groupId);
}
}
diff --git a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/ClientRunningTest.java b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/ClientRunningTest.java
index c968f0f3..8b1ffc1d 100644
--- a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/ClientRunningTest.java
+++ b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/ClientRunningTest.java
@@ -1,6 +1,6 @@
package com.alibaba.otter.canal.kafka.client.running;
-import com.alibaba.otter.canal.client.CanalConnector;
+import com.alibaba.otter.canal.kafka.client.KafkaCanalConnector;
import com.alibaba.otter.canal.kafka.client.KafkaCanalConnectors;
import com.alibaba.otter.canal.protocol.Message;
import org.junit.Test;
@@ -16,20 +16,19 @@ public class ClientRunningTest extends AbstractKafkaTest {
public void testKafkaConsumer() {
final ExecutorService executor = Executors.newFixedThreadPool(1);
- final CanalConnector kafkaCanalConnector = KafkaCanalConnectors.newKafkaConnector(servers, topic, partition, groupId);
- kafkaCanalConnector.connect();
+ final KafkaCanalConnector kafkaCanalConnector = KafkaCanalConnectors.newKafkaConnector(servers, topic, partition, groupId);
kafkaCanalConnector.subscribe();
executor.submit(new Runnable() {
@Override
public void run() {
while (running) {
- Message message = kafkaCanalConnector.getWithoutAck(1);
+ Message message = kafkaCanalConnector.getWithoutAck();
if (message != null) {
System.out.println(message);
sleep(40000);
}
- kafkaCanalConnector.ack(1);
+ kafkaCanalConnector.ack();
}
}
});
diff --git a/kafka/src/main/resources/kafka.yml b/kafka/src/main/resources/kafka.yml
index a869321c..3861fe58 100644
--- a/kafka/src/main/resources/kafka.yml
+++ b/kafka/src/main/resources/kafka.yml
@@ -5,7 +5,7 @@ lingerMs: 1
bufferMemory: 33554432
topics:
- - topic: expTest
+ - topic: example
partition:
canalDestination: example
# - topic: example2