kafka producer - call callback.onFailure in case of producer.send exception (like InterruptedException when buffer.memory and max.block.ms reached)

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

@@ -77,25 +77,34 @@ public class TbKafkaProducerTemplate<T extends TbQueueMsg> implements TbQueuePro
 
     @Override
     public void send(TopicPartitionInfo tpi, T msg, TbQueueCallback callback) {
-        createTopicIfNotExist(tpi);
-        String key = msg.getKey().toString();
-        byte[] data = msg.getData();
-        ProducerRecord<String, byte[]> record;
-        Iterable<Header> headers = msg.getHeaders().getData().entrySet().stream().map(e -> new RecordHeader(e.getKey(), e.getValue())).collect(Collectors.toList());
-        record = new ProducerRecord<>(tpi.getFullTopicName(), null, key, data, headers);
-        producer.send(record, (metadata, exception) -> {
-            if (exception == null) {
-                if (callback != null) {
-                    callback.onSuccess(new KafkaTbQueueMsgMetadata(metadata));
-                }
-            } else {
-                if (callback != null) {
-                    callback.onFailure(exception);
+        try {
+            createTopicIfNotExist(tpi);
+            String key = msg.getKey().toString();
+            byte[] data = msg.getData();
+            ProducerRecord<String, byte[]> record;
+            Iterable<Header> headers = msg.getHeaders().getData().entrySet().stream().map(e -> new RecordHeader(e.getKey(), e.getValue())).collect(Collectors.toList());
+            record = new ProducerRecord<>(tpi.getFullTopicName(), null, key, data, headers);
+            producer.send(record, (metadata, exception) -> {
+                if (exception == null) {
+                    if (callback != null) {
+                        callback.onSuccess(new KafkaTbQueueMsgMetadata(metadata));
+                    }
                 } else {
-                    log.warn("Producer template failure: {}", exception.getMessage(), exception);
+                    if (callback != null) {
+                        callback.onFailure(exception);
+                    } else {
+                        log.warn("Producer template failure: {}", exception.getMessage(), exception);
+                    }
                 }
+            });
+        } catch (Exception e) {
+            if (callback != null) {
+                callback.onFailure(e);
+            } else {
+                log.warn("Producer template failure (send method wrapper): {}", e.getMessage(), e);
             }
-        });
+            throw e;
+        }
     }
 
     private void createTopicIfNotExist(TopicPartitionInfo tpi) {