suvodeep-pyne opened a new pull request, #20076: URL: https://github.com/apache/pinot/pull/20076
Part 2 of 3. It uses the classifier from part 1, #20067, and is its first consumer. Part 3 will add Kinesis and Pulsar fixes. **Stacked on #20067.** The first 11 commits are part 1. To review only this part, see [the last 10 commits](https://github.com/suvodeep-pyne/pinot/compare/spyne/rt-source-error-classifier...spyne/rt-source-error-sites). After #20067 merges, I will rebase this PR onto master. ## Problem When a realtime table's stream refuses Pinot or stops serving it, operators see only a stall: lag grows, `llcPartitionConsuming` drops, and the reason is in a server log line, if anywhere. Three gaps hide it: - **An auth failure interleaved with empty fetches never trips the consume loop's give-up.** The loop never reaches its error path, so nothing records a failure. - **Typed evidence is discarded.** Kafka's partition metadata fetch gives up with a cause-less message, and a synthesized "Topic does not exist" carries nothing a classifier can read. - **Plugins swallow failures on purpose**, returning an empty or partial result, so no site sees them. ## Approach Each site that already sees a source failure classifies it and sets a timestamp gauge for the category: - **AUTH is recorded at every occurrence.** A refusal needs a human, and the loop may never give up on it. - **SOURCE_READ is recorded only once Pinot gives up.** That is six consecutive errors in the consume loop, or retries exhausted elsewhere, so a blip Pinot survives records nothing. - **Nothing is recorded while a segment is being stopped.** The stop's own interrupt is not the source giving up. The gauge's value is the epoch second of the last failure, so a dashboard or alert reads freshness without rate math. Gauges are never cleared on success. They are removed when the table goes away. ```mermaid flowchart LR subgraph Server F[Consume loop: fetch, create, checkpoint] --> R[IngestionSourceErrorRecorder<br/>per table] D[Decoder init, row decode] --> R T[Ingestion delay tracker] --> R end subgraph Controller M[getStreamMetadataList] --> CG[Table gauge] end P[Plugin swallows or retries a failure] -->|IngestionSourceErrorReporter.report| F P -->|IngestionSourceErrorReporter.report| M R --> SG[Partition gauge<br/>ingestionSource{Auth,Read}FailureTimeSeconds] R --> W[WARN line, rate-limited:<br/>category, sub-reason, code, target; never a message] CG --> W ``` ## Key changes - **Gauges.** `ServerGauge` and `ControllerGauge` gain `INGESTION_SOURCE_AUTH_FAILURE_TIME_SECONDS` and `INGESTION_SOURCE_READ_FAILURE_TIME_SECONDS`: a partition gauge on the server and a table gauge on the controller. They export through the existing JMX rules as `pinot_server_ingestionSource{Auth,Read}FailureTimeSeconds_Value{database,table,tableType,partition}` and the controller equivalent. - **Server.** `IngestionSourceErrorRecorder`, one per table data manager, is fed by the consume loop (fetch, consumer creation and re-creation, checkpoint), decoder init, row decode and the ingestion delay tracker. - It closes first on table shutdown. Writes serialize with close, so a stalled consumer thread can't write a gauge after the table is gone or remove a recreated table's gauge. - **Controller.** `PinotTableIdealStateBuilder.getStreamMetadataList` records what it gives up on, including a topic skipped as permanently failed in a multi-topic fetch. - Table deletion removes both gauges. A fetch still running at the deletion, on any controller, can write one afterwards, so each controller remembers the tables it wrote a gauge for, and its next `SegmentStatusChecker` run removes those whose table config is gone. - **`IngestionSourceErrorReporter`** (`pinot-spi`) lets a plugin hand a failure it swallows or retries to the site that called it, through a thread-local scope. The fetch, consumer-creation, checkpoint and metadata sites install the scope. - **Retried failures stay out of the exceptions Pinot shows.** Decoder init and Kafka consumer re-creation classify the last attempt's failure but don't attach it as a cause, because the segment error reaches the server's debug API and a client's message can quote stream config values. Those exceptions are unchanged from master. - **Kafka evidence.** - The partition metadata fetch keeps the last failure as the cause. - The synthesized "Topic does not exist" carries `UNKNOWN_TOPIC_OR_PARTITION`, and the controller's listing timeout carries `UNREACHABLE`. - For `AWS_MSK_IAM`, a SASL failure records whose credential signed: the process's own AWS credential chain, or role keys configured on the table. It reads only which JAAS options are present, and never logs them. - **Row decode.** The per-row decode error log is rate-limited. The WARN line carries the category, sub-reason, code, target, credential mode and deciding exception class, never an exception message. ## Where to start `IngestionSourceErrorRecorder` (the server's rules for when to write), then `RealtimeSegmentDataManager`'s fetch path and `recordSourceError`, then `PinotTableIdealStateBuilder.recordSourceFailure`. ## Testing - `IngestionSourceErrorSaslKafkaIntegrationTest` drives a real controller and server against an embedded Kafka with a SASL/PLAIN listener and the standard authorizer. A table created with bad credentials is rejected and deleted, so each case breaks a working table: - revoking the topic ACL sets the server partition's AUTH gauge at the first refusal, and the controller's after validation; - deleting the topic, and stopping the broker, each set SOURCE_READ through the delay tracker and the controller; - a schema registry answering 403 sets AUTH from decode while fetches succeed, and bytes that don't parse set nothing. - Unit suites: `RealtimeSegmentDataManagerSourceErrorTest`, `IngestionSourceErrorRecorderTest`, `PinotTableIdealStateBuilderSourceErrorTest`, `IngestionDelayTrackerTest`, `IngestionSourceErrorReporterTest`, and the Kafka 3.x and 4.x evidence tests. ### Verification An independent verifier ran each check at this change's head and at its base (#20067's head): | Check | Base | Head | Verdict | |---|---|---|---| | End to end on SASL Kafka with an authorizer: revoke the ACL, delete the topic, stop the broker, registry 403 vs bad bytes | No gauge on either role in any case | Every case sets the gauge it should on both roles | pass | | An auth failure interleaved with empty fetches writes AUTH; the loop never gives up | 0 writes | AUTH at every refusal | pass | | Five transient errors then a success write nothing; six write SOURCE_READ once | 0 writes | 0 for 5 + success; exactly 1 for 6 | pass | | A registry 403 during decode, with a malformed row earlier in the batch, sets and keeps AUTH | Never set | Set with no fetch errors; kept after the registry recovers | pass | | Decoder init failing with a registry 401 on every attempt sets AUTH; the debug API's segment error carries no client message | No gauge | AUTH set; the segment error is the retry policy's cause-less exception, as on master | pass | | Stopping a segment mid-retry or while its consumer is being recreated writes nothing | Gauges don't exist | Nothing written on table delete or disable. Without the stop guard, a disable wrote SOURCE_READ. | pass | | The gauge survives CONSUMING→OFFLINE and the next consuming segment, and is removed on table deletion, server and controller | Never set | Holds; a controller fetch landing after its table's deletion writes the gauge, and the next status checker run removes it | pass | | A classifier or recording failure never changes consumption | — | Same loop outcome and rows when every gauge write throws, or a rule overflows the stack | pass | | Both gauges export as the exact Prometheus series under the OSS JMX rules | Gauges don't exist | Exact series, same labels as the existing join gauges | pass | Not driven end to end: a late write from a second controller (the test cluster has one), and a table re-created while the sweep runs (a unit test with a latch covers it). -- 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]
