Jackie-Jiang commented on code in PR #20067: URL: https://github.com/apache/pinot/pull/20067#discussion_r4416240742
########## pinot-spi/src/main/java/org/apache/pinot/spi/ingestion/error/IngestionSourceErrorRules.java: ########## @@ -0,0 +1,788 @@ +/** + * 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 can move a failure to another category or leave it unclassified, so an upgrade of these SDKs reruns those +/// fixtures against real clients where they can. +/// +/// 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, below, 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, below, 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) { + IngestionSourceErrorVerdict refused = tlsAlert(node, null); + if (refused != null) { + return refused; + } + return isUnreachable(node) ? unreachable(node, "") : null; + } + + /// A TLS peer's alert rejecting the certificate we presented, or denying us access, or null for any other failure. + @Nullable + private static IngestionSourceErrorVerdict tlsAlert(Throwable node, @Nullable String target) { + if (!(node instanceof SSLException)) { + return null; + } + Matcher alert = TLS_ALERT.matcher(messageOf(node)); + if (!alert.find()) { + return null; + } + String name = alert.group(1); + if (TLS_CERTIFICATE_REJECTED_ALERTS.contains(name)) { + return auth(IngestionSourceErrorSubReason.AUTHENTICATION, name, target); + } + return TLS_ACCESS_DENIED_ALERT.equals(name) ? auth(IngestionSourceErrorSubReason.AUTHORIZATION, name, target) + : 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; + } + + /// An endpoint that never answered: the source's, unless it was a metadata endpoint serving our own identity, named + /// by the network exception or by the client wrapping it. + private static IngestionSourceErrorVerdict unreachable(Throwable cause, String wrapperMessage) { + String code = cause.getClass().getSimpleName(); + return namesMetadataEndpoint(messageOf(cause)) || namesMetadataEndpoint(wrapperMessage) ? none(code) + : sourceRead(code, null); + } + + private static boolean namesMetadataEndpoint(String message) { + for (String endpoint : METADATA_ENDPOINTS) { + if (message.contains(endpoint)) { + return true; + } + } + return false; + } + + // --------------------------------------------------------------------------------------------------------------- + // AWS SDK v2: Kinesis, STS, KMS, Glue + + private static final String AWS_SERVICE_EXCEPTION = "software.amazon.awssdk.awscore.exception.AwsServiceException"; + private static final String AWS_CLIENT_EXCEPTION = "software.amazon.awssdk.core.exception.SdkClientException"; + private static final String AWS_STS_EXCEPTION = "software.amazon.awssdk.services.sts.model.StsException"; + private static final String AWS_ABORTED = "software.amazon.awssdk.core.exception.AbortedException"; + // The SDK's Apache HTTP clients, 4 and 5: the endpoint closed the connection without answering + private static final Set<String> APACHE_NO_HTTP_RESPONSE = Set.of("org.apache.http.NoHttpResponseException", + "org.apache.hc.core5.http.NoHttpResponseException"); + private static final Set<String> AWS_CALL_TIMEOUTS = Set.of( + "software.amazon.awssdk.core.exception.ApiCallTimeoutException", + "software.amazon.awssdk.core.exception.ApiCallAttemptTimeoutException"); + // Codes that reject a credential. Which credential decides the verdict, see `awsCredentialCode`. + private static final Set<String> AWS_CREDENTIAL_CODES = Set.of("InvalidAccessKeyId", "SignatureDoesNotMatch", + "InvalidSignatureException", "ExpiredToken", "ExpiredTokenException", "InvalidToken", "TokenRefreshRequired", + "InvalidClientTokenId", "IncompleteSignature", "MissingAuthenticationToken", "UnrecognizedClientException"); + private static final Set<String> AWS_AUTHORIZATION_CODES = Set.of("RegionDisabledException", + "GlueEncryptionException", "AccountProblem", "NotSignedUp"); + private static final Set<String> AWS_KMS_AUTHORIZATION_CODES = Set.of("KMSAccessDeniedException", + "KMSOptInRequired", "KMSKeyNotAccessibleFault"); + private static final Set<String> AWS_SOURCE_READ_CODES = Set.of("NoSuchKey", "NoSuchBucket", "InvalidObjectState", + "ObjectNotInActiveTierError", "PermanentRedirect", "AuthorizationHeaderMalformed", + "IllegalLocationConstraintException", "SlowDown", "ServiceUnavailable", "ServiceUnavailableException", + "InternalError", "InternalFailure", "Throttling", "ThrottlingException", "EntityNotFoundException", + "InternalServiceException", "AllAccessDisabled", "InvalidPayer", "ResourceNotFoundException", + "ResourceInUseException", "ProvisionedThroughputExceededException", "LimitExceededException"); + private static final Set<String> AWS_KMS_SOURCE_READ_CODES = Set.of("KMS.DisabledException", + "KMS.KMSInvalidStateException", "KMS.NotFoundException", "KMSDisabledException", "KMSInvalidStateException", + "KMSNotFoundException", "KMSThrottlingException"); + // Our clock, a request we built badly, our own web identity, or a Kinesis iterator our consumer let expire. + private static final Set<String> AWS_NONE_CODES = Set.of("RequestTimeTooSkewed", "RequestExpired", + "RequestInTheFuture", "ValidationError", "ValidationException", "MalformedPolicyDocument", + "PackedPolicyTooLarge", "InvalidIdentityToken", "IDPRejectedClaim", "IDPCommunicationError", + "InvalidInputException", "InvalidArgumentException", "ExpiredIteratorException", "ExpiredNextTokenException"); + // AWS reports our clock skew under a signature code; only the message tells it from a wrong secret. + private static final List<String> AWS_CLOCK_SKEW = List.of("Signature expired", "Signature not yet current"); + + @Nullable + private static IngestionSourceErrorVerdict awsService(Throwable node, + @Nullable IngestionSourceErrorEvidence evidence) { + boolean sts = hasClass(node.getClass(), AWS_STS_EXCEPTION); + Object details = invoke(node, "awsErrorDetails"); + Object code = details != null ? invoke(details, "errorCode") : null; + Object status = invoke(node, "statusCode"); + return awsCode(sts, code != null ? code.toString() : null, status instanceof Integer ? (Integer) status : 0, + messageOf(node), evidence); + } + + @Nullable + private static IngestionSourceErrorVerdict awsCode(boolean sts, @Nullable String code, int status, String message, + @Nullable IngestionSourceErrorEvidence evidence) { + String target = sts ? TARGET_STS : evidence != null ? evidence.getTarget() : null; + if (code == null || code.isEmpty()) { + // A response with no body (a HEAD): a deny, a missing object without list permission and our skew look alike. + if (status == 403) { + return auth(IngestionSourceErrorSubReason.MISSING_OR_DENIED, "403", target); + } + if (status == 404 || status == 400 || status == 429 || (status >= 300 && status < 400) || status >= 500) { + return sourceRead(String.valueOf(status), target); + } + return null; + } + if (code.equals("AccessDenied") || code.equals("AccessDeniedException")) { + if (message.contains("AssumeRoleWithWebIdentity")) { + // Our own web identity, not the table's role. + return none(code); + } + if (message.contains("ciphertext refers to a customer master key that does not exist")) { + return auth(IngestionSourceErrorSubReason.MISSING_OR_DENIED, code, TARGET_KMS); + } + return auth(IngestionSourceErrorSubReason.AUTHORIZATION, code, message.contains("kms:") ? TARGET_KMS : target); + } + if (AWS_AUTHORIZATION_CODES.contains(code)) { + return auth(IngestionSourceErrorSubReason.AUTHORIZATION, code, target); + } + if (AWS_KMS_AUTHORIZATION_CODES.contains(code)) { + return auth(IngestionSourceErrorSubReason.AUTHORIZATION, code, TARGET_KMS); + } + if (AWS_CREDENTIAL_CODES.contains(code)) { + for (String skew : AWS_CLOCK_SKEW) { + if (message.contains(skew)) { + return none(code); + } + } + return awsCredentialCode(sts, code, target, evidence); + } + if (AWS_KMS_SOURCE_READ_CODES.contains(code)) { + return sourceRead(code, TARGET_KMS); + } + if (AWS_SOURCE_READ_CODES.contains(code) || status == 429 || status >= 500) { + return sourceRead(code, target); + } + if (AWS_NONE_CODES.contains(code)) { + return none(code); + } + return null; + } + + /// A credential code is the table's bad credential only when the table's static keys were what got rejected: by + /// STS on the AssumeRole call when a role is set, or by the service when none is. Our ambient identity, and the + /// session credentials the SDK refreshes for an assumed role, are ours: the table owner's role and policy decisions + /// arrive as `AccessDenied` instead. With no recorded credential mode the rejected credential is unknown, so it + /// fails closed. + private static IngestionSourceErrorVerdict awsCredentialCode(boolean sts, String code, @Nullable String target, + @Nullable IngestionSourceErrorEvidence evidence) { + if (evidence == null || evidence.getCredentialMode() != IngestionSourceCredentialMode.STATIC) { + return none(code); + } + boolean baseKeysRejected = sts || !evidence.isRoleArnConfigured(); + return baseKeysRejected ? auth(IngestionSourceErrorSubReason.AUTHENTICATION, code, target) : none(code); + } + + /// No service exception sits beneath this node, or it would have decided first. + @Nullable + private static IngestionSourceErrorVerdict awsClient(Throwable node, List<Throwable> below) { + Class<?> clazz = node.getClass(); + String message = messageOf(node); + // The process's own credential chain ("Unable to load credentials from any of the providers ..."), one of its + // providers called directly after a first success ("Failed to load credentials from IMDS." or "... from metadata + // service."), or our cancellation. + if (message.startsWith("Unable to load credentials") || message.startsWith("Failed to load credentials from") + || hasClass(clazz, AWS_ABORTED)) { + return none(clazz.getSimpleName()); + } + if (hasAnyClass(clazz, AWS_CALL_TIMEOUTS)) { + return sourceRead(clazz.getSimpleName(), null); + } + for (Throwable cause : below) { + IngestionSourceErrorVerdict refused = tlsAlert(cause, null); + if (refused != null) { + return refused; + } + if (isUnreachable(cause)) { + return unreachable(cause, message); + } + if (hasAnyClass(cause.getClass(), APACHE_NO_HTTP_RESPONSE)) { + return sourceRead(cause.getClass().getSimpleName(), null); + } + } + // Includes the cause-less "Request attempt N failure" records the SDK attaches as suppressed exceptions. + return null; + } + + // --------------------------------------------------------------------------------------------------------------- + // Kafka brokers (kafka-clients 3.x and 4.x) + + private static final String KAFKA_ERRORS = "org.apache.kafka.common.errors."; + private static final String KAFKA_AUTHENTICATION = KAFKA_ERRORS + "AuthenticationException"; + private static final String KAFKA_AUTHORIZATION = KAFKA_ERRORS + "AuthorizationException"; + private static final String KAFKA_SASL_AUTHENTICATION = KAFKA_ERRORS + "SaslAuthenticationException"; + private static final String KAFKA_SSL_AUTHENTICATION = KAFKA_ERRORS + "SslAuthenticationException"; + private static final String KAFKA_CONFIG_EXCEPTION = "org.apache.kafka.common.config.ConfigException"; + private static final Set<String> KAFKA_BROKER_SOURCE_READ = Set.of(KAFKA_ERRORS + "TimeoutException", + KAFKA_ERRORS + "DisconnectException", KAFKA_ERRORS + "UnknownTopicOrPartitionException", + KAFKA_ERRORS + "OffsetOutOfRangeException", "org.apache.kafka.clients.consumer.OffsetOutOfRangeException"); + // The schema-registry serde maps a 429 to this class and drops the cause. + private static final String KAFKA_THROTTLING_QUOTA = KAFKA_ERRORS + "ThrottlingQuotaExceededException"; + private static final String SASL_EXCEPTION = "javax.security.sasl.SaslException"; + // A certificate we failed to validate: we rejected the broker. + private static final Set<String> CERTIFICATE_REJECTED = Set.of("java.security.cert.CertificateException", + "java.security.cert.CertPathValidatorException", "sun.security.validator.ValidatorException"); + private static final Pattern TLS_ALERT = Pattern.compile("Received fatal alert: ([a-z_]+)"); + // Alerts a peer sends when it rejects the certificate we presented. + private static final Set<String> TLS_CERTIFICATE_REJECTED_ALERTS = Set.of("bad_certificate", + "unsupported_certificate", "certificate_revoked", "certificate_expired", "certificate_unknown", "unknown_ca", + "certificate_required"); + private static final String TLS_ACCESS_DENIED_ALERT = "access_denied"; + + @Nullable + private static IngestionSourceErrorVerdict kafka(Throwable node, List<Throwable> below, + @Nullable IngestionSourceErrorEvidence evidence) { + Class<?> clazz = node.getClass(); + String code = clazz.getSimpleName(); + if (hasClass(clazz, KAFKA_AUTHORIZATION)) { + return auth(IngestionSourceErrorSubReason.AUTHORIZATION, code, TARGET_BROKER); + } + if (hasClass(clazz, KAFKA_AUTHENTICATION)) { + return kafkaAuthentication(node, below, evidence); + } + if (hasClass(clazz, KAFKA_THROTTLING_QUOTA)) { + return sourceRead(code, TARGET_SCHEMA_REGISTRY); + } + if (hasAnyClass(clazz, KAFKA_BROKER_SOURCE_READ)) { + // A broker's timeout carries no cause. The schema-registry serde reuses these classes over the network cause of + // its own request, so with a cause the endpoint is unknown here. + return sourceRead(code, below.isEmpty() ? TARGET_BROKER : null); + } + if (hasClass(clazz, KAFKA_CONFIG_EXCEPTION)) { + // Only an unresolvable bootstrap list is the source's; any other config error is ours. + return messageOf(node).startsWith("No resolvable bootstrap urls") + ? sourceRead(code, TARGET_BROKER) + : none(code); + } + return null; + } + + /// Kafka's authentication classes say less than they seem to. MSK IAM reports a policy denial as a SASL + /// authentication failure, a client-side failure to build the SASL token (our credential chain, our Kerberos setup) + /// arrives under the same class with a `SaslException` cause, a broker's connection throttle reuses it, and an SSL + /// failure covers both directions of a rejected certificate. + private static IngestionSourceErrorVerdict kafkaAuthentication(Throwable node, List<Throwable> below, + @Nullable IngestionSourceErrorEvidence evidence) { + Class<?> clazz = node.getClass(); + String code = clazz.getSimpleName(); + if (hasClass(clazz, KAFKA_SASL_AUTHENTICATION) && messageOf(node).contains("Too many connects")) { + return sourceRead("too_many_connects", TARGET_BROKER); + } + if (hasClass(clazz, KAFKA_SSL_AUTHENTICATION)) { + return kafkaTls(code, below); + } + for (Throwable cause : below) { + if (hasClass(cause.getClass(), SASL_EXCEPTION)) { + // A broker rejection carries no cause; this one failed on our side before the broker judged anything. + return none(code); + } + } + // Signing with a cloud identity (MSK IAM with the process's identity or an assumed role), a policy denial arrives + // as an authentication failure: the class can't tell which. + boolean cloudIdentity = evidence != null + && (evidence.getCredentialMode() == IngestionSourceCredentialMode.AMBIENT || evidence.isRoleArnConfigured()); + IngestionSourceErrorSubReason subReason = + cloudIdentity ? IngestionSourceErrorSubReason.UNSPECIFIED : IngestionSourceErrorSubReason.AUTHENTICATION; + return auth(subReason, code, TARGET_BROKER); + } + + private static IngestionSourceErrorVerdict kafkaTls(String code, List<Throwable> below) { + for (Throwable cause : below) { + if (hasAnyClass(cause.getClass(), CERTIFICATE_REJECTED)) { + // We could not verify the broker: the endpoint, not our credential. + return sourceRead(code, TARGET_BROKER); + } + } + for (Throwable cause : below) { + IngestionSourceErrorVerdict refused = tlsAlert(cause, TARGET_BROKER); + if (refused != null) { + return refused; + } + } + // Any other handshake failure: the TLS connection to the endpoint failed. + return sourceRead(code, TARGET_BROKER); + } + + // --------------------------------------------------------------------------------------------------------------- + // Confluent schema registry (kafka-schema-registry-client) + + private static final String CONFLUENT_REST_CLIENT_EXCEPTION = + "io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException"; + + /// The registry's HTTP status decides; its error code does not (an HTML 403 from a proxy reads as 50005). The serde + /// wraps this exception in the same Kafka authentication and authorization classes a broker uses, so it decides + /// first, as the deeper node, and names the registry as the target. + @Nullable + private static IngestionSourceErrorVerdict schemaRegistry(Throwable node) { + Object statusValue = invoke(node, "getStatus"); + if (!(statusValue instanceof Integer)) { + return null; + } + int status = (Integer) statusValue; + IngestionSourceErrorVerdict verdict = httpStatus(status, TARGET_SCHEMA_REGISTRY); + // Another 4xx: a request Pinot built badly. + return verdict != null ? verdict : none(String.valueOf(status)); + } + + // --------------------------------------------------------------------------------------------------------------- + // Logins: Kerberos (Krb5LoginModule) and the Kafka OAUTHBEARER login, which both stringify their evidence + + private static final String LOGIN_EXCEPTION = "javax.security.auth.login.LoginException"; + // kafka-clients' token retriever, stringified into the LoginException: "The response code 401 and error response + // {"invalid_client" - "..."} was encountered ..." for a refusal, "The unexpected response code 503 ..." (4.x) for a + // failure. + private static final Pattern OAUTH_RESPONSE_CODE = Pattern.compile("The (?:unexpected )?response code (\\d{3})"); + private static final Pattern OAUTH_ERROR = + Pattern.compile("(?:\"error\"\\s*:\\s*\"|error response \\{\")([a-z_]{1,64})\""); + // The 3.x token retriever's text when the endpoint failed or never answered. + private static final String OAUTH_TOKEN_UNAVAILABLE = "An internal error occurred while retrieving token"; + // 4.x dumps the retriever's stack trace into the message instead: a network failure names its class there. + private static final List<String> DUMPED_UNREACHABLE = List.of("java.net.ConnectException", + "java.net.UnknownHostException", "java.net.SocketTimeoutException", "java.net.NoRouteToHostException", + "java.net.http.HttpTimeoutException", "javax.net.ssl.SSLHandshakeException"); + private static final Set<String> OAUTH_AUTHENTICATION = Set.of("invalid_client", "invalid_grant", + "unsupported_grant_type"); + private static final Set<String> OAUTH_AUTHORIZATION = Set.of("unauthorized_client", "invalid_scope", + "access_denied"); + private static final Set<String> OAUTH_UNAVAILABLE = Set.of("server_error", "temporarily_unavailable"); + private static final String OAUTH_INVALID_REQUEST = "invalid_request"; + private static final List<String> KERBEROS_UNREACHABLE = List.of("Cannot contact any KDC", "Receive timed out"); + // Our clock, keytab or krb5 setup. + private static final List<String> KERBEROS_LOCAL_FAILURES = List.of("Clock skew too great", + "Unable to obtain password", "Cannot locate default realm", "Cannot locate KDC"); + // The KDC rejecting the principal: client not found (6), credentials revoked (18), pre-authentication failed (24). + private static final Pattern KERBEROS_REJECTED = Pattern.compile("\\((6|18|24)\\)"); + + @Nullable + private static IngestionSourceErrorVerdict login(Throwable node) { + String message = messageOf(node); + Matcher responseCode = OAUTH_RESPONSE_CODE.matcher(message); + if (responseCode.find()) { + Matcher error = OAUTH_ERROR.matcher(message); + return oauth(error.find() ? error.group(1) : null, Integer.parseInt(responseCode.group(1))); + } + if (message.contains(OAUTH_TOKEN_UNAVAILABLE)) { + // The token endpoint failed or never answered; the client drops which. + return sourceRead("token_unavailable", TARGET_IDP); + } + for (String unreachable : DUMPED_UNREACHABLE) { + if (message.contains(unreachable)) { + return sourceRead("token_unavailable", TARGET_IDP); + } + } + for (String unreachable : KERBEROS_UNREACHABLE) { + if (message.contains(unreachable)) { + return sourceRead("kdc_unreachable", TARGET_IDP); + } + } + for (String ours : KERBEROS_LOCAL_FAILURES) { + if (message.contains(ours)) { + return none(node.getClass().getSimpleName()); + } + } + Matcher rejected = KERBEROS_REJECTED.matcher(message); + if (rejected.find()) { + return auth(IngestionSourceErrorSubReason.AUTHENTICATION, "krb5_" + rejected.group(1), TARGET_IDP); + } + return null; + } + + /// An OAuth 2.0 token endpoint's `error` (RFC 6749 section 5.2), falling back to its status. + @Nullable + private static IngestionSourceErrorVerdict oauth(@Nullable String error, int status) { + if (error != null) { + String normalized = error.toLowerCase(Locale.ROOT); + if (OAUTH_AUTHENTICATION.contains(normalized)) { + return auth(IngestionSourceErrorSubReason.AUTHENTICATION, normalized, TARGET_IDP); + } + if (OAUTH_AUTHORIZATION.contains(normalized)) { + return auth(IngestionSourceErrorSubReason.AUTHORIZATION, normalized, TARGET_IDP); + } + if (OAUTH_UNAVAILABLE.contains(normalized)) { + return sourceRead(normalized, TARGET_IDP); + } + if (OAUTH_INVALID_REQUEST.equals(normalized)) { + // A malformed request: ours. + return none(normalized); + } + } + return httpStatus(status, TARGET_IDP); + } + + /// An HTTP status from a service with no error code of our own to read: 401 and 403 refuse us, 404, 408, 429 and + /// 5xx can't serve us, anything else decides nothing. + @Nullable + private static IngestionSourceErrorVerdict httpStatus(int status, String target) { + String code = String.valueOf(status); + if (status == 401) { + return auth(IngestionSourceErrorSubReason.AUTHENTICATION, code, target); + } + if (status == 403) { + return auth(IngestionSourceErrorSubReason.AUTHORIZATION, code, target); + } + if (status == 404 || status == 408 || status == 429 || status >= 500) { + return sourceRead(code, target); + } + return null; + } + + // --------------------------------------------------------------------------------------------------------------- + // Pulsar client (v4 and v5 APIs) and admin + + private static final String PULSAR_CLIENT_EXCEPTION = "org.apache.pulsar.client.api.PulsarClientException"; + private static final String PULSAR_V5_CLIENT_EXCEPTION = "org.apache.pulsar.client.api.v5.PulsarClientException"; + private static final String PULSAR_ADMIN_EXCEPTION = "org.apache.pulsar.client.admin.PulsarAdminException"; + private static final Set<String> PULSAR_SOURCE_READ = Set.of("TopicDoesNotExistException", "NotFoundException", + "SubscriptionNotFoundException", "LookupException", "ConnectException", "TimeoutException", + "BrokerMetadataException", "BrokerPersistenceException", "TopicTerminatedException", "NotAllowedException", + "ConsumerBusyException", "NotConnectedException"); + private static final Set<String> PULSAR_NONE = Set.of("InvalidServiceURL", "InvalidServiceURLException", + "InvalidConfigurationException", "InvalidTopicNameException", "UnsupportedAuthenticationException", + "AlreadyClosedException", "NotSupportedException", "CryptoException", "ChecksumException", + "GettingAuthenticationDataException"); + // The OAuth2 plugin's token errors: "Unable to obtain an access token: <description> (<error>)". + private static final String PULSAR_OAUTH_TOKEN = "Unable to obtain an access token"; + private static final Pattern PULSAR_OAUTH_ERROR = Pattern.compile("\\(([a-z_]{1,64})\\)\\s*$"); + + /// Pulsar's class is its code: the broker's error maps one to one onto a nested subclass. Every subclass of the + /// base class is matched on its own name and the names of its superclasses up to the base, so + /// `ServiceNotReadyException` and `TooManyRequestsException` read as the `LookupException` they extend. + @Nullable + private static IngestionSourceErrorVerdict pulsar(Throwable node) { + String message = messageOf(node); + for (Class<?> current = node.getClass(); current != null; current = current.getSuperclass()) { + String name = current.getName(); + if (name.equals(PULSAR_CLIENT_EXCEPTION) || name.equals(PULSAR_V5_CLIENT_EXCEPTION)) { + // The base class alone: TLS failures surface only as "Connection already closed"; anything else, such as an + // HTTP lookup failure, is left to a deeper or more specific exception. + return message.contains("Connection already closed") ? sourceRead("connection_closed", TARGET_BROKER) : null; + } + String simple = name.substring(name.lastIndexOf('$') + 1); + if (simple.equals("AuthenticationException")) { + return pulsarAuthentication(message); + } + if (simple.equals("AuthorizationException")) { + return auth(IngestionSourceErrorSubReason.AUTHORIZATION, simple, TARGET_BROKER); + } + if (simple.equals("TooManyRequestsException") && message.contains("Requests number out of config")) { + // The client's own lookup semaphore, not the broker's throttle. + return none(simple); + } + if (PULSAR_SOURCE_READ.contains(simple)) { + return sourceRead(simple, TARGET_BROKER); + } + if (PULSAR_NONE.contains(simple)) { + return none(simple); + } + } + return null; + } + + /// Pulsar's authentication class also covers an identity provider that failed or never answered, and a key file + /// we couldn't read. Only the message tells them apart. + private static IngestionSourceErrorVerdict pulsarAuthentication(String message) { + String code = "AuthenticationException"; + if (message.startsWith("Unable to read private key")) { + return none(code); + } + if (message.contains("Unable to retrieve OAuth 2.0 server metadata")) { + return sourceRead(code, TARGET_IDP); + } + if (message.contains(PULSAR_OAUTH_TOKEN)) { + Matcher error = PULSAR_OAUTH_ERROR.matcher(message); + if (error.find()) { + // An error we don't recognise decides nothing: it could be a refusal. + return oauth(error.group(1), 0); + } + // Without an OAuth error the token endpoint returned another status or never answered. + return sourceRead(code, TARGET_IDP); Review Comment: **[P2] Preserve Pulsar OAuth HTTP 403 as AUTH.** Pulsar 5.0.1's real `TokenClient` and `ClientCredentialsFlow` turn a token-endpoint HTTP 403 into `AuthenticationException("Unable to obtain an access token: Failed to perform HTTP request. res: 403")`. This fallback reports `SOURCE_READ`, hiding an explicit access denial. I reproduced this by passing controlled HTTP responses through both SDK classes: 401 with `invalid_client` gives AUTH, 403 gives SOURCE_READ, and 503 gives SOURCE_READ. Parse the preserved HTTP status before falling back to SOURCE_READ, and add a 403 case alongside the existing 503 fixture. ########## pinot-spi/src/main/java/org/apache/pinot/spi/ingestion/error/IngestionSourceErrorVerdict.java: ########## @@ -0,0 +1,125 @@ +/** + * 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.regex.Pattern; +import javax.annotation.Nullable; + + +/// A rule's decision on one throwable: a category, or [#none] when the throwable shows the failure is not the source's +/// verdict on our request (our clock, our identity, our config). +/// +/// The code and target are bounded tokens, safe for log lines: SDK codes are short identifiers, but a configured +/// endpoint can answer with anything. +public final class IngestionSourceErrorVerdict { + /// The target for a Kafka broker. + public static final String TARGET_BROKER = "broker"; + /// The target for a schema registry a decoder looks schemas up in. + public static final String TARGET_SCHEMA_REGISTRY = "schema_registry"; + /// The target for AWS STS, answering for the credential rather than the source. + public static final String TARGET_STS = "sts"; + /// The target for AWS KMS, answering for the key that encrypts the source's data. + public static final String TARGET_KMS = "kms"; + /// The target for an identity provider: an OAuth token endpoint or a Kerberos KDC. + public static final String TARGET_IDP = "idp"; + /// The target for an admin endpoint of the source, e.g. Pulsar's admin API. + public static final String TARGET_ADMIN = "admin"; + + private static final Pattern TOKEN_UNSAFE = Pattern.compile("[^A-Za-z0-9_.:/-]"); + private static final int MAX_TOKEN_LENGTH = 64; + + @Nullable + private final IngestionSourceErrorCategory _category; + @Nullable + private final IngestionSourceErrorSubReason _subReason; + @Nullable + private final String _code; + @Nullable + private final String _target; + + private IngestionSourceErrorVerdict(@Nullable IngestionSourceErrorCategory category, + @Nullable IngestionSourceErrorSubReason subReason, @Nullable String code, @Nullable String target) { + _category = category; + _subReason = subReason; + _code = token(code); + _target = token(target); + } + + /// The source refused us. `target` names the service that answered when it isn't the source itself. + public static IngestionSourceErrorVerdict auth(IngestionSourceErrorSubReason subReason, @Nullable String code, + @Nullable String target) { + return new IngestionSourceErrorVerdict(IngestionSourceErrorCategory.AUTH, subReason, code, target); + } + + /// The source couldn't serve us, or never answered. + public static IngestionSourceErrorVerdict sourceRead(@Nullable String code, @Nullable String target) { + return new IngestionSourceErrorVerdict(IngestionSourceErrorCategory.SOURCE_READ, null, code, target); + } + + /// Not the source's verdict on our request. It still decides the chain it was found in, so a wrapper above it can't + /// reinstate a category. + public static IngestionSourceErrorVerdict none(@Nullable String code) { + return new IngestionSourceErrorVerdict(null, null, code, null); + } + + /// The category, or null for [#none]. + @Nullable + public IngestionSourceErrorCategory getCategory() { + return _category; + } + + public boolean isNone() { + return _category == null; + } + + /// The finer reason behind an [IngestionSourceErrorCategory#AUTH] verdict; null otherwise. + @Nullable + public IngestionSourceErrorSubReason getSubReason() { + return _subReason; + } + + /// The source's own code that decided, e.g. `AccessDeniedException`, `403`, `PERMISSION_DENIED`. + @Nullable + public String getCode() { + return _code; + } + + /// The service that answered when it isn't the source itself, e.g. [#TARGET_STS] or [#TARGET_SCHEMA_REGISTRY]. + @Nullable + public String getTarget() { + return _target; + } + + /// `value` with every character outside `[A-Za-z0-9_.:/-]` replaced and its length capped, or null. + @Nullable + static String token(@Nullable String value) { + if (value == null) { + return null; + } + String token = TOKEN_UNSAFE.matcher(value).replaceAll("_"); + return token.length() > MAX_TOKEN_LENGTH ? token.substring(0, MAX_TOKEN_LENGTH) : token; Review Comment: **Nonblocking optimization: bound sanitization work as well as the result.** `token()` sanitizes the entire SDK-provided string, then retains only 64 characters. A sanitizer that stops after emitting 64 output characters would avoid scanning and allocating for the remainder of a large error code. Preserve the existing replacement semantics, including supplementary Unicode characters; simply truncating the UTF-16 input first can change the result. This is bounded-work cleanup, not a measured throughput improvement. ########## pinot-plugins/pinot-stream-ingestion/pinot-kinesis/src/test/java/org/apache/pinot/plugin/stream/kinesis/KinesisSourceErrorClassificationTest.java: ########## @@ -0,0 +1,341 @@ +/** + * 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.plugin.stream.kinesis; + +import java.io.IOException; +import java.io.InputStream; +import java.net.ConnectException; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.net.URI; +import java.net.UnknownHostException; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import javax.net.ssl.SSLHandshakeException; +import org.apache.pinot.spi.ingestion.error.IngestionSourceCredentialMode; +import org.apache.pinot.spi.ingestion.error.IngestionSourceErrorEvidence; +import org.apache.pinot.spi.ingestion.error.IngestionSourceErrorSubReason; +import org.testng.annotations.Test; +import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; +import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; +import software.amazon.awssdk.awscore.exception.AwsErrorDetails; +import software.amazon.awssdk.awscore.exception.AwsServiceException; +import software.amazon.awssdk.core.exception.ApiCallTimeoutException; +import software.amazon.awssdk.core.exception.SdkClientException; +import software.amazon.awssdk.http.apache.ApacheHttpClient; +import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.services.kinesis.KinesisClient; +import software.amazon.awssdk.services.kinesis.model.AccessDeniedException; +import software.amazon.awssdk.services.kinesis.model.ExpiredIteratorException; +import software.amazon.awssdk.services.kinesis.model.KinesisException; +import software.amazon.awssdk.services.kinesis.model.KmsAccessDeniedException; +import software.amazon.awssdk.services.kinesis.model.KmsDisabledException; +import software.amazon.awssdk.services.kinesis.model.KmsInvalidStateException; +import software.amazon.awssdk.services.kinesis.model.KmsNotFoundException; +import software.amazon.awssdk.services.kinesis.model.KmsOptInRequiredException; +import software.amazon.awssdk.services.kinesis.model.KmsThrottlingException; +import software.amazon.awssdk.services.kinesis.model.LimitExceededException; +import software.amazon.awssdk.services.kinesis.model.ProvisionedThroughputExceededException; +import software.amazon.awssdk.services.kinesis.model.ResourceInUseException; +import software.amazon.awssdk.services.kinesis.model.ResourceNotFoundException; +import software.amazon.awssdk.services.sts.StsClient; +import software.amazon.awssdk.services.sts.model.RegionDisabledException; +import software.amazon.awssdk.services.sts.model.StsException; + +import static org.apache.pinot.spi.ingestion.error.IngestionSourceErrorAssertions.assertAuth; +import static org.apache.pinot.spi.ingestion.error.IngestionSourceErrorAssertions.assertNotClassified; +import static org.apache.pinot.spi.ingestion.error.IngestionSourceErrorAssertions.assertSourceRead; +import static org.apache.pinot.spi.ingestion.error.IngestionSourceErrorVerdict.TARGET_KMS; +import static org.apache.pinot.spi.ingestion.error.IngestionSourceErrorVerdict.TARGET_STS; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertTrue; +import static org.testng.Assert.expectThrows; + + +/// Pins the classifier's AWS rules (the code table, the credential-mode logic and the message carve-outs) against the +/// exception hierarchy of the SDK pinot-kinesis ships. AWS answers almost everything with 400 or 403, so the error code +/// decides, and for a code that rejects a credential, whose credential it was. Each code below is a row of the rule +/// table that the batch classifier shares. +public class KinesisSourceErrorClassificationTest { + + private static <T extends Throwable> T withCredentials(T failure, IngestionSourceCredentialMode mode, + boolean roleArnConfigured) { + return IngestionSourceErrorEvidence.attach(failure, + new IngestionSourceErrorEvidence().setCredentialMode(mode).setRoleArnConfigured(roleArnConfigured)); + } + + private static AwsErrorDetails details(String service, String code) { + return AwsErrorDetails.builder().serviceName(service).errorCode(code).build(); + } + + private static KinesisException kinesis(String code, String message) { + return (KinesisException) KinesisException.builder().awsErrorDetails(details("Kinesis", code)).statusCode(400) + .message(message).build(); + } + + private static StsException sts(String code, int status, String message) { + return (StsException) StsException.builder().awsErrorDetails(details("Sts", code)).statusCode(status) + .message(message).build(); + } + + /// The consume loop sees the SDK's exception raw, under Pinot's own wrappers. + private static Throwable underPinot(Throwable failure) { + return new RuntimeException("Pinot wrapper", failure); + } + + @Test + public void permissionDenied() { + Throwable denied = AccessDeniedException.builder().awsErrorDetails(details("Kinesis", "AccessDeniedException")) + .statusCode(400).message("User: arn is not authorized to perform: kinesis:GetRecords on resource: stream") + .build(); + assertAuth(underPinot(withCredentials(denied, IngestionSourceCredentialMode.STATIC, false)), + IngestionSourceErrorSubReason.AUTHORIZATION, null); + // A denial of the process's own identity is still a permission someone must grant. + assertAuth(underPinot(withCredentials(denied, IngestionSourceCredentialMode.AMBIENT, false)), + IngestionSourceErrorSubReason.AUTHORIZATION, null); Review Comment: **Nonblocking test cleanup: use a fresh exception for each credential mode.** Both assertions reuse `denied`, but `IngestionSourceErrorEvidence.attach()` preserves evidence already attached. After the first assertion attaches STATIC, the purported AMBIENT case still carries STATIC. A factory or data provider that creates a new exception for each mode would actually exercise both cases. The expected category is AUTH for either mode, so this is a test-coverage correction rather than another production defect. ########## 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) { + REGISTERED_RULES.add(rule); + } + + public static void unregisterRule(IngestionSourceErrorRule rule) { + REGISTERED_RULES.remove(rule); + } + + /// Adds generic rules: ones that read the transport rather than a source's answer, such as a file missing under a + /// path the caller knows is remote. Like the built-in network rules, they apply to a chain only when no source rule, + /// built-in or registered, decided it, and after the built-in network rules on each exception. + public static void registerGenericRule(IngestionSourceErrorRule rule) { + REGISTERED_GENERIC_RULES.add(rule); + } + + public static void unregisterGenericRule(IngestionSourceErrorRule rule) { + REGISTERED_GENERIC_RULES.remove(rule); + } + + /// The decision of the registered rules on `node`. When they disagree, "not the source's verdict" beats + /// [IngestionSourceErrorCategory#AUTH], which beats [IngestionSourceErrorCategory#SOURCE_READ], and remaining ties + /// break on sub-reason, code and target, so registration order never matters. + @Nullable + private static IngestionSourceErrorVerdict decideRegistered(Set<IngestionSourceErrorRule> rules, Throwable node, + List<Throwable> below, @Nullable IngestionSourceErrorEvidence evidence) { + IngestionSourceErrorVerdict decided = null; + for (IngestionSourceErrorRule rule : rules) { + IngestionSourceErrorVerdict verdict; + try { + verdict = rule.decide(node, below, evidence); + } catch (Throwable t) { + // A rule that throws anything, even a checked exception it smuggles out, counts as no decision. + rethrowIfFatal(t); + continue; + } + if (verdict != null && (decided == null || PRECEDENCE.compare(verdict, decided) < 0)) { + decided = verdict; + } + } + return decided; + } + + private static final Comparator<IngestionSourceErrorVerdict> PRECEDENCE = + Comparator.comparingInt(IngestionSourceErrorClassifier::rank) + .thenComparing(IngestionSourceErrorVerdict::getSubReason, Comparator.nullsLast(Comparator.naturalOrder())) + .thenComparing(IngestionSourceErrorVerdict::getCode, Comparator.nullsLast(Comparator.naturalOrder())) + .thenComparing(IngestionSourceErrorVerdict::getTarget, Comparator.nullsLast(Comparator.naturalOrder())); + + private static int rank(IngestionSourceErrorVerdict verdict) { + return verdict.isNone() ? 0 : verdict.getCategory() == IngestionSourceErrorCategory.AUTH ? 1 : 2; + } + + private static final class Decision { + final IngestionSourceErrorVerdict _verdict; + final Throwable _node; + @Nullable + final IngestionSourceErrorEvidence _evidence; + + Decision(IngestionSourceErrorVerdict verdict, Throwable node, @Nullable IngestionSourceErrorEvidence evidence) { + _verdict = verdict; + _node = node; + _evidence = evidence; + } + } + + /// One chain of the tree: an exception and its causes, with the evidence each node carries and the evidence nearest + /// above it. + private static final class Chain { + final List<Throwable> _nodes = new ArrayList<>(); + final List<IngestionSourceErrorEvidence> _direct = new ArrayList<>(); + final List<IngestionSourceErrorEvidence> _nearest = new ArrayList<>(); + // Whether a boundary sits at or above each node + final List<Boolean> _withinBoundary = new ArrayList<>(); + } + + private static final class Walk { + final Set<Throwable> _heads = Collections.newSetFromMap(new IdentityHashMap<>()); + final List<Chain> _chains = new ArrayList<>(); + final List<Decision> _decisions = new ArrayList<>(); Review Comment: **Nonblocking cleanup: simplify traversal bookkeeping.** `_withinBoundary` is monotonic within each chain: false until the first boundary, then true for every remaining node. A first-boundary index can replace that parallel Boolean list. Separately, `Walk.decide()` can retain the winning decision as it visits chains, eliminating `_decisions` and the subsequent aggregation pass. Preserve first-AUTH/first-SOURCE_READ tie behavior and evaluation of all chains. Keep the discovery phase separate: a boundary discovered in another branch changes eligibility across the graph. ########## pinot-spi/src/main/java/org/apache/pinot/spi/ingestion/error/IngestionSourceErrorRules.java: ########## @@ -0,0 +1,788 @@ +/** + * 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 can move a failure to another category or leave it unclassified, so an upgrade of these SDKs reruns those +/// fixtures against real clients where they can. +/// +/// 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, below, 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, below, 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) { + IngestionSourceErrorVerdict refused = tlsAlert(node, null); + if (refused != null) { + return refused; + } + return isUnreachable(node) ? unreachable(node, "") : null; + } + + /// A TLS peer's alert rejecting the certificate we presented, or denying us access, or null for any other failure. + @Nullable + private static IngestionSourceErrorVerdict tlsAlert(Throwable node, @Nullable String target) { + if (!(node instanceof SSLException)) { + return null; + } + Matcher alert = TLS_ALERT.matcher(messageOf(node)); + if (!alert.find()) { + return null; + } + String name = alert.group(1); + if (TLS_CERTIFICATE_REJECTED_ALERTS.contains(name)) { + return auth(IngestionSourceErrorSubReason.AUTHENTICATION, name, target); + } + return TLS_ACCESS_DENIED_ALERT.equals(name) ? auth(IngestionSourceErrorSubReason.AUTHORIZATION, name, target) + : 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; + } + + /// An endpoint that never answered: the source's, unless it was a metadata endpoint serving our own identity, named + /// by the network exception or by the client wrapping it. + private static IngestionSourceErrorVerdict unreachable(Throwable cause, String wrapperMessage) { + String code = cause.getClass().getSimpleName(); + return namesMetadataEndpoint(messageOf(cause)) || namesMetadataEndpoint(wrapperMessage) ? none(code) + : sourceRead(code, null); + } + + private static boolean namesMetadataEndpoint(String message) { + for (String endpoint : METADATA_ENDPOINTS) { + if (message.contains(endpoint)) { + return true; + } + } + return false; + } + + // --------------------------------------------------------------------------------------------------------------- + // AWS SDK v2: Kinesis, STS, KMS, Glue + + private static final String AWS_SERVICE_EXCEPTION = "software.amazon.awssdk.awscore.exception.AwsServiceException"; + private static final String AWS_CLIENT_EXCEPTION = "software.amazon.awssdk.core.exception.SdkClientException"; + private static final String AWS_STS_EXCEPTION = "software.amazon.awssdk.services.sts.model.StsException"; + private static final String AWS_ABORTED = "software.amazon.awssdk.core.exception.AbortedException"; + // The SDK's Apache HTTP clients, 4 and 5: the endpoint closed the connection without answering + private static final Set<String> APACHE_NO_HTTP_RESPONSE = Set.of("org.apache.http.NoHttpResponseException", + "org.apache.hc.core5.http.NoHttpResponseException"); + private static final Set<String> AWS_CALL_TIMEOUTS = Set.of( + "software.amazon.awssdk.core.exception.ApiCallTimeoutException", + "software.amazon.awssdk.core.exception.ApiCallAttemptTimeoutException"); + // Codes that reject a credential. Which credential decides the verdict, see `awsCredentialCode`. + private static final Set<String> AWS_CREDENTIAL_CODES = Set.of("InvalidAccessKeyId", "SignatureDoesNotMatch", + "InvalidSignatureException", "ExpiredToken", "ExpiredTokenException", "InvalidToken", "TokenRefreshRequired", + "InvalidClientTokenId", "IncompleteSignature", "MissingAuthenticationToken", "UnrecognizedClientException"); + private static final Set<String> AWS_AUTHORIZATION_CODES = Set.of("RegionDisabledException", + "GlueEncryptionException", "AccountProblem", "NotSignedUp"); + private static final Set<String> AWS_KMS_AUTHORIZATION_CODES = Set.of("KMSAccessDeniedException", + "KMSOptInRequired", "KMSKeyNotAccessibleFault"); + private static final Set<String> AWS_SOURCE_READ_CODES = Set.of("NoSuchKey", "NoSuchBucket", "InvalidObjectState", + "ObjectNotInActiveTierError", "PermanentRedirect", "AuthorizationHeaderMalformed", + "IllegalLocationConstraintException", "SlowDown", "ServiceUnavailable", "ServiceUnavailableException", + "InternalError", "InternalFailure", "Throttling", "ThrottlingException", "EntityNotFoundException", + "InternalServiceException", "AllAccessDisabled", "InvalidPayer", "ResourceNotFoundException", + "ResourceInUseException", "ProvisionedThroughputExceededException", "LimitExceededException"); + private static final Set<String> AWS_KMS_SOURCE_READ_CODES = Set.of("KMS.DisabledException", + "KMS.KMSInvalidStateException", "KMS.NotFoundException", "KMSDisabledException", "KMSInvalidStateException", + "KMSNotFoundException", "KMSThrottlingException"); + // Our clock, a request we built badly, our own web identity, or a Kinesis iterator our consumer let expire. + private static final Set<String> AWS_NONE_CODES = Set.of("RequestTimeTooSkewed", "RequestExpired", + "RequestInTheFuture", "ValidationError", "ValidationException", "MalformedPolicyDocument", + "PackedPolicyTooLarge", "InvalidIdentityToken", "IDPRejectedClaim", "IDPCommunicationError", + "InvalidInputException", "InvalidArgumentException", "ExpiredIteratorException", "ExpiredNextTokenException"); + // AWS reports our clock skew under a signature code; only the message tells it from a wrong secret. + private static final List<String> AWS_CLOCK_SKEW = List.of("Signature expired", "Signature not yet current"); + + @Nullable + private static IngestionSourceErrorVerdict awsService(Throwable node, + @Nullable IngestionSourceErrorEvidence evidence) { + boolean sts = hasClass(node.getClass(), AWS_STS_EXCEPTION); + Object details = invoke(node, "awsErrorDetails"); + Object code = details != null ? invoke(details, "errorCode") : null; + Object status = invoke(node, "statusCode"); + return awsCode(sts, code != null ? code.toString() : null, status instanceof Integer ? (Integer) status : 0, + messageOf(node), evidence); + } + + @Nullable + private static IngestionSourceErrorVerdict awsCode(boolean sts, @Nullable String code, int status, String message, + @Nullable IngestionSourceErrorEvidence evidence) { + String target = sts ? TARGET_STS : evidence != null ? evidence.getTarget() : null; + if (code == null || code.isEmpty()) { + // A response with no body (a HEAD): a deny, a missing object without list permission and our skew look alike. + if (status == 403) { + return auth(IngestionSourceErrorSubReason.MISSING_OR_DENIED, "403", target); + } + if (status == 404 || status == 400 || status == 429 || (status >= 300 && status < 400) || status >= 500) { + return sourceRead(String.valueOf(status), target); + } + return null; + } + if (code.equals("AccessDenied") || code.equals("AccessDeniedException")) { + if (message.contains("AssumeRoleWithWebIdentity")) { + // Our own web identity, not the table's role. + return none(code); + } + if (message.contains("ciphertext refers to a customer master key that does not exist")) { + return auth(IngestionSourceErrorSubReason.MISSING_OR_DENIED, code, TARGET_KMS); + } + return auth(IngestionSourceErrorSubReason.AUTHORIZATION, code, message.contains("kms:") ? TARGET_KMS : target); + } + if (AWS_AUTHORIZATION_CODES.contains(code)) { + return auth(IngestionSourceErrorSubReason.AUTHORIZATION, code, target); + } + if (AWS_KMS_AUTHORIZATION_CODES.contains(code)) { + return auth(IngestionSourceErrorSubReason.AUTHORIZATION, code, TARGET_KMS); + } + if (AWS_CREDENTIAL_CODES.contains(code)) { + for (String skew : AWS_CLOCK_SKEW) { + if (message.contains(skew)) { + return none(code); + } + } + return awsCredentialCode(sts, code, target, evidence); + } + if (AWS_KMS_SOURCE_READ_CODES.contains(code)) { + return sourceRead(code, TARGET_KMS); + } + if (AWS_SOURCE_READ_CODES.contains(code) || status == 429 || status >= 500) { + return sourceRead(code, target); + } + if (AWS_NONE_CODES.contains(code)) { + return none(code); + } + return null; + } + + /// A credential code is the table's bad credential only when the table's static keys were what got rejected: by + /// STS on the AssumeRole call when a role is set, or by the service when none is. Our ambient identity, and the + /// session credentials the SDK refreshes for an assumed role, are ours: the table owner's role and policy decisions + /// arrive as `AccessDenied` instead. With no recorded credential mode the rejected credential is unknown, so it + /// fails closed. + private static IngestionSourceErrorVerdict awsCredentialCode(boolean sts, String code, @Nullable String target, + @Nullable IngestionSourceErrorEvidence evidence) { + if (evidence == null || evidence.getCredentialMode() != IngestionSourceCredentialMode.STATIC) { + return none(code); + } + boolean baseKeysRejected = sts || !evidence.isRoleArnConfigured(); + return baseKeysRejected ? auth(IngestionSourceErrorSubReason.AUTHENTICATION, code, target) : none(code); + } + + /// No service exception sits beneath this node, or it would have decided first. + @Nullable + private static IngestionSourceErrorVerdict awsClient(Throwable node, List<Throwable> below) { + Class<?> clazz = node.getClass(); + String message = messageOf(node); + // The process's own credential chain ("Unable to load credentials from any of the providers ..."), one of its + // providers called directly after a first success ("Failed to load credentials from IMDS." or "... from metadata + // service."), or our cancellation. + if (message.startsWith("Unable to load credentials") || message.startsWith("Failed to load credentials from") + || hasClass(clazz, AWS_ABORTED)) { + return none(clazz.getSimpleName()); + } + if (hasAnyClass(clazz, AWS_CALL_TIMEOUTS)) { + return sourceRead(clazz.getSimpleName(), null); + } + for (Throwable cause : below) { + IngestionSourceErrorVerdict refused = tlsAlert(cause, null); + if (refused != null) { + return refused; + } + if (isUnreachable(cause)) { + return unreachable(cause, message); + } + if (hasAnyClass(cause.getClass(), APACHE_NO_HTTP_RESPONSE)) { + return sourceRead(cause.getClass().getSimpleName(), null); + } + } + // Includes the cause-less "Request attempt N failure" records the SDK attaches as suppressed exceptions. + return null; + } + + // --------------------------------------------------------------------------------------------------------------- + // Kafka brokers (kafka-clients 3.x and 4.x) + + private static final String KAFKA_ERRORS = "org.apache.kafka.common.errors."; + private static final String KAFKA_AUTHENTICATION = KAFKA_ERRORS + "AuthenticationException"; + private static final String KAFKA_AUTHORIZATION = KAFKA_ERRORS + "AuthorizationException"; + private static final String KAFKA_SASL_AUTHENTICATION = KAFKA_ERRORS + "SaslAuthenticationException"; + private static final String KAFKA_SSL_AUTHENTICATION = KAFKA_ERRORS + "SslAuthenticationException"; + private static final String KAFKA_CONFIG_EXCEPTION = "org.apache.kafka.common.config.ConfigException"; + private static final Set<String> KAFKA_BROKER_SOURCE_READ = Set.of(KAFKA_ERRORS + "TimeoutException", + KAFKA_ERRORS + "DisconnectException", KAFKA_ERRORS + "UnknownTopicOrPartitionException", + KAFKA_ERRORS + "OffsetOutOfRangeException", "org.apache.kafka.clients.consumer.OffsetOutOfRangeException"); + // The schema-registry serde maps a 429 to this class and drops the cause. + private static final String KAFKA_THROTTLING_QUOTA = KAFKA_ERRORS + "ThrottlingQuotaExceededException"; + private static final String SASL_EXCEPTION = "javax.security.sasl.SaslException"; + // A certificate we failed to validate: we rejected the broker. + private static final Set<String> CERTIFICATE_REJECTED = Set.of("java.security.cert.CertificateException", + "java.security.cert.CertPathValidatorException", "sun.security.validator.ValidatorException"); + private static final Pattern TLS_ALERT = Pattern.compile("Received fatal alert: ([a-z_]+)"); + // Alerts a peer sends when it rejects the certificate we presented. + private static final Set<String> TLS_CERTIFICATE_REJECTED_ALERTS = Set.of("bad_certificate", + "unsupported_certificate", "certificate_revoked", "certificate_expired", "certificate_unknown", "unknown_ca", + "certificate_required"); + private static final String TLS_ACCESS_DENIED_ALERT = "access_denied"; + + @Nullable + private static IngestionSourceErrorVerdict kafka(Throwable node, List<Throwable> below, + @Nullable IngestionSourceErrorEvidence evidence) { + Class<?> clazz = node.getClass(); + String code = clazz.getSimpleName(); + if (hasClass(clazz, KAFKA_AUTHORIZATION)) { + return auth(IngestionSourceErrorSubReason.AUTHORIZATION, code, TARGET_BROKER); + } + if (hasClass(clazz, KAFKA_AUTHENTICATION)) { + return kafkaAuthentication(node, below, evidence); + } + if (hasClass(clazz, KAFKA_THROTTLING_QUOTA)) { + return sourceRead(code, TARGET_SCHEMA_REGISTRY); + } + if (hasAnyClass(clazz, KAFKA_BROKER_SOURCE_READ)) { + // A broker's timeout carries no cause. The schema-registry serde reuses these classes over the network cause of + // its own request, so with a cause the endpoint is unknown here. + return sourceRead(code, below.isEmpty() ? TARGET_BROKER : null); + } + if (hasClass(clazz, KAFKA_CONFIG_EXCEPTION)) { + // Only an unresolvable bootstrap list is the source's; any other config error is ours. + return messageOf(node).startsWith("No resolvable bootstrap urls") + ? sourceRead(code, TARGET_BROKER) + : none(code); + } + return null; + } + + /// Kafka's authentication classes say less than they seem to. MSK IAM reports a policy denial as a SASL + /// authentication failure, a client-side failure to build the SASL token (our credential chain, our Kerberos setup) + /// arrives under the same class with a `SaslException` cause, a broker's connection throttle reuses it, and an SSL + /// failure covers both directions of a rejected certificate. + private static IngestionSourceErrorVerdict kafkaAuthentication(Throwable node, List<Throwable> below, + @Nullable IngestionSourceErrorEvidence evidence) { + Class<?> clazz = node.getClass(); + String code = clazz.getSimpleName(); + if (hasClass(clazz, KAFKA_SASL_AUTHENTICATION) && messageOf(node).contains("Too many connects")) { + return sourceRead("too_many_connects", TARGET_BROKER); + } + if (hasClass(clazz, KAFKA_SSL_AUTHENTICATION)) { + return kafkaTls(code, below); + } + for (Throwable cause : below) { + if (hasClass(cause.getClass(), SASL_EXCEPTION)) { + // A broker rejection carries no cause; this one failed on our side before the broker judged anything. + return none(code); + } + } + // Signing with a cloud identity (MSK IAM with the process's identity or an assumed role), a policy denial arrives + // as an authentication failure: the class can't tell which. + boolean cloudIdentity = evidence != null + && (evidence.getCredentialMode() == IngestionSourceCredentialMode.AMBIENT || evidence.isRoleArnConfigured()); + IngestionSourceErrorSubReason subReason = + cloudIdentity ? IngestionSourceErrorSubReason.UNSPECIFIED : IngestionSourceErrorSubReason.AUTHENTICATION; + return auth(subReason, code, TARGET_BROKER); + } + + private static IngestionSourceErrorVerdict kafkaTls(String code, List<Throwable> below) { + for (Throwable cause : below) { + if (hasAnyClass(cause.getClass(), CERTIFICATE_REJECTED)) { + // We could not verify the broker: the endpoint, not our credential. + return sourceRead(code, TARGET_BROKER); + } + } + for (Throwable cause : below) { + IngestionSourceErrorVerdict refused = tlsAlert(cause, TARGET_BROKER); + if (refused != null) { + return refused; + } + } + // Any other handshake failure: the TLS connection to the endpoint failed. + return sourceRead(code, TARGET_BROKER); + } + + // --------------------------------------------------------------------------------------------------------------- + // Confluent schema registry (kafka-schema-registry-client) + + private static final String CONFLUENT_REST_CLIENT_EXCEPTION = + "io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException"; + + /// The registry's HTTP status decides; its error code does not (an HTML 403 from a proxy reads as 50005). The serde + /// wraps this exception in the same Kafka authentication and authorization classes a broker uses, so it decides + /// first, as the deeper node, and names the registry as the target. + @Nullable + private static IngestionSourceErrorVerdict schemaRegistry(Throwable node) { + Object statusValue = invoke(node, "getStatus"); + if (!(statusValue instanceof Integer)) { + return null; + } + int status = (Integer) statusValue; + IngestionSourceErrorVerdict verdict = httpStatus(status, TARGET_SCHEMA_REGISTRY); + // Another 4xx: a request Pinot built badly. + return verdict != null ? verdict : none(String.valueOf(status)); + } + + // --------------------------------------------------------------------------------------------------------------- + // Logins: Kerberos (Krb5LoginModule) and the Kafka OAUTHBEARER login, which both stringify their evidence + + private static final String LOGIN_EXCEPTION = "javax.security.auth.login.LoginException"; + // kafka-clients' token retriever, stringified into the LoginException: "The response code 401 and error response + // {"invalid_client" - "..."} was encountered ..." for a refusal, "The unexpected response code 503 ..." (4.x) for a + // failure. + private static final Pattern OAUTH_RESPONSE_CODE = Pattern.compile("The (?:unexpected )?response code (\\d{3})"); + private static final Pattern OAUTH_ERROR = + Pattern.compile("(?:\"error\"\\s*:\\s*\"|error response \\{\")([a-z_]{1,64})\""); + // The 3.x token retriever's text when the endpoint failed or never answered. + private static final String OAUTH_TOKEN_UNAVAILABLE = "An internal error occurred while retrieving token"; + // 4.x dumps the retriever's stack trace into the message instead: a network failure names its class there. + private static final List<String> DUMPED_UNREACHABLE = List.of("java.net.ConnectException", + "java.net.UnknownHostException", "java.net.SocketTimeoutException", "java.net.NoRouteToHostException", + "java.net.http.HttpTimeoutException", "javax.net.ssl.SSLHandshakeException"); + private static final Set<String> OAUTH_AUTHENTICATION = Set.of("invalid_client", "invalid_grant", + "unsupported_grant_type"); + private static final Set<String> OAUTH_AUTHORIZATION = Set.of("unauthorized_client", "invalid_scope", + "access_denied"); + private static final Set<String> OAUTH_UNAVAILABLE = Set.of("server_error", "temporarily_unavailable"); + private static final String OAUTH_INVALID_REQUEST = "invalid_request"; + private static final List<String> KERBEROS_UNREACHABLE = List.of("Cannot contact any KDC", "Receive timed out"); + // Our clock, keytab or krb5 setup. + private static final List<String> KERBEROS_LOCAL_FAILURES = List.of("Clock skew too great", + "Unable to obtain password", "Cannot locate default realm", "Cannot locate KDC"); + // The KDC rejecting the principal: client not found (6), credentials revoked (18), pre-authentication failed (24). + private static final Pattern KERBEROS_REJECTED = Pattern.compile("\\((6|18|24)\\)"); + + @Nullable + private static IngestionSourceErrorVerdict login(Throwable node) { + String message = messageOf(node); + Matcher responseCode = OAUTH_RESPONSE_CODE.matcher(message); + if (responseCode.find()) { + Matcher error = OAUTH_ERROR.matcher(message); + return oauth(error.find() ? error.group(1) : null, Integer.parseInt(responseCode.group(1))); + } + if (message.contains(OAUTH_TOKEN_UNAVAILABLE)) { + // The token endpoint failed or never answered; the client drops which. + return sourceRead("token_unavailable", TARGET_IDP); + } + for (String unreachable : DUMPED_UNREACHABLE) { + if (message.contains(unreachable)) { + return sourceRead("token_unavailable", TARGET_IDP); + } + } + for (String unreachable : KERBEROS_UNREACHABLE) { + if (message.contains(unreachable)) { + return sourceRead("kdc_unreachable", TARGET_IDP); + } + } + for (String ours : KERBEROS_LOCAL_FAILURES) { + if (message.contains(ours)) { + return none(node.getClass().getSimpleName()); + } + } + Matcher rejected = KERBEROS_REJECTED.matcher(message); + if (rejected.find()) { + return auth(IngestionSourceErrorSubReason.AUTHENTICATION, "krb5_" + rejected.group(1), TARGET_IDP); + } + return null; + } + + /// An OAuth 2.0 token endpoint's `error` (RFC 6749 section 5.2), falling back to its status. + @Nullable + private static IngestionSourceErrorVerdict oauth(@Nullable String error, int status) { + if (error != null) { + String normalized = error.toLowerCase(Locale.ROOT); + if (OAUTH_AUTHENTICATION.contains(normalized)) { + return auth(IngestionSourceErrorSubReason.AUTHENTICATION, normalized, TARGET_IDP); + } + if (OAUTH_AUTHORIZATION.contains(normalized)) { + return auth(IngestionSourceErrorSubReason.AUTHORIZATION, normalized, TARGET_IDP); + } + if (OAUTH_UNAVAILABLE.contains(normalized)) { + return sourceRead(normalized, TARGET_IDP); + } + if (OAUTH_INVALID_REQUEST.equals(normalized)) { + // A malformed request: ours. + return none(normalized); + } + } + return httpStatus(status, TARGET_IDP); + } + + /// An HTTP status from a service with no error code of our own to read: 401 and 403 refuse us, 404, 408, 429 and + /// 5xx can't serve us, anything else decides nothing. + @Nullable + private static IngestionSourceErrorVerdict httpStatus(int status, String target) { + String code = String.valueOf(status); + if (status == 401) { + return auth(IngestionSourceErrorSubReason.AUTHENTICATION, code, target); + } + if (status == 403) { + return auth(IngestionSourceErrorSubReason.AUTHORIZATION, code, target); + } + if (status == 404 || status == 408 || status == 429 || status >= 500) { + return sourceRead(code, target); + } + return null; + } + + // --------------------------------------------------------------------------------------------------------------- + // Pulsar client (v4 and v5 APIs) and admin + + private static final String PULSAR_CLIENT_EXCEPTION = "org.apache.pulsar.client.api.PulsarClientException"; + private static final String PULSAR_V5_CLIENT_EXCEPTION = "org.apache.pulsar.client.api.v5.PulsarClientException"; + private static final String PULSAR_ADMIN_EXCEPTION = "org.apache.pulsar.client.admin.PulsarAdminException"; + private static final Set<String> PULSAR_SOURCE_READ = Set.of("TopicDoesNotExistException", "NotFoundException", + "SubscriptionNotFoundException", "LookupException", "ConnectException", "TimeoutException", + "BrokerMetadataException", "BrokerPersistenceException", "TopicTerminatedException", "NotAllowedException", + "ConsumerBusyException", "NotConnectedException"); + private static final Set<String> PULSAR_NONE = Set.of("InvalidServiceURL", "InvalidServiceURLException", + "InvalidConfigurationException", "InvalidTopicNameException", "UnsupportedAuthenticationException", + "AlreadyClosedException", "NotSupportedException", "CryptoException", "ChecksumException", + "GettingAuthenticationDataException"); + // The OAuth2 plugin's token errors: "Unable to obtain an access token: <description> (<error>)". + private static final String PULSAR_OAUTH_TOKEN = "Unable to obtain an access token"; + private static final Pattern PULSAR_OAUTH_ERROR = Pattern.compile("\\(([a-z_]{1,64})\\)\\s*$"); + + /// Pulsar's class is its code: the broker's error maps one to one onto a nested subclass. Every subclass of the + /// base class is matched on its own name and the names of its superclasses up to the base, so + /// `ServiceNotReadyException` and `TooManyRequestsException` read as the `LookupException` they extend. + @Nullable + private static IngestionSourceErrorVerdict pulsar(Throwable node) { + String message = messageOf(node); + for (Class<?> current = node.getClass(); current != null; current = current.getSuperclass()) { + String name = current.getName(); + if (name.equals(PULSAR_CLIENT_EXCEPTION) || name.equals(PULSAR_V5_CLIENT_EXCEPTION)) { + // The base class alone: TLS failures surface only as "Connection already closed"; anything else, such as an + // HTTP lookup failure, is left to a deeper or more specific exception. + return message.contains("Connection already closed") ? sourceRead("connection_closed", TARGET_BROKER) : null; + } + String simple = name.substring(name.lastIndexOf('$') + 1); + if (simple.equals("AuthenticationException")) { + return pulsarAuthentication(message); + } + if (simple.equals("AuthorizationException")) { + return auth(IngestionSourceErrorSubReason.AUTHORIZATION, simple, TARGET_BROKER); + } + if (simple.equals("TooManyRequestsException") && message.contains("Requests number out of config")) { + // The client's own lookup semaphore, not the broker's throttle. + return none(simple); + } + if (PULSAR_SOURCE_READ.contains(simple)) { + return sourceRead(simple, TARGET_BROKER); + } + if (PULSAR_NONE.contains(simple)) { + return none(simple); + } + } + return null; + } + + /// Pulsar's authentication class also covers an identity provider that failed or never answered, and a key file + /// we couldn't read. Only the message tells them apart. + private static IngestionSourceErrorVerdict pulsarAuthentication(String message) { + String code = "AuthenticationException"; + if (message.startsWith("Unable to read private key")) { + return none(code); + } + if (message.contains("Unable to retrieve OAuth 2.0 server metadata")) { + return sourceRead(code, TARGET_IDP); + } + if (message.contains(PULSAR_OAUTH_TOKEN)) { + Matcher error = PULSAR_OAUTH_ERROR.matcher(message); + if (error.find()) { + // An error we don't recognise decides nothing: it could be a refusal. + return oauth(error.group(1), 0); + } + // Without an OAuth error the token endpoint returned another status or never answered. + return sourceRead(code, TARGET_IDP); + } + // The broker rejected the token: it gives one generic message for every reason. + return auth(IngestionSourceErrorSubReason.AUTHENTICATION, code, TARGET_BROKER); + } + + /// The admin client maps both 401 and 403 to `NotAuthorizedException`: only the status tells them apart. The base + /// class without an HTTP error is a client-side failure that defaults its status to 500, so its status means nothing. + @Nullable + private static IngestionSourceErrorVerdict pulsarAdmin(Throwable node) { + String simple = node.getClass().getSimpleName(); + if (simple.equals("ConnectException") || simple.equals("TimeoutException")) { + return sourceRead(simple, TARGET_ADMIN); + } + if (simple.equals("GettingAuthenticationDataException")) { + return none(simple); + } + if (node.getClass().getName().equals(PULSAR_ADMIN_EXCEPTION) && invoke(node, "getHttpError") == null) { + return null; + } + Object status = invoke(node, "getStatusCode"); + return status instanceof Integer ? httpStatus((Integer) status, TARGET_ADMIN) : null; + } + + // --------------------------------------------------------------------------------------------------------------- + // gRPC and gax (Google Cloud clients), and google-auth + + private static final String GAX_API_EXCEPTION = "com.google.api.gax.rpc.ApiException"; + private static final String GRPC_STATUS_RUNTIME_EXCEPTION = "io.grpc.StatusRuntimeException"; + private static final String GRPC_STATUS_EXCEPTION = "io.grpc.StatusException"; + private static final String GOOGLE_AUTH_EXCEPTION = "com.google.auth.oauth2.GoogleAuthException"; + private static final Set<String> GRPC_SOURCE_READ = Set.of("NOT_FOUND", "FAILED_PRECONDITION", + "RESOURCE_EXHAUSTED", "UNAVAILABLE", "DEADLINE_EXCEEDED", "INTERNAL"); + private static final Set<String> GRPC_NONE = Set.of("CANCELLED", "INVALID_ARGUMENT", "ALREADY_EXISTS"); + // Google Cloud's deliberate conflation of a missing project with a denied one. + private static final String GOOGLE_MISSING_OR_DENIED = "not found or user does not have access"; + // "Token must be a short-lived token ... Check your iat and exp values": our clock. + private static final String GOOGLE_CLOCK_SKEW = "short-lived token"; + + @Nullable + private static IngestionSourceErrorVerdict grpc(@Nullable Object codeValue, Throwable node, List<Throwable> below, + @Nullable IngestionSourceErrorEvidence evidence) { + if (!(codeValue instanceof Enum)) { + return null; + } + // Both gax's StatusCode.Code and gRPC's Status.Code are enums: read the constant's name, never a toString(). + String code = ((Enum<?>) codeValue).name(); + switch (code) { + case "UNAUTHENTICATED": + // A rejected credential is the table owner's only when it was the table's: under the process's own + // identity, grpc-auth reports our failed token refresh this way. + return evidence != null && evidence.getCredentialMode() == IngestionSourceCredentialMode.STATIC + ? auth(IngestionSourceErrorSubReason.AUTHENTICATION, code, null) : none(code); + case "PERMISSION_DENIED": + return auth(IngestionSourceErrorSubReason.AUTHORIZATION, code, null); + case "NOT_FOUND": + if (messageOf(node).contains(GOOGLE_MISSING_OR_DENIED)) { + return auth(IngestionSourceErrorSubReason.MISSING_OR_DENIED, code, null); + } + return sourceRead(code, null); + case "UNAVAILABLE": + // The transport failed, unless the peer rejected the certificate we presented + for (Throwable cause : below) { + IngestionSourceErrorVerdict refused = tlsAlert(cause, null); + if (refused != null) { + return refused; + } + } + return sourceRead(code, null); + default: + if (GRPC_SOURCE_READ.contains(code)) { + return sourceRead(code, null); + } + return GRPC_NONE.contains(code) ? none(code) : null; + } + } + + /// A token refresh that failed. Under the process's own identity the token is ours to get; our clock skew is ours + /// whoever's key it is. Otherwise the status the transport built above it decides. + @Nullable + private static IngestionSourceErrorVerdict googleAuth(Throwable node, List<Throwable> below, + @Nullable IngestionSourceErrorEvidence evidence) { + String code = node.getClass().getSimpleName(); + if (evidence != null && evidence.getCredentialMode() == IngestionSourceCredentialMode.AMBIENT) { + return none(code); + } + if (messageOf(node).contains(GOOGLE_CLOCK_SKEW)) { + return none(code); + } + for (Throwable cause : below) { + Object content = invoke(cause, "getContent"); + String body = content != null ? content.toString() : messageOf(cause); + if (body.contains(GOOGLE_CLOCK_SKEW)) { + return none(code); + } + } + return null; + } + + // --------------------------------------------------------------------------------------------------------------- + // Shorthands + + private static IngestionSourceErrorVerdict auth(IngestionSourceErrorSubReason subReason, @Nullable String code, + @Nullable String target) { + return IngestionSourceErrorVerdict.auth(subReason, code, target); + } + + private static IngestionSourceErrorVerdict sourceRead(@Nullable String code, @Nullable String target) { + return IngestionSourceErrorVerdict.sourceRead(code, target); + } + + private static IngestionSourceErrorVerdict none(@Nullable String code) { + return IngestionSourceErrorVerdict.none(code); Review Comment: **Nonblocking cleanup: replace forwarding declarations with static imports.** These verdict factories add no behavior beyond delegation. Static imports from `IngestionSourceErrorVerdict` would preserve the concise call sites while removing the wrappers; the same applies to the target-constant aliases at lines 55–60 and the `messageOf`/`invoke` forwarding methods near the end of the class. This is readability cleanup, not a performance claim. ########## pinot-spi/src/test/java/org/apache/pinot/spi/ingestion/error/IngestionSourceErrorAssertions.java: ########## @@ -0,0 +1,64 @@ +/** + * 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.Optional; +import javax.annotation.Nullable; +import org.apache.commons.lang3.exception.ExceptionUtils; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertTrue; + + +/// Assertions for the fixture tests that pin the classifier's rules against each SDK. A failed assertion prints the +/// whole exception tree, so a rule can be corrected from it. +public final class IngestionSourceErrorAssertions { + private IngestionSourceErrorAssertions() { + } + + public static IngestionSourceErrorClassification classify(Throwable failure) { + return IngestionSourceErrorClassifier.classify(failure) + .orElseThrow(() -> new AssertionError("Not classified: " + describe(failure))); + } + + public static IngestionSourceErrorClassification assertAuth(Throwable failure, + IngestionSourceErrorSubReason subReason, @Nullable String target) { + IngestionSourceErrorClassification classification = classify(failure); + assertEquals(classification.getCategory(), IngestionSourceErrorCategory.AUTH, describe(failure)); + assertEquals(classification.getSubReason(), subReason, describe(failure)); + assertEquals(classification.getTarget(), target, describe(failure)); Review Comment: **Nonblocking test optimization: render exception diagnostics once.** `describe(failure)` renders the entire stack trace three times in `assertAuth()` and twice in `assertSourceRead()`, including on successful assertions. Compute the description once per helper invocation, or render it only on a mismatch, to remove repeated work while retaining useful failure diagnostics. ########## pinot-spi/src/main/java/org/apache/pinot/spi/ingestion/error/IngestionSourceErrorRules.java: ########## @@ -0,0 +1,788 @@ +/** + * 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 can move a failure to another category or leave it unclassified, so an upgrade of these SDKs reruns those +/// fixtures against real clients where they can. +/// +/// 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, below, 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, below, 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) { + IngestionSourceErrorVerdict refused = tlsAlert(node, null); + if (refused != null) { + return refused; + } + return isUnreachable(node) ? unreachable(node, "") : null; + } + + /// A TLS peer's alert rejecting the certificate we presented, or denying us access, or null for any other failure. + @Nullable + private static IngestionSourceErrorVerdict tlsAlert(Throwable node, @Nullable String target) { + if (!(node instanceof SSLException)) { + return null; + } + Matcher alert = TLS_ALERT.matcher(messageOf(node)); + if (!alert.find()) { + return null; + } + String name = alert.group(1); + if (TLS_CERTIFICATE_REJECTED_ALERTS.contains(name)) { + return auth(IngestionSourceErrorSubReason.AUTHENTICATION, name, target); + } + return TLS_ACCESS_DENIED_ALERT.equals(name) ? auth(IngestionSourceErrorSubReason.AUTHORIZATION, name, target) + : 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; + } + + /// An endpoint that never answered: the source's, unless it was a metadata endpoint serving our own identity, named + /// by the network exception or by the client wrapping it. + private static IngestionSourceErrorVerdict unreachable(Throwable cause, String wrapperMessage) { + String code = cause.getClass().getSimpleName(); + return namesMetadataEndpoint(messageOf(cause)) || namesMetadataEndpoint(wrapperMessage) ? none(code) + : sourceRead(code, null); + } + + private static boolean namesMetadataEndpoint(String message) { + for (String endpoint : METADATA_ENDPOINTS) { + if (message.contains(endpoint)) { + return true; + } + } + return false; + } + + // --------------------------------------------------------------------------------------------------------------- + // AWS SDK v2: Kinesis, STS, KMS, Glue + + private static final String AWS_SERVICE_EXCEPTION = "software.amazon.awssdk.awscore.exception.AwsServiceException"; + private static final String AWS_CLIENT_EXCEPTION = "software.amazon.awssdk.core.exception.SdkClientException"; + private static final String AWS_STS_EXCEPTION = "software.amazon.awssdk.services.sts.model.StsException"; + private static final String AWS_ABORTED = "software.amazon.awssdk.core.exception.AbortedException"; + // The SDK's Apache HTTP clients, 4 and 5: the endpoint closed the connection without answering + private static final Set<String> APACHE_NO_HTTP_RESPONSE = Set.of("org.apache.http.NoHttpResponseException", + "org.apache.hc.core5.http.NoHttpResponseException"); + private static final Set<String> AWS_CALL_TIMEOUTS = Set.of( + "software.amazon.awssdk.core.exception.ApiCallTimeoutException", + "software.amazon.awssdk.core.exception.ApiCallAttemptTimeoutException"); + // Codes that reject a credential. Which credential decides the verdict, see `awsCredentialCode`. + private static final Set<String> AWS_CREDENTIAL_CODES = Set.of("InvalidAccessKeyId", "SignatureDoesNotMatch", + "InvalidSignatureException", "ExpiredToken", "ExpiredTokenException", "InvalidToken", "TokenRefreshRequired", + "InvalidClientTokenId", "IncompleteSignature", "MissingAuthenticationToken", "UnrecognizedClientException"); + private static final Set<String> AWS_AUTHORIZATION_CODES = Set.of("RegionDisabledException", + "GlueEncryptionException", "AccountProblem", "NotSignedUp"); + private static final Set<String> AWS_KMS_AUTHORIZATION_CODES = Set.of("KMSAccessDeniedException", + "KMSOptInRequired", "KMSKeyNotAccessibleFault"); + private static final Set<String> AWS_SOURCE_READ_CODES = Set.of("NoSuchKey", "NoSuchBucket", "InvalidObjectState", + "ObjectNotInActiveTierError", "PermanentRedirect", "AuthorizationHeaderMalformed", + "IllegalLocationConstraintException", "SlowDown", "ServiceUnavailable", "ServiceUnavailableException", + "InternalError", "InternalFailure", "Throttling", "ThrottlingException", "EntityNotFoundException", + "InternalServiceException", "AllAccessDisabled", "InvalidPayer", "ResourceNotFoundException", + "ResourceInUseException", "ProvisionedThroughputExceededException", "LimitExceededException"); + private static final Set<String> AWS_KMS_SOURCE_READ_CODES = Set.of("KMS.DisabledException", + "KMS.KMSInvalidStateException", "KMS.NotFoundException", "KMSDisabledException", "KMSInvalidStateException", + "KMSNotFoundException", "KMSThrottlingException"); + // Our clock, a request we built badly, our own web identity, or a Kinesis iterator our consumer let expire. + private static final Set<String> AWS_NONE_CODES = Set.of("RequestTimeTooSkewed", "RequestExpired", + "RequestInTheFuture", "ValidationError", "ValidationException", "MalformedPolicyDocument", + "PackedPolicyTooLarge", "InvalidIdentityToken", "IDPRejectedClaim", "IDPCommunicationError", + "InvalidInputException", "InvalidArgumentException", "ExpiredIteratorException", "ExpiredNextTokenException"); + // AWS reports our clock skew under a signature code; only the message tells it from a wrong secret. + private static final List<String> AWS_CLOCK_SKEW = List.of("Signature expired", "Signature not yet current"); + + @Nullable + private static IngestionSourceErrorVerdict awsService(Throwable node, + @Nullable IngestionSourceErrorEvidence evidence) { + boolean sts = hasClass(node.getClass(), AWS_STS_EXCEPTION); + Object details = invoke(node, "awsErrorDetails"); + Object code = details != null ? invoke(details, "errorCode") : null; + Object status = invoke(node, "statusCode"); + return awsCode(sts, code != null ? code.toString() : null, status instanceof Integer ? (Integer) status : 0, + messageOf(node), evidence); + } + + @Nullable + private static IngestionSourceErrorVerdict awsCode(boolean sts, @Nullable String code, int status, String message, + @Nullable IngestionSourceErrorEvidence evidence) { + String target = sts ? TARGET_STS : evidence != null ? evidence.getTarget() : null; + if (code == null || code.isEmpty()) { + // A response with no body (a HEAD): a deny, a missing object without list permission and our skew look alike. + if (status == 403) { + return auth(IngestionSourceErrorSubReason.MISSING_OR_DENIED, "403", target); + } + if (status == 404 || status == 400 || status == 429 || (status >= 300 && status < 400) || status >= 500) { + return sourceRead(String.valueOf(status), target); + } + return null; + } + if (code.equals("AccessDenied") || code.equals("AccessDeniedException")) { + if (message.contains("AssumeRoleWithWebIdentity")) { + // Our own web identity, not the table's role. + return none(code); + } + if (message.contains("ciphertext refers to a customer master key that does not exist")) { + return auth(IngestionSourceErrorSubReason.MISSING_OR_DENIED, code, TARGET_KMS); + } + return auth(IngestionSourceErrorSubReason.AUTHORIZATION, code, message.contains("kms:") ? TARGET_KMS : target); + } + if (AWS_AUTHORIZATION_CODES.contains(code)) { + return auth(IngestionSourceErrorSubReason.AUTHORIZATION, code, target); + } + if (AWS_KMS_AUTHORIZATION_CODES.contains(code)) { + return auth(IngestionSourceErrorSubReason.AUTHORIZATION, code, TARGET_KMS); + } + if (AWS_CREDENTIAL_CODES.contains(code)) { + for (String skew : AWS_CLOCK_SKEW) { + if (message.contains(skew)) { + return none(code); + } + } + return awsCredentialCode(sts, code, target, evidence); + } + if (AWS_KMS_SOURCE_READ_CODES.contains(code)) { + return sourceRead(code, TARGET_KMS); + } + if (AWS_SOURCE_READ_CODES.contains(code) || status == 429 || status >= 500) { + return sourceRead(code, target); + } + if (AWS_NONE_CODES.contains(code)) { + return none(code); + } + return null; + } + + /// A credential code is the table's bad credential only when the table's static keys were what got rejected: by + /// STS on the AssumeRole call when a role is set, or by the service when none is. Our ambient identity, and the + /// session credentials the SDK refreshes for an assumed role, are ours: the table owner's role and policy decisions + /// arrive as `AccessDenied` instead. With no recorded credential mode the rejected credential is unknown, so it + /// fails closed. + private static IngestionSourceErrorVerdict awsCredentialCode(boolean sts, String code, @Nullable String target, + @Nullable IngestionSourceErrorEvidence evidence) { + if (evidence == null || evidence.getCredentialMode() != IngestionSourceCredentialMode.STATIC) { + return none(code); + } + boolean baseKeysRejected = sts || !evidence.isRoleArnConfigured(); + return baseKeysRejected ? auth(IngestionSourceErrorSubReason.AUTHENTICATION, code, target) : none(code); + } + + /// No service exception sits beneath this node, or it would have decided first. + @Nullable + private static IngestionSourceErrorVerdict awsClient(Throwable node, List<Throwable> below) { + Class<?> clazz = node.getClass(); + String message = messageOf(node); + // The process's own credential chain ("Unable to load credentials from any of the providers ..."), one of its + // providers called directly after a first success ("Failed to load credentials from IMDS." or "... from metadata + // service."), or our cancellation. + if (message.startsWith("Unable to load credentials") || message.startsWith("Failed to load credentials from") + || hasClass(clazz, AWS_ABORTED)) { + return none(clazz.getSimpleName()); + } + if (hasAnyClass(clazz, AWS_CALL_TIMEOUTS)) { + return sourceRead(clazz.getSimpleName(), null); + } + for (Throwable cause : below) { + IngestionSourceErrorVerdict refused = tlsAlert(cause, null); + if (refused != null) { + return refused; + } + if (isUnreachable(cause)) { + return unreachable(cause, message); + } + if (hasAnyClass(cause.getClass(), APACHE_NO_HTTP_RESPONSE)) { + return sourceRead(cause.getClass().getSimpleName(), null); + } + } + // Includes the cause-less "Request attempt N failure" records the SDK attaches as suppressed exceptions. + return null; + } + + // --------------------------------------------------------------------------------------------------------------- + // Kafka brokers (kafka-clients 3.x and 4.x) + + private static final String KAFKA_ERRORS = "org.apache.kafka.common.errors."; + private static final String KAFKA_AUTHENTICATION = KAFKA_ERRORS + "AuthenticationException"; + private static final String KAFKA_AUTHORIZATION = KAFKA_ERRORS + "AuthorizationException"; + private static final String KAFKA_SASL_AUTHENTICATION = KAFKA_ERRORS + "SaslAuthenticationException"; + private static final String KAFKA_SSL_AUTHENTICATION = KAFKA_ERRORS + "SslAuthenticationException"; + private static final String KAFKA_CONFIG_EXCEPTION = "org.apache.kafka.common.config.ConfigException"; + private static final Set<String> KAFKA_BROKER_SOURCE_READ = Set.of(KAFKA_ERRORS + "TimeoutException", + KAFKA_ERRORS + "DisconnectException", KAFKA_ERRORS + "UnknownTopicOrPartitionException", + KAFKA_ERRORS + "OffsetOutOfRangeException", "org.apache.kafka.clients.consumer.OffsetOutOfRangeException"); + // The schema-registry serde maps a 429 to this class and drops the cause. + private static final String KAFKA_THROTTLING_QUOTA = KAFKA_ERRORS + "ThrottlingQuotaExceededException"; + private static final String SASL_EXCEPTION = "javax.security.sasl.SaslException"; + // A certificate we failed to validate: we rejected the broker. + private static final Set<String> CERTIFICATE_REJECTED = Set.of("java.security.cert.CertificateException", + "java.security.cert.CertPathValidatorException", "sun.security.validator.ValidatorException"); + private static final Pattern TLS_ALERT = Pattern.compile("Received fatal alert: ([a-z_]+)"); + // Alerts a peer sends when it rejects the certificate we presented. + private static final Set<String> TLS_CERTIFICATE_REJECTED_ALERTS = Set.of("bad_certificate", + "unsupported_certificate", "certificate_revoked", "certificate_expired", "certificate_unknown", "unknown_ca", + "certificate_required"); + private static final String TLS_ACCESS_DENIED_ALERT = "access_denied"; + + @Nullable + private static IngestionSourceErrorVerdict kafka(Throwable node, List<Throwable> below, + @Nullable IngestionSourceErrorEvidence evidence) { + Class<?> clazz = node.getClass(); + String code = clazz.getSimpleName(); + if (hasClass(clazz, KAFKA_AUTHORIZATION)) { + return auth(IngestionSourceErrorSubReason.AUTHORIZATION, code, TARGET_BROKER); + } + if (hasClass(clazz, KAFKA_AUTHENTICATION)) { + return kafkaAuthentication(node, below, evidence); + } + if (hasClass(clazz, KAFKA_THROTTLING_QUOTA)) { + return sourceRead(code, TARGET_SCHEMA_REGISTRY); + } + if (hasAnyClass(clazz, KAFKA_BROKER_SOURCE_READ)) { + // A broker's timeout carries no cause. The schema-registry serde reuses these classes over the network cause of + // its own request, so with a cause the endpoint is unknown here. + return sourceRead(code, below.isEmpty() ? TARGET_BROKER : null); + } + if (hasClass(clazz, KAFKA_CONFIG_EXCEPTION)) { + // Only an unresolvable bootstrap list is the source's; any other config error is ours. + return messageOf(node).startsWith("No resolvable bootstrap urls") + ? sourceRead(code, TARGET_BROKER) + : none(code); + } + return null; + } + + /// Kafka's authentication classes say less than they seem to. MSK IAM reports a policy denial as a SASL + /// authentication failure, a client-side failure to build the SASL token (our credential chain, our Kerberos setup) + /// arrives under the same class with a `SaslException` cause, a broker's connection throttle reuses it, and an SSL + /// failure covers both directions of a rejected certificate. + private static IngestionSourceErrorVerdict kafkaAuthentication(Throwable node, List<Throwable> below, + @Nullable IngestionSourceErrorEvidence evidence) { + Class<?> clazz = node.getClass(); + String code = clazz.getSimpleName(); + if (hasClass(clazz, KAFKA_SASL_AUTHENTICATION) && messageOf(node).contains("Too many connects")) { + return sourceRead("too_many_connects", TARGET_BROKER); + } + if (hasClass(clazz, KAFKA_SSL_AUTHENTICATION)) { + return kafkaTls(code, below); + } + for (Throwable cause : below) { + if (hasClass(cause.getClass(), SASL_EXCEPTION)) { + // A broker rejection carries no cause; this one failed on our side before the broker judged anything. + return none(code); + } + } + // Signing with a cloud identity (MSK IAM with the process's identity or an assumed role), a policy denial arrives + // as an authentication failure: the class can't tell which. + boolean cloudIdentity = evidence != null + && (evidence.getCredentialMode() == IngestionSourceCredentialMode.AMBIENT || evidence.isRoleArnConfigured()); + IngestionSourceErrorSubReason subReason = + cloudIdentity ? IngestionSourceErrorSubReason.UNSPECIFIED : IngestionSourceErrorSubReason.AUTHENTICATION; + return auth(subReason, code, TARGET_BROKER); + } + + private static IngestionSourceErrorVerdict kafkaTls(String code, List<Throwable> below) { + for (Throwable cause : below) { + if (hasAnyClass(cause.getClass(), CERTIFICATE_REJECTED)) { + // We could not verify the broker: the endpoint, not our credential. + return sourceRead(code, TARGET_BROKER); + } + } + for (Throwable cause : below) { + IngestionSourceErrorVerdict refused = tlsAlert(cause, TARGET_BROKER); + if (refused != null) { + return refused; + } + } + // Any other handshake failure: the TLS connection to the endpoint failed. + return sourceRead(code, TARGET_BROKER); + } + + // --------------------------------------------------------------------------------------------------------------- + // Confluent schema registry (kafka-schema-registry-client) + + private static final String CONFLUENT_REST_CLIENT_EXCEPTION = + "io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException"; + + /// The registry's HTTP status decides; its error code does not (an HTML 403 from a proxy reads as 50005). The serde + /// wraps this exception in the same Kafka authentication and authorization classes a broker uses, so it decides + /// first, as the deeper node, and names the registry as the target. + @Nullable + private static IngestionSourceErrorVerdict schemaRegistry(Throwable node) { + Object statusValue = invoke(node, "getStatus"); + if (!(statusValue instanceof Integer)) { + return null; + } + int status = (Integer) statusValue; + IngestionSourceErrorVerdict verdict = httpStatus(status, TARGET_SCHEMA_REGISTRY); + // Another 4xx: a request Pinot built badly. + return verdict != null ? verdict : none(String.valueOf(status)); + } + + // --------------------------------------------------------------------------------------------------------------- + // Logins: Kerberos (Krb5LoginModule) and the Kafka OAUTHBEARER login, which both stringify their evidence + + private static final String LOGIN_EXCEPTION = "javax.security.auth.login.LoginException"; + // kafka-clients' token retriever, stringified into the LoginException: "The response code 401 and error response + // {"invalid_client" - "..."} was encountered ..." for a refusal, "The unexpected response code 503 ..." (4.x) for a + // failure. + private static final Pattern OAUTH_RESPONSE_CODE = Pattern.compile("The (?:unexpected )?response code (\\d{3})"); + private static final Pattern OAUTH_ERROR = + Pattern.compile("(?:\"error\"\\s*:\\s*\"|error response \\{\")([a-z_]{1,64})\""); + // The 3.x token retriever's text when the endpoint failed or never answered. + private static final String OAUTH_TOKEN_UNAVAILABLE = "An internal error occurred while retrieving token"; + // 4.x dumps the retriever's stack trace into the message instead: a network failure names its class there. + private static final List<String> DUMPED_UNREACHABLE = List.of("java.net.ConnectException", + "java.net.UnknownHostException", "java.net.SocketTimeoutException", "java.net.NoRouteToHostException", + "java.net.http.HttpTimeoutException", "javax.net.ssl.SSLHandshakeException"); + private static final Set<String> OAUTH_AUTHENTICATION = Set.of("invalid_client", "invalid_grant", + "unsupported_grant_type"); + private static final Set<String> OAUTH_AUTHORIZATION = Set.of("unauthorized_client", "invalid_scope", + "access_denied"); + private static final Set<String> OAUTH_UNAVAILABLE = Set.of("server_error", "temporarily_unavailable"); + private static final String OAUTH_INVALID_REQUEST = "invalid_request"; + private static final List<String> KERBEROS_UNREACHABLE = List.of("Cannot contact any KDC", "Receive timed out"); + // Our clock, keytab or krb5 setup. + private static final List<String> KERBEROS_LOCAL_FAILURES = List.of("Clock skew too great", + "Unable to obtain password", "Cannot locate default realm", "Cannot locate KDC"); + // The KDC rejecting the principal: client not found (6), credentials revoked (18), pre-authentication failed (24). + private static final Pattern KERBEROS_REJECTED = Pattern.compile("\\((6|18|24)\\)"); + + @Nullable + private static IngestionSourceErrorVerdict login(Throwable node) { + String message = messageOf(node); + Matcher responseCode = OAUTH_RESPONSE_CODE.matcher(message); + if (responseCode.find()) { + Matcher error = OAUTH_ERROR.matcher(message); + return oauth(error.find() ? error.group(1) : null, Integer.parseInt(responseCode.group(1))); + } + if (message.contains(OAUTH_TOKEN_UNAVAILABLE)) { + // The token endpoint failed or never answered; the client drops which. + return sourceRead("token_unavailable", TARGET_IDP); + } + for (String unreachable : DUMPED_UNREACHABLE) { + if (message.contains(unreachable)) { + return sourceRead("token_unavailable", TARGET_IDP); + } + } + for (String unreachable : KERBEROS_UNREACHABLE) { + if (message.contains(unreachable)) { + return sourceRead("kdc_unreachable", TARGET_IDP); + } + } + for (String ours : KERBEROS_LOCAL_FAILURES) { + if (message.contains(ours)) { + return none(node.getClass().getSimpleName()); + } + } + Matcher rejected = KERBEROS_REJECTED.matcher(message); + if (rejected.find()) { + return auth(IngestionSourceErrorSubReason.AUTHENTICATION, "krb5_" + rejected.group(1), TARGET_IDP); + } + return null; + } + + /// An OAuth 2.0 token endpoint's `error` (RFC 6749 section 5.2), falling back to its status. + @Nullable + private static IngestionSourceErrorVerdict oauth(@Nullable String error, int status) { + if (error != null) { + String normalized = error.toLowerCase(Locale.ROOT); + if (OAUTH_AUTHENTICATION.contains(normalized)) { + return auth(IngestionSourceErrorSubReason.AUTHENTICATION, normalized, TARGET_IDP); + } + if (OAUTH_AUTHORIZATION.contains(normalized)) { + return auth(IngestionSourceErrorSubReason.AUTHORIZATION, normalized, TARGET_IDP); + } + if (OAUTH_UNAVAILABLE.contains(normalized)) { + return sourceRead(normalized, TARGET_IDP); + } + if (OAUTH_INVALID_REQUEST.equals(normalized)) { + // A malformed request: ours. + return none(normalized); + } + } + return httpStatus(status, TARGET_IDP); + } + + /// An HTTP status from a service with no error code of our own to read: 401 and 403 refuse us, 404, 408, 429 and + /// 5xx can't serve us, anything else decides nothing. + @Nullable + private static IngestionSourceErrorVerdict httpStatus(int status, String target) { + String code = String.valueOf(status); + if (status == 401) { + return auth(IngestionSourceErrorSubReason.AUTHENTICATION, code, target); + } + if (status == 403) { + return auth(IngestionSourceErrorSubReason.AUTHORIZATION, code, target); + } + if (status == 404 || status == 408 || status == 429 || status >= 500) { + return sourceRead(code, target); + } + return null; + } + + // --------------------------------------------------------------------------------------------------------------- + // Pulsar client (v4 and v5 APIs) and admin + + private static final String PULSAR_CLIENT_EXCEPTION = "org.apache.pulsar.client.api.PulsarClientException"; + private static final String PULSAR_V5_CLIENT_EXCEPTION = "org.apache.pulsar.client.api.v5.PulsarClientException"; + private static final String PULSAR_ADMIN_EXCEPTION = "org.apache.pulsar.client.admin.PulsarAdminException"; + private static final Set<String> PULSAR_SOURCE_READ = Set.of("TopicDoesNotExistException", "NotFoundException", + "SubscriptionNotFoundException", "LookupException", "ConnectException", "TimeoutException", + "BrokerMetadataException", "BrokerPersistenceException", "TopicTerminatedException", "NotAllowedException", + "ConsumerBusyException", "NotConnectedException"); + private static final Set<String> PULSAR_NONE = Set.of("InvalidServiceURL", "InvalidServiceURLException", + "InvalidConfigurationException", "InvalidTopicNameException", "UnsupportedAuthenticationException", + "AlreadyClosedException", "NotSupportedException", "CryptoException", "ChecksumException", + "GettingAuthenticationDataException"); + // The OAuth2 plugin's token errors: "Unable to obtain an access token: <description> (<error>)". + private static final String PULSAR_OAUTH_TOKEN = "Unable to obtain an access token"; + private static final Pattern PULSAR_OAUTH_ERROR = Pattern.compile("\\(([a-z_]{1,64})\\)\\s*$"); + + /// Pulsar's class is its code: the broker's error maps one to one onto a nested subclass. Every subclass of the + /// base class is matched on its own name and the names of its superclasses up to the base, so + /// `ServiceNotReadyException` and `TooManyRequestsException` read as the `LookupException` they extend. + @Nullable + private static IngestionSourceErrorVerdict pulsar(Throwable node) { + String message = messageOf(node); + for (Class<?> current = node.getClass(); current != null; current = current.getSuperclass()) { + String name = current.getName(); + if (name.equals(PULSAR_CLIENT_EXCEPTION) || name.equals(PULSAR_V5_CLIENT_EXCEPTION)) { + // The base class alone: TLS failures surface only as "Connection already closed"; anything else, such as an + // HTTP lookup failure, is left to a deeper or more specific exception. + return message.contains("Connection already closed") ? sourceRead("connection_closed", TARGET_BROKER) : null; + } + String simple = name.substring(name.lastIndexOf('$') + 1); + if (simple.equals("AuthenticationException")) { + return pulsarAuthentication(message); + } + if (simple.equals("AuthorizationException")) { + return auth(IngestionSourceErrorSubReason.AUTHORIZATION, simple, TARGET_BROKER); + } + if (simple.equals("TooManyRequestsException") && message.contains("Requests number out of config")) { + // The client's own lookup semaphore, not the broker's throttle. + return none(simple); + } + if (PULSAR_SOURCE_READ.contains(simple)) { + return sourceRead(simple, TARGET_BROKER); + } + if (PULSAR_NONE.contains(simple)) { + return none(simple); + } + } + return null; + } + + /// Pulsar's authentication class also covers an identity provider that failed or never answered, and a key file + /// we couldn't read. Only the message tells them apart. + private static IngestionSourceErrorVerdict pulsarAuthentication(String message) { + String code = "AuthenticationException"; + if (message.startsWith("Unable to read private key")) { + return none(code); + } + if (message.contains("Unable to retrieve OAuth 2.0 server metadata")) { + return sourceRead(code, TARGET_IDP); + } + if (message.contains(PULSAR_OAUTH_TOKEN)) { + Matcher error = PULSAR_OAUTH_ERROR.matcher(message); + if (error.find()) { + // An error we don't recognise decides nothing: it could be a refusal. + return oauth(error.group(1), 0); + } + // Without an OAuth error the token endpoint returned another status or never answered. + return sourceRead(code, TARGET_IDP); + } + // The broker rejected the token: it gives one generic message for every reason. + return auth(IngestionSourceErrorSubReason.AUTHENTICATION, code, TARGET_BROKER); + } + + /// The admin client maps both 401 and 403 to `NotAuthorizedException`: only the status tells them apart. The base + /// class without an HTTP error is a client-side failure that defaults its status to 500, so its status means nothing. + @Nullable + private static IngestionSourceErrorVerdict pulsarAdmin(Throwable node) { + String simple = node.getClass().getSimpleName(); + if (simple.equals("ConnectException") || simple.equals("TimeoutException")) { + return sourceRead(simple, TARGET_ADMIN); + } + if (simple.equals("GettingAuthenticationDataException")) { + return none(simple); + } + if (node.getClass().getName().equals(PULSAR_ADMIN_EXCEPTION) && invoke(node, "getHttpError") == null) { + return null; + } + Object status = invoke(node, "getStatusCode"); + return status instanceof Integer ? httpStatus((Integer) status, TARGET_ADMIN) : null; + } + + // --------------------------------------------------------------------------------------------------------------- + // gRPC and gax (Google Cloud clients), and google-auth + + private static final String GAX_API_EXCEPTION = "com.google.api.gax.rpc.ApiException"; + private static final String GRPC_STATUS_RUNTIME_EXCEPTION = "io.grpc.StatusRuntimeException"; + private static final String GRPC_STATUS_EXCEPTION = "io.grpc.StatusException"; + private static final String GOOGLE_AUTH_EXCEPTION = "com.google.auth.oauth2.GoogleAuthException"; + private static final Set<String> GRPC_SOURCE_READ = Set.of("NOT_FOUND", "FAILED_PRECONDITION", + "RESOURCE_EXHAUSTED", "UNAVAILABLE", "DEADLINE_EXCEEDED", "INTERNAL"); + private static final Set<String> GRPC_NONE = Set.of("CANCELLED", "INVALID_ARGUMENT", "ALREADY_EXISTS"); + // Google Cloud's deliberate conflation of a missing project with a denied one. + private static final String GOOGLE_MISSING_OR_DENIED = "not found or user does not have access"; + // "Token must be a short-lived token ... Check your iat and exp values": our clock. + private static final String GOOGLE_CLOCK_SKEW = "short-lived token"; + + @Nullable + private static IngestionSourceErrorVerdict grpc(@Nullable Object codeValue, Throwable node, List<Throwable> below, + @Nullable IngestionSourceErrorEvidence evidence) { + if (!(codeValue instanceof Enum)) { + return null; + } + // Both gax's StatusCode.Code and gRPC's Status.Code are enums: read the constant's name, never a toString(). + String code = ((Enum<?>) codeValue).name(); + switch (code) { + case "UNAUTHENTICATED": + // A rejected credential is the table owner's only when it was the table's: under the process's own + // identity, grpc-auth reports our failed token refresh this way. + return evidence != null && evidence.getCredentialMode() == IngestionSourceCredentialMode.STATIC + ? auth(IngestionSourceErrorSubReason.AUTHENTICATION, code, null) : none(code); Review Comment: **[P2] Inspect Google token-refresh failures before trusting UNAUTHENTICATED.** With STATIC credentials, a token-endpoint HTTP 502 or 504 becomes `AUTH` here. I reproduced this through the shipped `ServiceAccountCredentials` 1.54.0 and actual grpc-auth `MoreCallCredentials` 1.85.1: those refresh failures become gRPC `UNAUTHENTICATED`, while `googleAuth()` leaves their preserved HTTP status undecided. In contrast, HTTP 500/503 and a refused connection become `UNAVAILABLE` and correctly classify as SOURCE_READ. Read the underlying refresh response before accepting the synthesized gRPC status, so 502/504 remain SOURCE_READ. The regression should invoke `MoreCallCredentials`; the current fixture supplies the gRPC status manually and misses this mapping. -- 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]
