fix(transaction): failed to select transaction producer to call local state checker (#246)

* fix(transaction): failed to select transaction producer to call local state checker

* fix(transaction): fix the coredump when the transaction producer shutdown.

Author: ShannonDing

src/MQClientFactory.cpp

@@ -22,7 +22,7 @@
 #include "PullRequest.h"
 #include "Rebalance.h"
 #include "TopicPublishInfo.h"
-#include "TransactionMQProducer.h"
+#include "TransactionMQProducerImpl.h"
 
 #define MAX_BUFF_SIZE 8192
 #define SAFE_BUFF_SIZE 7936  // 8192 - 256 = 7936
@@ -698,7 +698,7 @@ void MQClientFactory::checkTransactionState(const std::string& addr,
   if (!group.empty()) {
     MQProducer* producer = selectProducer(group);
     if (producer != nullptr) {
-      TransactionMQProducer* transProducer = dynamic_cast<TransactionMQProducer*>(producer);
+      TransactionMQProducerImpl* transProducer = dynamic_cast<TransactionMQProducerImpl*>(producer);
       if (transProducer != nullptr) {
         transProducer->checkTransactionState(addr, messageExt, checkRequestHeader.m_tranStateTableOffset,
                                              checkRequestHeader.m_commitLogOffset, checkRequestHeader.m_msgId,

src/extern/CProducer.cpp

@@ -261,10 +261,6 @@ int DestroyProducer(CProducer* pProducer) {
       delete defaultMQProducer->innerTransactionProducer;
       defaultMQProducer->innerTransactionProducer = NULL;
     }
-    if (defaultMQProducer->listenerInner != NULL) {
-      delete defaultMQProducer->listenerInner;
-      defaultMQProducer->listenerInner = NULL;
-    }
   } else {
     if (defaultMQProducer->innerProducer != NULL) {
       delete defaultMQProducer->innerProducer;