[2.5]fix ConcurrentModificationException (#2560)

* fix ConcurrentModificationException

* kafka consumer improvements

* kafka consumer improvements

* refactored kafka consumer

* refactored kafka consumer

common/queue/src/main/java/org/thingsboard/server/queue/kafka/TBKafkaConsumerTemplate.java

@@ -33,6 +33,8 @@ import java.util.Collections;
 import java.util.List;
 import java.util.Properties;
 import java.util.Set;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.stream.Collectors;
 
 /**
@@ -46,6 +48,7 @@ public class TBKafkaConsumerTemplate<T extends TbQueueMsg> implements TbQueueCon
     private final TbKafkaDecoder<T> decoder;
     private volatile boolean subscribed;
     private volatile Set<TopicPartitionInfo> partitions;
+    private final Lock consumerLock;
 
     @Getter
     private final String topic;
@@ -71,6 +74,7 @@ public class TBKafkaConsumerTemplate<T extends TbQueueMsg> implements TbQueueCon
         this.consumer = new KafkaConsumer<>(props);
         this.decoder = decoder;
         this.topic = topic;
+        this.consumerLock = new ReentrantLock();
     }
 
     @Override
@@ -94,12 +98,16 @@ public class TBKafkaConsumerTemplate<T extends TbQueueMsg> implements TbQueueCon
                 log.debug("Failed to await subscription", e);
             }
         } else {
+            try {
+                consumerLock.lock();
+
                 if (!subscribed) {
                     List<String> topicNames = partitions.stream().map(TopicPartitionInfo::getFullTopicName).collect(Collectors.toList());
                     topicNames.forEach(admin::createTopicIfNotExists);
                     consumer.subscribe(topicNames);
                     subscribed = true;
                 }
+
                 ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(durationInMillis));
                 if (records.count() > 0) {
                     List<T> result = new ArrayList<>();
@@ -112,18 +120,34 @@ public class TBKafkaConsumerTemplate<T extends TbQueueMsg> implements TbQueueCon
                     });
                     return result;
                 }
+            } finally {
+                consumerLock.unlock();
+            }
         }
         return Collections.emptyList();
     }
 
     @Override
     public void commit() {
+        try {
+            consumerLock.lock();
             consumer.commitAsync();
+        } finally {
+            consumerLock.unlock();
+        }
     }
 
     @Override
     public void unsubscribe() {
+        try {
+            consumerLock.lock();
+            if (consumer != null) {
                 consumer.unsubscribe();
+                consumer.close();
+            }
+        } finally {
+            consumerLock.unlock();
+        }
     }
 
     public T decode(ConsumerRecord<String, byte[]> record) throws IOException {