Changed m_is_subscribed state behavior

- Do not update m_is_subscribed variable from assign / unassign method
- Removed IsSubscribed call in Consumer::Consume & ConsumerConsumeLoop's
  execute method

Author: ArtemisMucaj

src/consumer.cc

@@ -153,8 +153,6 @@ Baton Consumer::Assign(std::vector<RdKafka::TopicPartition*> partitions) {
   m_partition_cnt = partitions.size();
   m_partitions.swap(partitions);
 
-  m_is_subscribed = true;
-
   return Baton(RdKafka::ERR_NO_ERROR);
 }
 
@@ -172,8 +170,6 @@ Baton Consumer::Unassign() {
     return Baton(errcode);
   }
 
-  m_is_subscribed = false;
-
   m_partitions.empty();
   m_partition_cnt = 0;
 
@@ -250,9 +246,9 @@ Baton Consumer::Subscribe(std::vector<std::string> topics) {
 NodeKafka::Message* Consumer::Consume() {
   NodeKafka::Message* m;
 
-  if (IsConnected() && IsSubscribed()) {
+  if (IsConnected()) {
     scoped_mutex_lock lock(m_connection_lock);
-    if (!IsConnected() && IsSubscribed()) {
+    if (!IsConnected()) {
       m = new NodeKafka::Message(RdKafka::ERR__STATE);
     } else {
       RdKafka::KafkaConsumer* consumer =

src/workers.cc

@@ -505,7 +505,7 @@ ConsumerConsumeLoop::~ConsumerConsumeLoop() {}
 
 void ConsumerConsumeLoop::Execute(const ExecutionMessageBus& bus) {
   // Do one check here before we move forward
-  while (consumer->IsConnected() && consumer->IsSubscribed()) {
+  while (consumer->IsConnected()) {
     NodeKafka::Message* message = consumer->Consume();
     if (message->errcode() == RdKafka::ERR__PARTITION_EOF) {
       delete message;