The architecture of that software its done like this:

1. there is a queue that its filled by `io.aiven.connect.elasticsearch.ElasticsearchSinkTask.put`
2. there is a task that takes a chunk of messages at a time and attempt to write those in ElasticSearch `io.aiven.connect.elasticsearch.ElasticsearchWriter.write`

Unfortunately, seems that if we reach the max retries for a record (or the record its not retriable).

We fill the atomic reference `io.aiven.connect.elasticsearch.bulk.BulkProcessor.error` without never freeing that variable.
Because of that, since each method its guarded by `io.aiven.connect.elasticsearch.bulk.BulkProcessor.throwIfTerminal`, like for e.g. the `io.aiven.connect.elasticsearch.bulk.BulkProcessor.add` method; isn't possible to forward once we fill this exception.

When the `dropInvalidMessage` its set to `true` in the `io.aiven.connect.elasticsearch.ElasticsearchWriter.tryWriteRecord` method (by setting the `drop.invalid.message` config to `"true"`) the try catches also the `bulkProcessor.add(record, flushTimeoutMs);` instruction while it should catch only the `convertRecord` method causing the connector to continuously dropping the new messages instead of dying, the semantics should be "I try to convert the record and if the shape it's not the expected one I skip the record", while the code was doing: "I try to convert and send async the messages in the queue and if any error happens I throw an exception and I skip the records".

This pr fixes it and add a test that expects an exception is thrown from the method when trying to add new records after an error happens in the OS client.

src/main/java/io/aiven/connect/elasticsearch/ElasticsearchWriter.java

@@ -284,16 +284,15 @@ private void tryWriteRecord(
         final boolean ignoreKey,
         final boolean ignoreSchema) {
 
+        IndexableRecord record = null;
         try {
-            final IndexableRecord record = converter.convertRecord(
+            record = converter.convertRecord(
                 sinkRecord,
                 index,
                 type,
                 ignoreKey,
-                ignoreSchema);
-            if (record != null) {
-                bulkProcessor.add(record, flushTimeoutMs);
-            }
+                ignoreSchema
+            );
         } catch (final ConnectException convertException) {
             if (dropInvalidMessage) {
                 log.error(
@@ -308,6 +307,10 @@ private void tryWriteRecord(
                 throw convertException;
             }
         }
+
+        if (record != null) {
+            bulkProcessor.add(record, flushTimeoutMs);
+        }
     }
 
     /**

src/test/java/io/aiven/connect/elasticsearch/ElasticsearchSinkFailureTest.java

@@ -0,0 +1,109 @@
+/*
+ * Copyright 2024 Aiven Oy
+ * Copyright 2016 Confluent Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.aiven.connect.elasticsearch;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.kafka.connect.errors.ConnectException;
+import org.apache.kafka.connect.sink.SinkRecord;
+
+import io.aiven.connect.elasticsearch.bulk.BulkResponse;
+import io.aiven.connect.elasticsearch.jest.JestElasticsearchClient;
+
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.when;
+
+
+public class ElasticsearchSinkFailureTest {
+
+    @Test
+    public void testRetryIfRecoverable() throws IOException {
+        final ElasticsearchSinkTask elasticsearchSinkTask = new ElasticsearchSinkTask();
+        final int numbRetriesBeforeSucceeding = 3;
+
+        final JestElasticsearchClient failingClient = Mockito.mock(JestElasticsearchClient.class);
+        final AtomicInteger apiCallCounter = new AtomicInteger(0);
+        when(failingClient.executeBulk(any())).thenAnswer(i -> {
+            final int numAttempt = apiCallCounter.incrementAndGet();
+            if (numAttempt < numbRetriesBeforeSucceeding) {
+                return BulkResponse.failure(true, "");
+            }
+            return BulkResponse.success();
+        });
+
+        final Map<String, String> props  = new HashMap<>();
+        props.put(ElasticsearchSinkConnectorConfig.TYPE_NAME_CONFIG, "kafka-connect");
+        props.put(ElasticsearchSinkConnectorConfig.CONNECTION_URL_CONFIG, "localhost");
+        props.put(ElasticsearchSinkConnectorConfig.KEY_IGNORE_CONFIG, "true");
+        props.put(ElasticsearchSinkConnectorConfig.DROP_INVALID_MESSAGE_CONFIG, "true");
+        props.put(ElasticsearchSinkConnectorConfig.MAX_RETRIES_CONFIG, String.valueOf(numbRetriesBeforeSucceeding));
+
+        elasticsearchSinkTask.start(props, failingClient);
+        elasticsearchSinkTask.put(Collections.singletonList(new SinkRecord("topic", 0, null, null, null, "foo", 0)));
+        elasticsearchSinkTask.flush(Collections.emptyMap());
+        // 1 success
+        assertEquals(numbRetriesBeforeSucceeding, apiCallCounter.get());
+        elasticsearchSinkTask.put(Collections.singletonList(new SinkRecord("topic", 0, null, null, null, "bar", 0)));
+        elasticsearchSinkTask.flush(Collections.emptyMap());
+        // 3 retries (the max allowed) + 1 success
+        assertEquals(numbRetriesBeforeSucceeding + 1, apiCallCounter.get());
+    }
+
+    @Test
+    public void testRaiseExceptionIfNot() throws IOException {
+        final ElasticsearchSinkTask elasticsearchSinkTask = new ElasticsearchSinkTask();
+        final JestElasticsearchClient failingClient = Mockito.mock(JestElasticsearchClient.class);
+        final AtomicInteger apiCallCounter = new AtomicInteger(0);
+        when(failingClient.executeBulk(any())).thenAnswer(i -> {
+            apiCallCounter.incrementAndGet();
+            return BulkResponse.failure(false, "");
+        });
+
+        final Map<String, String> props = new HashMap<>();
+        props.put(ElasticsearchSinkConnectorConfig.TYPE_NAME_CONFIG, "kafka-connect");
+        props.put(ElasticsearchSinkConnectorConfig.CONNECTION_URL_CONFIG, "localhost");
+        props.put(ElasticsearchSinkConnectorConfig.KEY_IGNORE_CONFIG, "true");
+        props.put(ElasticsearchSinkConnectorConfig.DROP_INVALID_MESSAGE_CONFIG, "true");
+
+        elasticsearchSinkTask.start(props, failingClient);
+        elasticsearchSinkTask.put(Collections.singletonList(new SinkRecord("topic", 0, null, null, null, "foo", 0)));
+
+        // test that flush throws an exception
+        assertThrows(ConnectException.class, () -> elasticsearchSinkTask.flush(Collections.emptyMap()));
+        assertEquals(1, apiCallCounter.get());
+
+        // test that the exception is raised also if we try to add another record
+        final ConnectException e = assertThrows(ConnectException.class, () -> elasticsearchSinkTask.put(
+                Collections.singletonList(
+                    new SinkRecord("topic", 0, null, null, null, "bar", 0))
+            )
+        );
+        // the atomic variable isn't changed, the first exception is raised. The connector
+        // instance should die after the first exception since kafka connect will restart it
+        assertEquals(1, apiCallCounter.get());
+    }
+}