Explain errors.retry.timeout and errors.retry.delay.max.ms. How do retries interact with errors.tolerance?

## The problem retries solve Not all failures are equal. A **transient** failure — the downstream database is briefly unreachable, a connection reset, a momentary timeout — will likely succeed if you simply try again a moment later. A **deterministic** failure — malformed JSON, a schema violation — fails identically every time. Kafka Connect's retry mechanism exists for the transient class. ## errors.retry.timeout This is the **total time budget** (in ms) Connect will spend retrying a *single* failed operation before giving up: - **0** (default): no retries — fail (or tolerate) on the first error. - **positive N**: keep retrying for up to N milliseconds. - **-1**: retry indefinitely (use with care — a permanently broken downstream stalls the task forever). ## errors.retry.delay.max.ms Connect uses **exponential backoff with jitter** between retry attempts: the delay grows (roughly doubling) each attempt to avoid hammering a struggling downstream. `errors.retry.delay.max.ms` (default **60000**, i.e. 60s) is the **ceiling** on that per-attempt delay. So delays might go 300ms → 600ms → 1.2s → … capped at 60s. Jitter randomizes the exact value to prevent thundering-herd retries across tasks. ## Ordering: retries THEN tolerance This is the most-tested nuance. The pipeline for a failing operation is: 1. Operation fails. 2. Connect retries within the `errors.retry.timeout` budget, backing off up to `errors.retry.delay.max.ms`. 3. If a retry succeeds → processing continues normally, no error recorded. 4. If the budget is exhausted and it still fails → **now** `errors.tolerance` decides: - `none` → task goes to FAILED and stops. - `all` → the record is logged (if `errors.log.enable`) and/or sent to the DLQ, then skipped. So retries and tolerance are complementary, not alternatives: retries handle 'maybe transient', tolerance handles 'definitely give up now'. ## Which failures are retryable Retries apply primarily to operations that can throw a **RetriableException** — typically the sink connector's `put()` against the external system, or producer/consumer interactions. Pure conversion/SMT failures on a bad record are deterministic and effectively non-retriable; configuring a large retry timeout there just delays the inevitable skip/fail by the whole budget. ## Practical guidance - For sinks writing to flaky downstreams: set a modest `errors.retry.timeout` (e.g. 30000–300000) so brief outages self-heal without failing the task. - Keep `errors.retry.delay.max.ms` reasonable so backoff doesn't grow to many minutes and stall the partition. - Avoid `-1` unless you have external monitoring/alerting on a stuck task. - Combine with `errors.tolerance=all` + DLQ so that records still failing after the retry budget are quarantined rather than crashing the task.