Hi everyone, I’m migrating a Flink elastic search storage job from the ES7 connector to the ES8 connector and need guidance on failure handling. Setup: • Flink 2.2.1, Java 21. • Existing:

flink-connector-elasticsearch7:4.0.0-2.0

, client 7.10.2. • New:

flink-connector-elasticsearch8:4.0.0-2.0

, overriding the Java client to 8.19.22.

.setBulkResponseInspectorFactory(new ElasticsearchFailureHandler(host))
.setBulkFlushBackoffStrategy(FlushBackoffType.EXPONENTIAL, maxRetries, delay)

Our custom inspector classified failed bulk items and decided whether to ignore selected permanent document errors, perform restricted recovery, or fail the task. With

Elasticsearch8AsyncSinkBuilder

, I cannot find equivalent failure-handler or bounded-retry options. The released writer appears to requeue all failed bulk items without distinguishing permanent errors from transient ones or limiting attempts. Permanent failures could therefore retry indefinitely; task restarts can also replay the same problematic records. Is there a supported replacement/extension point for error classification, bounded retries/backoff, and durable DLQ handling, without maintaining a custom sink/writer? Was this intentionally changed with the async-sink architecture, or is it a feature-parity gap? I found FLINK-35546 and PR #107. Are these the right items to follow? We want to quarantine individual bad records or run time exceptions from elastic search such as index closed , shrink exceptions safely.