suvodeep-pyne commented on code in PR #20067: URL: https://github.com/apache/pinot/pull/20067#discussion_r4406679660
########## pinot-spi/src/main/java/org/apache/pinot/spi/ingestion/error/IngestionSourceErrorClassifier.java: ########## @@ -0,0 +1,295 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.spi.ingestion.error; + +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.Deque; +import java.util.IdentityHashMap; +import java.util.List; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.CopyOnWriteArraySet; +import javax.annotation.Nullable; + + +/// Folds a failure to reach an ingestion source into [IngestionSourceErrorCategory#AUTH] or +/// [IngestionSourceErrorCategory#SOURCE_READ], from the source's own verdict on our request. Status codes and error +/// codes are only evidence for that verdict, read per source. +/// +/// The exception tree is split into chains: the failure with its causes, and each suppressed exception with its own +/// causes. When an exception in the tree carries boundary evidence ([IngestionSourceErrorEvidence#setBoundary]), only +/// the parts of chains at or beneath a boundary are read; otherwise the caller vouches that the whole failure came +/// from the source. Within a chain: +/// +/// 1. walking from the root cause up, the first source-specific rule that decides wins. A rule that says the failure +/// is not the source's verdict (our clock, our ambient identity, a credential that failed to load) therefore +/// decides only when no source exception sits beneath it, and a wrapper above it can't reinstate a category; +/// 2. generic rules, which read the transport rather than a source's answer (`java.net` / `javax.net.ssl`, and any +/// registered with [#registerGenericRule]), apply only when no source rule decided the chain. +/// +/// Across chains [IngestionSourceErrorCategory#AUTH] beats [IngestionSourceErrorCategory#SOURCE_READ]: a refusal +/// recorded as a suppressed exception still surfaces. +/// +/// Built-in rules cover kafka-clients, the Confluent schema-registry client, the AWS SDK v2, the Pulsar client and +/// admin APIs, gRPC and gax, Kerberos and OAuth logins, and JDK network types. They match classes by name anywhere in +/// the hierarchy, because SDKs load in plugin classloaders, sometimes in two versions at once. They read codes +/// reflectively and fail closed. A plugin for another SDK adds rules with [#registerRule]. +/// +/// [#classify] is pure and never throws, apart from rethrowing the JVM's own fatal errors (`OutOfMemoryError`, +/// `InternalError`). +public final class IngestionSourceErrorClassifier { + // Bounds the walk over pathological exception graphs; real failures are a few dozen nodes. + private static final int MAX_NODES = 1_000; + private static final Set<IngestionSourceErrorRule> REGISTERED_RULES = new CopyOnWriteArraySet<>(); + private static final Set<IngestionSourceErrorRule> REGISTERED_GENERIC_RULES = new CopyOnWriteArraySet<>(); + + private IngestionSourceErrorClassifier() { + } + + /// Classifies `failure`, or returns empty when it is not the source's verdict on our request. + public static Optional<IngestionSourceErrorClassification> classify(Throwable failure) { + try { + Walk walk = new Walk(); + walk.visit(failure); + walk.decide(); + Decision decided = null; + for (Decision decision : walk._decisions) { + if (decision._verdict.getCategory() == IngestionSourceErrorCategory.AUTH) { + decided = decision; + break; + } + if (decided == null && decision._verdict.getCategory() == IngestionSourceErrorCategory.SOURCE_READ) { + decided = decision; + } + } + if (decided == null) { + return Optional.empty(); + } + return Optional.of(new IngestionSourceErrorClassification(decided._verdict, decided._node, decided._evidence)); + } catch (Throwable t) { + // Observation only: a classification bug, or a hostile exception, must never change how the failure itself is + // handled. + rethrowIfFatal(t); + return Optional.empty(); + } + } + + /// Running out of memory or a broken JVM is not ours to swallow; a stack overflow from a hostile exception is. + static void rethrowIfFatal(Throwable t) { + if (t instanceof VirtualMachineError && !(t instanceof StackOverflowError)) { + throw (VirtualMachineError) t; + } + } + + /// Adds rules for an SDK Pinot doesn't ship. Registering the same rule twice has no effect, and the order rules are + /// registered in never changes an outcome. + public static void registerRule(IngestionSourceErrorRule rule) { Review Comment: Part 2 is open as #20076, stacked on this PR, so the classifier now has its first consumer to review against. It calls `classify()` from the server consume loop, decoder init, the ingestion delay tracker and the controller's `getStreamMetadataList`, and adds one more `pinot-spi` type, `IngestionSourceErrorReporter`. Its last 10 commits are part 2 only: https://github.com/suvodeep-pyne/pinot/compare/spyne/rt-source-error-classifier...spyne/rt-source-error-sites -- 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]
