This is an automated email from the ASF dual-hosted git repository. lhotari pushed a commit to branch branch-3.3 in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 63ca5ecb9e1bc9c6781d2fe8ef29f65f52a4f592 Author: fengyubiao <[email protected]> AuthorDate: Fri Sep 27 16:34:19 2024 +0800 [fix] [log] Do not print error log if tenant/namespace does not exist when calling get topic metadata (#23291) (cherry picked from commit 5583102aae135f5f62884f83e1ddd927b24ee737) --- .../apache/pulsar/broker/service/ServerCnx.java | 41 ++++++++++++++--- .../broker/admin/GetPartitionMetadataTest.java | 51 ++++++++++++++++++++++ 2 files changed, 85 insertions(+), 7 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java index e846d144d58..c98caa874c7 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java @@ -62,6 +62,8 @@ import java.util.regex.Pattern; import java.util.stream.Collectors; import javax.naming.AuthenticationException; import javax.net.ssl.SSLSession; +import javax.ws.rs.WebApplicationException; +import javax.ws.rs.core.Response; import org.apache.bookkeeper.mledger.AsyncCallbacks; import org.apache.bookkeeper.mledger.Entry; import org.apache.bookkeeper.mledger.ManagedLedgerException; @@ -660,8 +662,6 @@ public class ServerCnx extends PulsarHandler implements TransportCnx { commandSender.sendPartitionMetadataResponse(ServerError.AuthorizationError, ex.getMessage(), requestId); } else { - log.warn("Failed to get Partitioned Metadata [{}] {}: {}", remoteAddress, - topicName, ex.getMessage(), ex); ServerError error = ServerError.ServiceNotReady; if (ex instanceof MetadataStoreException) { error = ServerError.MetadataError; @@ -673,6 +673,14 @@ public class ServerCnx extends PulsarHandler implements TransportCnx { error = ServerError.MetadataError; } } + if (error == ServerError.TopicNotFound) { + log.info("Trying to get Partitioned Metadata for a resource not exist" + + "[{}] {}: {}", remoteAddress, + topicName, ex.getMessage()); + } else { + log.warn("Failed to get Partitioned Metadata [{}] {}: {}", + remoteAddress, topicName, ex.getMessage(), ex); + } commandSender.sendPartitionMetadataResponse(error, ex.getMessage(), requestId); } @@ -690,6 +698,16 @@ public class ServerCnx extends PulsarHandler implements TransportCnx { return null; }).exceptionally(ex -> { logAuthException(remoteAddress, "partition-metadata", getPrincipal(), Optional.of(topicName), ex); + Throwable actEx = FutureUtil.unwrapCompletionException(ex); + if (actEx instanceof WebApplicationException restException) { + if (restException.getResponse().getStatus() == Response.Status.NOT_FOUND.getStatusCode()) { + writeAndFlush(Commands.newPartitionMetadataResponse(ServerError.MetadataError, + "Tenant or namespace or topic does not exist: " + topicName.getNamespace() , + requestId)); + lookupSemaphore.release(); + return null; + } + } final String msg = "Exception occurred while trying to authorize get Partition Metadata"; writeAndFlush(Commands.newPartitionMetadataResponse(ServerError.AuthorizationError, msg, requestId)); @@ -3661,13 +3679,22 @@ public class ServerCnx extends PulsarHandler implements TransportCnx { private static void logAuthException(SocketAddress remoteAddress, String operation, String principal, Optional<TopicName> topic, Throwable ex) { String topicString = topic.map(t -> ", topic=" + t.toString()).orElse(""); - if (ex instanceof AuthenticationException) { + Throwable actEx = FutureUtil.unwrapCompletionException(ex); + if (actEx instanceof AuthenticationException) { log.info("[{}] Failed to authenticate: operation={}, principal={}{}, reason={}", - remoteAddress, operation, principal, topicString, ex.getMessage()); - } else { - log.error("[{}] Error trying to authenticate: operation={}, principal={}{}", - remoteAddress, operation, principal, topicString, ex); + remoteAddress, operation, principal, topicString, actEx.getMessage()); + return; + } else if (actEx instanceof WebApplicationException restException){ + // Do not print error log if users tries to access a not found resource. + if (restException.getResponse().getStatus() == Response.Status.NOT_FOUND.getStatusCode()) { + log.info("[{}] Trying to authenticate for a topic which under a namespace not exists: operation={}," + + " principal={}{}, reason: {}", + remoteAddress, operation, principal, topicString, actEx.getMessage()); + return; + } } + log.error("[{}] Error trying to authenticate: operation={}, principal={}{}", + remoteAddress, operation, principal, topicString, ex); } private static void logNamespaceNameAuthException(SocketAddress remoteAddress, String operation, diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/GetPartitionMetadataTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/GetPartitionMetadataTest.java index 87bc4267b48..e9a639697d9 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/GetPartitionMetadataTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/GetPartitionMetadataTest.java @@ -578,4 +578,55 @@ public class GetPartitionMetadataTest { assertEquals(getLookupRequestPermits(), lookupPermitsBefore); }); } + + @Test(dataProvider = "topicDomains") + public void testNamespaceNotExist(TopicDomain topicDomain) throws Exception { + int lookupPermitsBefore = getLookupRequestPermits(); + final String namespaceNotExist = BrokerTestUtil.newUniqueName("public/ns"); + final String topicNameStr = BrokerTestUtil.newUniqueName(topicDomain.toString() + "://" + namespaceNotExist + "/tp"); + PulsarClientImpl[] clientArray = getClientsToTest(false); + for (PulsarClientImpl client : clientArray) { + try { + PartitionedTopicMetadata topicMetadata = client + .getPartitionedTopicMetadata(topicNameStr, true, true) + .join(); + log.info("Get topic metadata: {}", topicMetadata.partitions); + fail("Expected a not found ex"); + } catch (Exception ex) { + Throwable unwrapEx = FutureUtil.unwrapCompletionException(ex); + assertTrue(unwrapEx instanceof PulsarClientException.BrokerMetadataException || + unwrapEx instanceof PulsarClientException.TopicDoesNotExistException); + } + } + // Verify: lookup semaphore has been releases. + Awaitility.await().untilAsserted(() -> { + assertEquals(getLookupRequestPermits(), lookupPermitsBefore); + }); + } + + @Test(dataProvider = "topicDomains") + public void testTenantNotExist(TopicDomain topicDomain) throws Exception { + int lookupPermitsBefore = getLookupRequestPermits(); + final String tenantNotExist = BrokerTestUtil.newUniqueName("tenant"); + final String namespaceNotExist = BrokerTestUtil.newUniqueName(tenantNotExist + "/default"); + final String topicNameStr = BrokerTestUtil.newUniqueName(topicDomain.toString() + "://" + namespaceNotExist + "/tp"); + PulsarClientImpl[] clientArray = getClientsToTest(false); + for (PulsarClientImpl client : clientArray) { + try { + PartitionedTopicMetadata topicMetadata = client + .getPartitionedTopicMetadata(topicNameStr, true, true) + .join(); + log.info("Get topic metadata: {}", topicMetadata.partitions); + fail("Expected a not found ex"); + } catch (Exception ex) { + Throwable unwrapEx = FutureUtil.unwrapCompletionException(ex); + assertTrue(unwrapEx instanceof PulsarClientException.BrokerMetadataException || + unwrapEx instanceof PulsarClientException.TopicDoesNotExistException); + } + } + // Verify: lookup semaphore has been releases. + Awaitility.await().untilAsserted(() -> { + assertEquals(getLookupRequestPermits(), lookupPermitsBefore); + }); + } }
