diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminMetadataManager.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminMetadataManager.java index 9dc2e190d134c..0ac5419991e1b 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminMetadataManager.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminMetadataManager.java @@ -23,11 +23,10 @@ import org.apache.kafka.common.Node; import org.apache.kafka.common.errors.ApiException; import org.apache.kafka.common.errors.AuthenticationException; -import org.apache.kafka.common.errors.MismatchedEndpointTypeException; -import org.apache.kafka.common.errors.UnsupportedEndpointTypeException; import org.apache.kafka.common.errors.UnsupportedVersionException; import org.apache.kafka.common.requests.MetadataResponse; import org.apache.kafka.common.requests.RequestHeader; +import org.apache.kafka.common.requests.RequestUtils; import org.apache.kafka.common.utils.LogContext; import org.slf4j.Logger; @@ -277,23 +276,21 @@ public void updateFailed(Throwable exception) { // We depend on pending calls to request another metadata update this.state = State.QUIESCENT; - if (exception instanceof AuthenticationException) { - log.warn("Metadata update failed due to authentication error", exception); - this.fatalException = (ApiException) exception; - } else if (exception instanceof MismatchedEndpointTypeException) { - log.warn("Metadata update failed due to mismatched endpoint type error", exception); - this.fatalException = (ApiException) exception; - } else if (exception instanceof UnsupportedEndpointTypeException) { - log.warn("Metadata update failed due to unsupported endpoint type error", exception); - this.fatalException = (ApiException) exception; - } else if (exception instanceof UnsupportedVersionException) { - if (usingBootstrapControllers) { - log.warn("The remote node is not a CONTROLLER that supports the KIP-919 " + - "DESCRIBE_CLUSTER api.", exception); - } else { - log.warn("The remote node is not a BROKER that supports the METADATA api.", exception); + if (RequestUtils.isFatalException(exception)) { + log.warn("Fatal error during metadata update", exception); + // avoid unchecked/unconfirmed cast to ApiException + if (exception instanceof ApiException) { + this.fatalException = (ApiException) exception; + } + + if (exception instanceof UnsupportedVersionException) { + if (usingBootstrapControllers) { + log.warn("The remote node is not a CONTROLLER that supports the KIP-919 " + + "DESCRIBE_CLUSTER api.", exception); + } else { + log.warn("The remote node is not a BROKER that supports the METADATA api.", exception); + } } - this.fatalException = (ApiException) exception; } else { log.info("Metadata update failed", exception); } diff --git a/clients/src/main/java/org/apache/kafka/common/requests/RequestUtils.java b/clients/src/main/java/org/apache/kafka/common/requests/RequestUtils.java index cc6e5a2303879..d434e6e7b185e 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/RequestUtils.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/RequestUtils.java @@ -16,6 +16,13 @@ */ package org.apache.kafka.common.requests; +import org.apache.kafka.common.errors.AuthenticationException; +import org.apache.kafka.common.errors.AuthorizationException; +import org.apache.kafka.common.errors.MismatchedEndpointTypeException; +import org.apache.kafka.common.errors.SecurityDisabledException; +import org.apache.kafka.common.errors.UnsupportedEndpointTypeException; +import org.apache.kafka.common.errors.UnsupportedForMessageFormatException; +import org.apache.kafka.common.errors.UnsupportedVersionException; import org.apache.kafka.common.message.ProduceRequestData; import org.apache.kafka.common.protocol.ByteBufferAccessor; import org.apache.kafka.common.protocol.Message; @@ -77,4 +84,14 @@ public static ByteBuffer serialize( writable.flip(); return writable.buffer(); } + + public static boolean isFatalException(Throwable e) { + return e instanceof AuthenticationException || + e instanceof AuthorizationException || + e instanceof MismatchedEndpointTypeException || + e instanceof SecurityDisabledException || + e instanceof UnsupportedVersionException || + e instanceof UnsupportedEndpointTypeException || + e instanceof UnsupportedForMessageFormatException; + } } diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/internals/AdminMetadataManagerTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/internals/AdminMetadataManagerTest.java index 5620dd06a5e71..1e6823ea8ee64 100644 --- a/clients/src/test/java/org/apache/kafka/clients/admin/internals/AdminMetadataManagerTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/admin/internals/AdminMetadataManagerTest.java @@ -20,6 +20,7 @@ import org.apache.kafka.common.Cluster; import org.apache.kafka.common.Node; import org.apache.kafka.common.errors.AuthenticationException; +import org.apache.kafka.common.errors.AuthorizationException; import org.apache.kafka.common.utils.LogContext; import org.apache.kafka.common.utils.MockTime; @@ -98,6 +99,16 @@ public void testAuthenticationFailure() { assertTrue(mgr.isReady()); } + @Test + public void testAuthorizationFailure() { + mgr.transitionToUpdatePending(time.milliseconds()); + mgr.updateFailed(new AuthorizationException("Authorization failed")); + assertEquals(refreshBackoffMs, mgr.metadataFetchDelayMs(time.milliseconds())); + assertThrows(AuthorizationException.class, mgr::isReady); + mgr.update(mockCluster(), time.milliseconds()); + assertTrue(mgr.isReady()); + } + @Test public void testNeedsRebootstrap() { long rebootstrapTriggerMs = 1000; diff --git a/clients/src/test/java/org/apache/kafka/common/requests/RequestUtilsTest.java b/clients/src/test/java/org/apache/kafka/common/requests/RequestUtilsTest.java new file mode 100644 index 0000000000000..ec7789f1a0263 --- /dev/null +++ b/clients/src/test/java/org/apache/kafka/common/requests/RequestUtilsTest.java @@ -0,0 +1,48 @@ +/* + * 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.kafka.common.requests; + +import org.apache.kafka.common.errors.AuthenticationException; +import org.apache.kafka.common.errors.AuthorizationException; +import org.apache.kafka.common.errors.DisconnectException; +import org.apache.kafka.common.errors.MismatchedEndpointTypeException; +import org.apache.kafka.common.errors.SecurityDisabledException; +import org.apache.kafka.common.errors.UnsupportedEndpointTypeException; +import org.apache.kafka.common.errors.UnsupportedForMessageFormatException; +import org.apache.kafka.common.errors.UnsupportedVersionException; + +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class RequestUtilsTest { + @Test + public void testIsFatalException() { + assertTrue(RequestUtils.isFatalException(new AuthenticationException(""))); + assertTrue(RequestUtils.isFatalException(new AuthorizationException(""))); + assertTrue(RequestUtils.isFatalException(new MismatchedEndpointTypeException(""))); + assertTrue(RequestUtils.isFatalException(new SecurityDisabledException(""))); + assertTrue(RequestUtils.isFatalException(new UnsupportedEndpointTypeException(""))); + assertTrue(RequestUtils.isFatalException(new UnsupportedForMessageFormatException(""))); + assertTrue(RequestUtils.isFatalException(new UnsupportedVersionException(""))); + + // retriable exceptions + assertFalse(RequestUtils.isFatalException(new DisconnectException(""))); + } +} diff --git a/test-common/test-common-api/src/test/java/org/apache/kafka/common/test/api/ClusterTestExtensionsTest.java b/test-common/test-common-api/src/test/java/org/apache/kafka/common/test/api/ClusterTestExtensionsTest.java index 20d2d3fd696d8..c680ff004ae79 100644 --- a/test-common/test-common-api/src/test/java/org/apache/kafka/common/test/api/ClusterTestExtensionsTest.java +++ b/test-common/test-common-api/src/test/java/org/apache/kafka/common/test/api/ClusterTestExtensionsTest.java @@ -20,7 +20,6 @@ import org.apache.kafka.clients.admin.Admin; import org.apache.kafka.clients.admin.AdminClientConfig; import org.apache.kafka.clients.admin.Config; -import org.apache.kafka.clients.admin.DescribeAclsOptions; import org.apache.kafka.clients.admin.DescribeLogDirsResult; import org.apache.kafka.clients.admin.NewTopic; import org.apache.kafka.clients.consumer.Consumer; @@ -35,7 +34,6 @@ import org.apache.kafka.common.config.SaslConfigs; import org.apache.kafka.common.errors.ClusterAuthorizationException; import org.apache.kafka.common.errors.SaslAuthenticationException; -import org.apache.kafka.common.errors.TimeoutException; import org.apache.kafka.common.errors.TopicAuthorizationException; import org.apache.kafka.common.security.auth.SecurityProtocol; import org.apache.kafka.common.serialization.StringDeserializer; @@ -464,12 +462,12 @@ public void testSaslPlaintext(ClusterInstance clusterInstance) throws Cancellati } ) public void testSaslPlaintextWithController(ClusterInstance clusterInstance) throws CancellationException, ExecutionException, InterruptedException { - // test with admin + // default ClusterInstance#admin helper with admin credentials try (Admin admin = clusterInstance.admin(Map.of(), true)) { admin.describeAcls(AclBindingFilter.ANY).values().get(); } - // test with non-admin + // client with non-admin credentials Map nonAdminConfig = Map.of( SaslConfigs.SASL_JAAS_CONFIG, String.format( @@ -480,9 +478,25 @@ public void testSaslPlaintextWithController(ClusterInstance clusterInstance) thr try (Admin admin = clusterInstance.admin(nonAdminConfig, true)) { ExecutionException exception = assertThrows( ExecutionException.class, - () -> admin.describeAcls(AclBindingFilter.ANY, new DescribeAclsOptions().timeoutMs(5000)).values().get() + () -> admin.describeAcls(AclBindingFilter.ANY).values().get() + ); + assertInstanceOf(ClusterAuthorizationException.class, exception.getCause()); + } + + // client with unknown credentials + Map unknownUserConfig = Map.of( + SaslConfigs.SASL_JAAS_CONFIG, + String.format( + "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"%s\" password=\"%s\";", + "unknown", "unknown" + ) + ); + try (Admin admin = clusterInstance.admin(unknownUserConfig)) { + ExecutionException exception = assertThrows( + ExecutionException.class, + () -> admin.describeAcls(AclBindingFilter.ANY).values().get() ); - assertInstanceOf(TimeoutException.class, exception.getCause()); + assertInstanceOf(SaslAuthenticationException.class, exception.getCause()); } } }