fix(kafka): close consumer before retrying on failures (#3018)

* fix(kafka): close consumer before retrying on failures

* fix test

(cherry picked from commit 71fe743)

connectors/kafka/src/main/java/io/camunda/connector/kafka/inbound/KafkaConnectorConsumer.java

@@ -43,6 +43,7 @@
 import org.slf4j.LoggerFactory;
 
 public class KafkaConnectorConsumer {
+
   private static final Logger LOG = LoggerFactory.getLogger(KafkaConnectorConsumer.class);
 
   private final InboundConnectorContext context;
@@ -102,6 +103,14 @@ public void startConsumer() {
             return null;
           } catch (Exception ex) {
             LOG.error("Consumer loop failure, retry pending: {}", ex.getMessage());
+            try {
+              consumer.close();
+            } catch (Exception e) {
+              LOG.error(
+                  "Failed to close consumer before retrying, reason: {}. "
+                      + "This error will be ignored. If the consumer is still running, it will be disconnected after max.poll.interval.ms.",
+                  e.getMessage());
+            }
             throw ex;
           }
         };

connectors/kafka/src/test/java/io/camunda/connector/kafka/inbound/KafkaExecutableTest.java

@@ -174,10 +174,10 @@ void testActivateAndDeactivate_consumerThrows() {
         .atMost(Duration.ofSeconds(5))
         .pollInterval(Duration.ofMillis(500))
         .untilAsserted(() -> assertFalse(kafkaExecutable.kafkaConnectorConsumer.shouldLoop));
-    kafkaExecutable.deactivate();
 
     // Then
     verify(mockConsumer, times(MAX_ATTEMPTS)).poll(any(Duration.class));
+    verify(mockConsumer, times(MAX_ATTEMPTS)).close();
   }
 
   @Test