From b6f1e0d13c4feb3e93a200688008dad4e5cf6b07 Mon Sep 17 00:00:00 2001
From: agapple <jianghang115@gmail.com>
Date: Tue, 20 Nov 2018 17:14:29 +0800
Subject: [PATCH] fixed rocketmq ack npe

---
 .../client/rocketmq/RocketMQCanalConnector.java      | 12 +++++++++---
 1 file changed, 9 insertions(+), 3 deletions(-)

diff --git a/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java b/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java
index da21ead8..2cf92a36 100644
--- a/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java
+++ b/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java
@@ -214,9 +214,13 @@ public class RocketMQCanalConnector implements CanalMQConnector {
     @Override
     public void ack() throws CanalClientException {
         try {
-            this.lastGetBatchMessage.ack();
+            if (this.lastGetBatchMessage != null) {
+                this.lastGetBatchMessage.ack();
+            }
         } catch (Throwable e) {
-            this.lastGetBatchMessage.fail();
+            if (this.lastGetBatchMessage != null) {
+                this.lastGetBatchMessage.fail();
+            }
         } finally {
             this.lastGetBatchMessage = null;
         }
@@ -225,7 +229,9 @@ public class RocketMQCanalConnector implements CanalMQConnector {
     @Override
     public void rollback() throws CanalClientException {
         try {
-            this.lastGetBatchMessage.fail();
+            if (this.lastGetBatchMessage != null) {
+                this.lastGetBatchMessage.fail();
+            }
         } finally {
             this.lastGetBatchMessage = null;
         }