suvodeep-pyne opened a new pull request, #20082: URL: https://github.com/apache/pinot/pull/20082
Part 3 of 3. It builds on part 2, #20076, which records source failures as gauges, and on part 1, #20067, the classifier. It changes the Kinesis and Pulsar plugins only. **Stacked on #20076.** The first 22 commits are parts 1 and 2. To review only this part, see [the last 3 commits](https://github.com/suvodeep-pyne/pinot/compare/spyne/rt-source-error-sites...spyne/rt-source-error-kinesis-pulsar). As the earlier parts merge, I will rebase this PR onto master. ## Problem Three failures never reach the classifier part 2 wired in: - **Kinesis throttling is invisible.** Once its retries run out, the consumer returns an empty batch, and the consume loop counts that as a successful fetch. A stream throttled for hours looks like a quiet one. - **A Kinesis credential rejection can't say whose credential it was.** The same AWS code (`UnrecognizedClientException`, `InvalidSignatureException`, or STS `InvalidClientTokenId`) means the table owner's bad key under static keys, but Pinot's own identity under the default chain. Without that, the classifier can only call it none, which misses the table owner's AUTH. - **A Pulsar reader can hang forever.** `readNext()` blocks without a deadline. When a lagging reader's connection drops and its grant has been revoked, the reconnect is refused in the background and the fetch never returns, so nothing is reported and the partition stalls. Separately, `computePartitionGroupMetadata` swallows a refused partition list or subscribe, and a denied new table silently gets no segments. ## Approach ```mermaid flowchart LR subgraph Kinesis K1["Each client call"] -->|"attaches whose credential signed"| E["Evidence on the SDK exception"] K2["Fetch exhausts throttle retries"] --> C{"6th in a row<br/>for this client id and shard?"} C -->|"yes"| R["Report the throttle exception,<br/>still return the empty batch"] end subgraph Pulsar P1["readNext bounded by the fetch deadline"] --> D{"Disconnected 30 s?"} D -->|"yes"| T["Transient exception: recreate,<br/>the fresh handshake fails with the real refusal"] P2["Partition list or subscribe refused"] --> M["Report it, keep the partial list"] end E --> CL["Classifier"] R --> CL T --> CL M --> CL ``` ## Key changes - **Kinesis evidence.** `KinesisConnectionHandler` attaches the credential mode (the table's static keys or the process's chain, and whether a role is assumed) to every exception a client call throws, including the raw `StsException` from the assume-role provider. - **Kinesis throttling.** From the sixth throttle-exhausted fetch in a row, the consumer reports the throttling exception and still returns the empty batch, so consumption is unchanged. - The count is kept per client id and shard. It survives the consume loop recreating the consumer, and another table's healthy consumer of the same stream can't reset it. - Only a fetch that returns resets it. Pinot's own request limiter timing out neither counts nor resets. - **Pulsar reads.** `readNext` is bounded by the fetch deadline. A reader disconnected for 30 seconds throws a transient exception, so the consume loop recreates it and the new handshake fails with the broker's actual refusal, with or without idle recreation. This is the only consumption change in the PR. - **Pulsar metadata.** `computePartitionGroupMetadata` reports the failures it used to swallow and keeps returning the partial list. ## Where to start `KinesisConsumer` (the throttle count), then `PulsarPartitionLevelConsumer` (the bounded read). ## Testing - `PulsarSourceErrorComponentTest`: a Testcontainers Pulsar with token auth and authorization on. It covers a bad token; a grant revoked under a lagging reader that is then forced to reconnect; and the refused partition list and subscribe. - `KinesisCredentialEvidenceTest`: a stub Kinesis and STS behind the real `KinesisClient`, for static keys, the default chain, and an assumed role. - `KinesisConsumerTest`: throttle exhaustion through the real factory and SDK. ### Verification An independent verifier ran each check at this change's head and at its base: | Check | Base | Head | Verdict | |---|---|---|---| | Pulsar, token auth and authorization: a bad token; a grant revoked under a lagging reader | The lagging reader hangs in `readNext` (the test hits its 120 s timeout) | Fetches return empty in about 1 s; after 30 s disconnected the reader gives up, and the recreate fails as AUTH/authorization | pass | | Pulsar metadata refusals | Nothing reported; a denied new table gets no segments | Each refusal is reported once as AUTH. Through the controller's real `getStreamMetadataList`, the table's AUTH gauge is set. A granted role sets nothing. | pass | | Kinesis credential mode, through the real `KinesisClient` and a stub STS | The table's rejected keys classify as none | The table's static keys rejected: AUTH. The process's chain or minted session credentials: none. `AccessDenied`: AUTH in every mode. | pass | | Kinesis throttle exhaustion | Never reported across 20 exhausted fetches | Nothing for fetches 1–5, reported from the 6th; a healthy fetch resets the count; it survives recreation and is isolated from other tables | pass | | Observation only | 52 existing tests pass | All existing tests in both modules pass; the only consumption change is the Pulsar one above | pass | Not run: a real AWS account (LocalStack doesn't enforce IAM), and the server's real consume loop for either plugin. The Pulsar recreate went through the same factory call the consume loop makes, and Kinesis reports reached a listener installed the way the consume loop installs it. Known limit: a throttle count is removed only by a fetch that returns, so a table deleted or a shard closed while throttled leaves one small entry behind. -- 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]
