suraj-goel opened a new issue, #20727: URL: https://github.com/apache/druid/issues/20727
### Motivation When a streaming ingestion task fails, `SeekableStreamSupervisor` creates a replacement task almost at once, with no delay and no limit. This is the right behavior for a one-off failure. But if the failure is *persistent and fast* — every new task fails within milliseconds of starting — the supervisor creates and fails tasks in a tight loop until an operator steps in. **How the loop happens** (master `a9d7cf96`): 1. A task status change adds a `RunNotice` ([`SeekableStreamSupervisor.java#L2184-L2186`](https://github.com/apache/druid/blob/a9d7cf9629b5e29832c0143df568ad42421ed136/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java#L2184-L2186)). Notices run at most once per second per supervisor ([`MAX_RUN_FREQUENCY_MILLIS`, #L174](https://github.com/apache/druid/blob/a9d7cf9629b5e29832c0143df568ad42421ed136/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java#L174)). 2. `checkCurrentTaskState()` removes the failed task from its group ([#L4422](https://github.com/apache/druid/blob/a9d7cf9629b5e29832c0143df568ad42421ed136/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java#L4422)). Or `checkPendingCompletionTasks()` removes the whole group when all its publishing tasks fail or time out ([#L4335](https://github.com/apache/druid/blob/a9d7cf9629b5e29832c0143df568ad42421ed136/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java#L4335)). 3. In the same run, `createNewTasks()` recreates the group ([#L4530](https://github.com/apache/druid/blob/a9d7cf9629b5e29832c0143df568ad42421ed136/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java#L4530)) or tops up its replicas ([#L4663](https://github.com/apache/druid/blob/a9d7cf9629b5e29832c0143df568ad42421ed136/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java#L4663)). 4. This continues even when the supervisor's own stream calls fail: `runInternal()` returns early only if the supervisor has never had a successful run ([#L2095](https://github.com/apache/druid/blob/a9d7cf9629b5e29832c0143df568ad42421ed136/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java#L2095)). The supervisor-health thresholds (`druid.supervisor.taskUnhealthinessThreshold`) change the supervisor *state*, but they do not slow down relaunches. **Failures that make every new task fail fast.** With the Kafka client (3.9.2 in Druid 34, 4.3.1 on master; same code paths in `ClientUtils` and `Metadata`), the task fails when the consumer is built, or at the first `poll()`, for: * a SASL authentication failure (for example, revoked or wrongly rotated credentials), * a topic authorization failure (an ACL was removed), * `ConfigException: No resolvable bootstrap urls given in bootstrap.servers` (DNS failure), which surfaces as `KafkaException: Failed to construct kafka consumer`. Broker outages, broker throttling and a deleted topic do **not** fail the task: the Kafka client retries internally. So this proposal is *not* about broker outages. **What we observed.** We run about 1,000 Kafka indexing tasks on Druid 34 with `HttpRemoteTaskRunner` and Indexers. During a DNS outage in one availability zone, tasks on one Indexer failed with `Failed to construct kafka consumer` 41–79 ms after they started. Worker blacklisting stopped the loop after 6 consecutive failures, about 10 s later. It worked only because the fault was local to a few workers. **Why worker blacklisting is not enough:** * `HttpRemoteTaskRunner` blacklists a worker after more than `maxRetriesBeforeBlacklist` (default 5) consecutive failures, but only while fewer than `maxPercentageBlacklistWorkers` (default 20%) of workers are blacklisted ([`HttpRemoteTaskRunner.java#L797`](https://github.com/apache/druid/blob/a9d7cf9629b5e29832c0143df568ad42421ed136/indexing-service/src/main/java/org/apache/druid/indexing/overlord/hrtr/HttpRemoteTaskRunner.java#L797), [`HttpRemoteTaskRunnerConfig.java#L49-L62`](https://github.com/apache/druid/blob/a9d7cf9629b5e29832c0143df568ad42421ed136/indexing-service/src/main/java/org/apache/druid/indexing/overlord/config/HttpRemoteTaskRunnerConfig.java#L49-L62)). For a cluster-wide fault (credentials, ACLs, cluster-wide DNS, a bad `consumerProperties` value), the other 80% of workers keep receiving tasks that fail at once. * `KubernetesTaskRunner` has no worker blacklisting ([`getBlacklistedTaskSlotCount()` returns an empty map, #L513](https://github.com/apache/druid/blob/a9d7cf9629b5e29832c0143df568ad42421ed136/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunner.java#L513)). There, each failed attempt is a new pod. * Blacklisting acts per worker and for all task types. The problem is per task group of one supervisor. **Cost of each attempt** in the loop: * Overlord: a `TaskQueue` entry, task assignment and status sync. * Metadata store: one task row with the full task payload (5–60 KB per task with our specs), plus a status update. Both are kept until the task log retention expires. * Deep storage: the task report and the task log (two objects per attempt). * Kafka: one connection with a failed TLS + SASL handshake (for authentication failures). * Kubernetes runner: one pod. The related upstream fix [#20387](https://github.com/apache/druid/pull/20387) stopped a similar endless replica top-up loop for bounded supervisors with a targeted guard. This proposal asks for a general safety net for the streaming case. ### Proposed changes Track consecutive failures **per task group** and delay task creation for that group with exponential backoff and jitter. **Behavior** * Failure 1: relaunch at once (today's behavior). This keeps fast recovery for one-off failures, such as a worker restart. * Failure n ≥ 2: wait `min(maxDelay, initialDelay × multiplier^(n−2)) × U(1 − jitter, 1 + jitter)` before creating tasks for that group. With the proposed defaults the waits are 0, 1, 1.6, 2.6, 4.1 … s, reaching the 5 min cap at about the 15th consecutive failure. * Reset the counter when a task of the group succeeds, or when a task of the group has been running for `resetAfter` after it reported its start time (`TaskData.startTime`). The second rule means a recovered group does not wait for the full `taskDuration`. * Clear all state on `resetInternal()`, so an operator reset relaunches at once, and on `clearPartitionAssignmentsForScaling()`, because group IDs change. * Do not count supervisor-initiated stops: tasks that are no longer current, partition-set changes, graceful shutdown, suspend, `handoffTaskGroupsEarly`. * When a group is waiting, schedule one `RunNotice` at the earliest expiry. Without it, the delay rounds up to the next `ioConfig.period`. **Configuration** (Overlord runtime properties, following the existing `druid.supervisor.idleConfig.*` pattern in `SupervisorStateManagerConfig`): | Property | Proposed default | |---|---| | `druid.supervisor.taskRelaunchBackoff.enabled` | `false` | | `druid.supervisor.taskRelaunchBackoff.initialDelayMillis` | `1000` | | `druid.supervisor.taskRelaunchBackoff.multiplier` | `1.6` | | `druid.supervisor.taskRelaunchBackoff.maxDelayMillis` | `300000` | | `druid.supervisor.taskRelaunchBackoff.jitter` | `0.2` | | `druid.supervisor.taskRelaunchBackoff.resetAfterMillis` | `300000` | Optional: a per-supervisor override in `ioConfig.taskRelaunchBackoff`, the same way `ioConfig.idleConfig` overrides the cluster default. **Observability** * A WARN log once per backoff step: `taskGroup[%d] of supervisor[%s] failed [%d] times in a row; next task launch in [%d] ms`. * A metric, for example `ingest/supervisor/taskGroupsInBackoff`, with `dataSource` and `supervisor` dimensions (name to be agreed). * Optional: the groups in backoff and their next launch times in the supervisor `/status` report. **Implementation shape:** a small package-private helper (per-group state, injectable clock and random source) plus a few hooks in `SeekableStreamSupervisor`. The change applies to Kafka, Kinesis and RabbitStream through the base class. We are happy to open the PR. ### Rationale Alternatives we considered: 1. **Supervisor-level gate.** Skip `createNewTasks()` while the supervisor's own stream call fails with a non-retriable error (authentication, authorization, config). It is simpler and covers the Kafka cases above. But it does not cover fast failures on the Druid side (for example, a bad spec property or a task setup error). It also stops all task creation if only the Overlord has a network problem. 2. **Change worker blacklisting** (higher percentage, longer backoff). It applies to every task type, does not help the Kubernetes runner, and can blacklist the whole cluster for an hour after the fault is fixed. 3. **Rate limit in `TaskQueue` per datasource.** Generic, but the queue does not know the stream-specific structure: task groups, replicas and the difference between a failed task and a supervisor-initiated stop. 4. **Backoff only for some failure types.** `TaskStatus.errorMsg` is free text, so classifying failures would be fragile. Counting consecutive failures per group is simpler and independent of the cause. Per-group backoff keeps healthy groups unaffected, keeps the first relaunch immediate, and caps the cost of a persistent failure. With the proposed defaults, at the cap, 1,000 failing groups make about 3 attempts/s in total instead of hundreds. ### Operational impact * Default `enabled=false`: no behavior change unless an operator enables it. * Only the Overlord runs this code. Rolling upgrade and downgrade are safe: older versions ignore the new properties, because `DefaultObjectMapper` sets `FAIL_ON_UNKNOWN_PROPERTIES=false`. * When enabled, recovery after *repeated* failures can be delayed by up to `maxDelayMillis` plus the task start time. The first failure is not delayed. * Backoff state is in memory. An Overlord leader change resets it. ### Test plan (optional) * Unit tests for the helper: delay sequence, cap, jitter bounds, no delay on the first failure, reset and clear. * `SeekableStreamSupervisorStateTest` and `KafkaSupervisorTest`: a failing task is not recreated inside the backoff window; success and `resetInternal` reset it; disabled mode matches current behavior. * An embedded test (next to `KafkaIndexFaultToleranceTest` in `embedded-tests`): a supervisor whose tasks fail at once (unresolvable bootstrap or bad SASL credentials). Assert that the number of task launches in a fixed window is bounded, and that ingestion recovers once the fault is fixed. ### Future work (optional) * Show groups in backoff in the web console supervisor view. * Decide whether the same safety net is useful for other supervisor types. ### Questions for committers We would value opinions on these before we write the PR: 1. Is a relaunch backoff wanted upstream, or has this been discussed before or handled somewhere we missed? 2. Is `SeekableStreamSupervisor` the right layer, or should this live in `TaskQueue` or the task runner? 3. Per task group, or per supervisor? 4. Configuration: Overlord-wide defaults plus a per-supervisor override (like `idleConfig`)? Default disabled, or enabled with conservative values? 5. Reset rule: is "running for `resetAfter` after the start time was reported" a good signal, or would you prefer success only? 6. How should this interact with the autoscaler, bounded mode (#20387), server priorities (#19040) and `handoffTaskGroupsEarly`? 7. Should the supervisor state show the backoff (for example, a detailed state under `UNHEALTHY_TASKS`)? 8. Do you prefer a specific metric name or dimension convention? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
