fix: restrict DispatchingTransport retries to transient failures only

sendAsync retried on any non-null error, wasting the retry budget on
permanent 4xx failures. The exception hierarchy made this subtle:
ResponseException and TransportException both extend IOException, so a
naive instanceof-IOException check would retry HTTP 400/401/403/404/405
(delivered as ElasticsearchException via the LLRC ignore list) and HTTP
409 Conflict (delivered as ResponseException).

Add isRetryable(Throwable): retry genuine transport failures (IOException
subclasses that are neither ResponseException nor TransportException),
HTTP 429 and HTTP 5xx (transient server pressure); do not retry
ElasticsearchException (permanent app-level 4xx), TransportException
(infrastructure misconfiguration), or ResponseException for other 4xx.

Residual note: retrying IOException on bulk requests with
use.autogenerated.ids=true can still create duplicate documents if ES
applied operations before the connection dropped. This cannot be resolved
inside DispatchingTransport without external versioning or ES-side
deduplication.

Author: yeikel

src/main/java/io/confluent/connect/elasticsearch/ElasticsearchClient.java

@@ -25,10 +25,13 @@
 import co.elastic.clients.elasticsearch.indices.get_mapping.IndexMappingRecord;
 import co.elastic.clients.json.JsonpMapper;
 import co.elastic.clients.json.jackson.JacksonJsonpMapper;
+import co.elastic.clients.elasticsearch._types.ElasticsearchException;
 import co.elastic.clients.transport.ElasticsearchTransport;
 import co.elastic.clients.transport.Endpoint;
+import co.elastic.clients.transport.TransportException;
 import co.elastic.clients.transport.TransportOptions;
 import co.elastic.clients.transport.rest_client.RestClientTransport;
+import org.elasticsearch.client.ResponseException;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
@@ -760,10 +763,7 @@ private <RequestT, ResponseT, ErrorT> void sendAsync(
       delegate.performRequestAsync(request, endpoint, options)
           .whenCompleteAsync(
               (resp, err) -> {
-                // Retry any transport failure (SocketTimeoutException, ConnectException,
-                // ResponseException for 4xx/5xx, etc.), mirroring callWithRetries() in the
-                // original HLRC buildConsumer() which caught Exception unconditionally.
-                if (err != null && retriesLeft > 0) {
+                if (err != null && retriesLeft > 0 && isRetryable(err)) {
                   scheduler.schedule(
                       () -> sendAsync(request, endpoint, options, result,
                           retriesLeft - 1, backoffMs * 2),
@@ -778,6 +778,29 @@ private <RequestT, ResponseT, ErrorT> void sendAsync(
           );
     }
 
+    // ResponseException and TransportException are both IOException subclasses; they must be
+    // checked before the general IOException branch so the more-specific type wins.
+    private static boolean isRetryable(Throwable err) {
+      // HTTP-level error from the LLRC: only retry transient codes
+      if (err instanceof ResponseException) {
+        int status = ((ResponseException) err).getResponse().getStatusLine().getStatusCode();
+        return status == 429 || status >= 500;
+      }
+      // TransportException covers permanent structural failures: missing/invalid X-Elastic-Product
+      // header, wrong content-type, missing response body, undecodable response, unrecognised
+      // status code. None of these improve on retry.
+      if (err instanceof TransportException) {
+        return false;
+      }
+      // Permanent application-level ES errors (4xx via LLRC ignore list, deserialized by
+      // ElasticsearchTransportBase as ElasticsearchException)
+      if (err instanceof ElasticsearchException) {
+        return false;
+      }
+      // Remaining IOExceptions are genuine transport failures (SocketTimeout, ConnectException…)
+      return err instanceof IOException;
+    }
+
     @Override
     public JsonpMapper jsonpMapper() {
       return delegate.jsonpMapper();