xiangfu0 commented on code in PR #20067: URL: https://github.com/apache/pinot/pull/20067#discussion_r4397855535
########## pinot-spi/src/main/java/org/apache/pinot/spi/ingestion/error/IngestionSourceErrorRules.java: ########## @@ -0,0 +1,758 @@ +/** + * 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.net.BindException; +import java.net.SocketException; +import java.net.SocketTimeoutException; +import java.net.UnknownHostException; +import java.net.http.HttpTimeoutException; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Locale; +import java.util.Set; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import javax.annotation.Nullable; +import javax.net.ssl.SSLException; + + +/// The built-in rules [IngestionSourceErrorClassifier] applies, organised by SDK. +/// +/// Rules key on the source's own class and code first, and on an HTTP status only where the source has no code of +/// its own (a schema registry, Pulsar's admin API, an OAuth token endpoint) or AWS returned none. They read a message +/// only where the client stringified its evidence or the class covers both our side and the source's; each such match +/// is commented at its rule and pinned by a fixture test against the SDK version Pinot ships. A phrase an SDK upgrade +/// changes leaves the failure unclassified, never in the wrong category. +/// +/// They never key on Pinot's own wrappers, an outer plugin message, a bare Kafka `SerializationException` or +/// `InvalidConfigurationException`, or which exceptions a client considers retriable. +/// +/// A rule returns a verdict to decide its node, [IngestionSourceErrorVerdict#none] when the failure is not the +/// source's verdict on our request, or null to leave the node undecided. +final class IngestionSourceErrorRules { + private IngestionSourceErrorRules() { + } + + private static final String TARGET_ADMIN = IngestionSourceErrorVerdict.TARGET_ADMIN; + private static final String TARGET_BROKER = IngestionSourceErrorVerdict.TARGET_BROKER; + private static final String TARGET_IDP = IngestionSourceErrorVerdict.TARGET_IDP; + private static final String TARGET_KMS = IngestionSourceErrorVerdict.TARGET_KMS; + private static final String TARGET_SCHEMA_REGISTRY = IngestionSourceErrorVerdict.TARGET_SCHEMA_REGISTRY; + private static final String TARGET_STS = IngestionSourceErrorVerdict.TARGET_STS; + + // Our own interrupts and cancellations, matched exactly: SocketTimeoutException extends InterruptedIOException. + private static final Set<String> INTERRUPTS = Set.of("java.lang.InterruptedException", + "java.io.InterruptedIOException", "java.nio.channels.ClosedByInterruptException", + "java.util.concurrent.CancellationException"); + // Metadata endpoints serve the process's own ambient identity: AWS and Azure IMDS, the ECS and EKS Pod Identity + // container endpoints, the GCE metadata server. + private static final List<String> METADATA_ENDPOINTS = List.of("169.254.169.254", "169.254.170.2", + "169.254.170.23", "metadata.google.internal"); + // Socket failures on our side before any request: ephemeral ports or file descriptors exhausted. + private static final String OUR_SOCKETS_EXHAUSTED = "Too many open files"; + + // --------------------------------------------------------------------------------------------------------------- + // Evidence recorded by a client that is about to discard it + + @Nullable + static IngestionSourceErrorVerdict matchEvidence(IngestionSourceErrorEvidence evidence) { + String code = evidence.getCode(); + if (IngestionSourceErrorEvidence.UNREACHABLE.equals(code)) { + return sourceRead(code, evidence.getTarget()); + } + if (IngestionSourceErrorEvidence.LOCAL_PRECHECK.equals(code)) { + return auth(IngestionSourceErrorSubReason.AUTHENTICATION, code, evidence.getTarget()); + } + if (IngestionSourceErrorEvidence.LOCAL_SETUP.equals(code)) { + return none(code); + } + if (IngestionSourceErrorEvidence.UNKNOWN_TOPIC_OR_PARTITION.equals(code)) { + return sourceRead(code, evidence.getTarget() != null ? evidence.getTarget() : TARGET_BROKER); + } + // A status or another code means something only to the source that recorded it: a registered rule reads it. + return null; + } + + /// Source-specific rules for one node, reading the part of its chain beneath it where a rule needs to. + @Nullable + static IngestionSourceErrorVerdict matchSource(Throwable node, List<Throwable> below, + @Nullable IngestionSourceErrorEvidence evidence) { + Class<?> clazz = node.getClass(); + if (INTERRUPTS.contains(clazz.getName())) { + return none(clazz.getSimpleName()); + } + if (hasClass(clazz, AWS_SERVICE_EXCEPTION)) { + return awsService(node, evidence); + } + if (hasClass(clazz, AWS_CLIENT_EXCEPTION)) { + return awsClient(node, below); + } + if (hasClass(clazz, CONFLUENT_REST_CLIENT_EXCEPTION)) { + return schemaRegistry(node); + } + IngestionSourceErrorVerdict kafka = kafka(node, below, evidence); + if (kafka != null) { + return kafka; + } + if (hasClass(clazz, LOGIN_EXCEPTION)) { + return login(node); + } + if (hasClass(clazz, PULSAR_CLIENT_EXCEPTION) || hasClass(clazz, PULSAR_V5_CLIENT_EXCEPTION)) { + return pulsar(node); + } + if (hasClass(clazz, PULSAR_ADMIN_EXCEPTION)) { + return pulsarAdmin(node); + } + if (hasClass(clazz, GAX_API_EXCEPTION)) { + Object statusCode = invoke(node, "getStatusCode"); + return grpc(statusCode != null ? invoke(statusCode, "getCode") : null, node, evidence); + } + if (hasClass(clazz, GRPC_STATUS_RUNTIME_EXCEPTION) || hasClass(clazz, GRPC_STATUS_EXCEPTION)) { + Object status = invoke(node, "getStatus"); + return grpc(status != null ? invoke(status, "getCode") : null, node, evidence); + } + if (hasClass(clazz, GOOGLE_AUTH_EXCEPTION)) { + return googleAuth(node, below, evidence); + } + return null; + } + + /// Generic transport rules, applied only when no source-specific rule decided the chain. + @Nullable + static IngestionSourceErrorVerdict matchGeneric(Throwable node) { + return isUnreachable(node) ? unreachable(node, "") : null; + } + + /// Whether `throwable` means the endpoint never answered: DNS, connect, TLS or a socket timeout. + private static boolean isUnreachable(Throwable throwable) { + if (throwable instanceof BindException + || (throwable instanceof SocketException && messageOf(throwable).contains(OUR_SOCKETS_EXHAUSTED))) { + return false; + } + return throwable instanceof UnknownHostException || throwable instanceof SocketException + || throwable instanceof SocketTimeoutException || throwable instanceof HttpTimeoutException + || throwable instanceof SSLException; Review Comment: **Major:** outside Kafka, a TLS 1.2 peer alert rejecting our client certificate is classified SOURCE_READ, not AUTH. Only `kafkaTls` reads the alert name; this generic branch treats every `SSLException` as "unreachable". I confirmed this with a probe against this head: `RuntimeException("Error retrieving Avro schema", new SSLHandshakeException("Received fatal alert: bad_certificate"))` gives `SOURCE_READ code=SSLHandshakeException`. `certificate_required` and `access_denied` give the same result. The Kafka path maps those same alerts to AUTH. The PR's known-gap note only covers TLS 1.3, where no alert is available. Could the `TLS_ALERT` check move into the generic rule so mTLS to a schema registry, Pulsar or an HTTP source gets the same verdict as a Kafka broker? ########## 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) { Review Comment: **Minor / question:** a `none` decided on the main exception doesn't stop a suppressed exception from deciding the result. Probe: `RuntimeException("op", new CancellationException())` with a suppressed `RuntimeException("close", new SocketException("Socket closed"))` gives SOURCE_READ. With try-with-resources, a close failure after our own cancellation lands exactly there. The PR says cancellation and our clock are "not classified". Should a `none` on the main exception's chain also outrank suppressed chains, or is this left to callers through `setBoundary`? ########## 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: **Design question for the SPI owner (C3.8 / C3.24 / C8.1):** once released, this becomes a public `pinot-spi` contract with no consumer in this PR. That covers the process-wide static registries, the public setters on `IngestionSourceErrorEvidence`, and about 750 lines of SDK-specific matching that partly relies on message text. The built-in Kafka, AWS, Pulsar and gRPC rules always outrank plugin-registered rules, so any SDK upgrade that changes a message string means a change in `pinot-spi`. Would it be safer to land this together with part 2's call sites, or to keep the rules out of the public SPI (for example a runtime module, or rules registered by each plugin) until the API has a consumer? Not a blocker; it needs an owner decision. -- 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]
