suvodeep-pyne commented on code in PR #20067: URL: https://github.com/apache/pinot/pull/20067#discussion_r4398264706
########## 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: Updated the known-gaps note in the description to describe the TLS 1.2 case: `handshake_failure`, `Broken pipe` or `Connection reset` reads as SOURCE_READ. It also says TLS 1.3 usually gives AUTH but can fail first with `IOException: Error writing to server`, which stays unclassified. -- 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]
