修改kafka consumer关闭方式

kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java

@@ -4,11 +4,9 @@ import com.alibaba.otter.canal.protocol.Message;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.errors.WakeupException;
 import org.apache.kafka.common.serialization.StringDeserializer;
 
 import java.util.Collections;
-import java.util.ConcurrentModificationException;
 import java.util.Properties;
 import java.util.concurrent.TimeUnit;
 
@@ -59,35 +57,24 @@ public class KafkaCanalConnector {
     }
 
     /**
-     * 释放链接
+     * 关闭链接
      */
-    public void disconnect() {
+    public void close() {
-        if (kafkaConsumer != null) {
+        kafkaConsumer.close();
-            try {
-                kafkaConsumer.close();
-            } catch (ConcurrentModificationException e) {
-                kafkaConsumer.wakeup(); //通过wakeup异常间接关闭consumer
-            }
-        }
     }
 
     /**
      * 订阅topic
      */
     public void subscribe() {
-        try {
+        if (kafkaConsumer == null) {
-            if (kafkaConsumer == null) {
+            kafkaConsumer = new KafkaConsumer<String, Message>(properties);
-                kafkaConsumer = new KafkaConsumer<String, Message>(properties);
+        }
-            }
+        if (partition == null) {
-            if (partition == null) {
+            kafkaConsumer.subscribe(Collections.singletonList(topic));
-                kafkaConsumer.subscribe(Collections.singletonList(topic));
+        } else {
-            } else {
+            TopicPartition topicPartition = new TopicPartition(topic, partition);
-                kafkaConsumer.subscribe(Collections.singletonList(topic));
+            kafkaConsumer.assign(Collections.singletonList(topicPartition));
-                TopicPartition topicPartition = new TopicPartition(topic, partition);
-                kafkaConsumer.assign(Collections.singletonList(topicPartition));
-            }
-        } catch (WakeupException e) {
-            closeByWakeupException(e);
         }
     }
 
@@ -95,11 +82,7 @@ public class KafkaCanalConnector {
      * 取消订阅
      */
     public void unsubscribe() {
-        try {
+        kafkaConsumer.unsubscribe();
-            kafkaConsumer.unsubscribe();
-        } catch (WakeupException e) {
-            closeByWakeupException(e);
-        }
     }
 
     /**
@@ -127,15 +110,11 @@ public class KafkaCanalConnector {
      * @return
      */
     public Message getWithoutAck(Long timeout, TimeUnit unit) {
-        try {
+        ConsumerRecords<String, Message> records =
-            ConsumerRecords<String, Message> records =
+                kafkaConsumer.poll(unit.toMillis(timeout)); //基于配置，最多只能poll到一条数据
-                    kafkaConsumer.poll(unit.toMillis(timeout)); //基于配置，最多只能poll到一条数据
 
-            if (!records.isEmpty()) {
+        if (!records.isEmpty()) {
-                return records.iterator().next().value();
+            return records.iterator().next().value();
-            }
-        } catch (WakeupException e) {
-            closeByWakeupException(e);
         }
         return null;
     }
@@ -144,15 +123,6 @@ public class KafkaCanalConnector {
      * 提交offset，如果超过 session.timeout.ms 设置的时间没有ack则会抛出异常，ack失败
      */
     public void ack() {
-        try {
+        kafkaConsumer.commitSync();
-            kafkaConsumer.commitSync();
-        } catch (WakeupException e) {
-            closeByWakeupException(e);
-        }
-    }
-
-    private void closeByWakeupException(WakeupException e) {
-        kafkaConsumer.close();
-        throw e;
     }
 }

kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/CanalKafkaClientExample.java

@@ -80,7 +80,6 @@ public class CanalKafkaClientExample {
         if (!running) {
             return;
         }
-        connector.disconnect();
         running = false;
         if (thread != null) {
             try {
@@ -114,20 +113,19 @@ public class CanalKafkaClientExample {
                     }
 
                     connector.ack(); // 提交确认
-                } catch (WakeupException e) {
-                    try {
-                        Thread.sleep(500); //延时确保running状态的改变
-                    } catch (InterruptedException ie) {
-                        //ignore
-                    }
                 } catch (Exception e) {
                     logger.error(e.getMessage(), e);
                 }
             }
-        } catch (WakeupException e) {
-            //ignore
         } catch (Exception e) {
             logger.error(e.getMessage(), e);
         }
+
+        try {
+            connector.unsubscribe();
+        } catch (WakeupException e) {
+            // No-op. Continue process
+        }
+        connector.close();
     }
 }