From e75cd3b0476a981ff33ff7c454dc675d9b1adce5 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Mon, 20 May 2024 22:06:36 +0800 Subject: [PATCH 1/9] [fix] [client] PIP-344 Do not create partitioned metadata when calling pulsarClient.getPartitionsForTopic(topicName) --- .../pulsar/broker/service/ServerCnx.java | 49 +++ .../admin/GetPartitionMetadataTest.java | 289 ++++++++++++++++++ .../broker/admin/TopicAutoCreationTest.java | 3 +- .../broker/service/BrokerServiceTest.java | 10 +- .../service/BrokerServiceThrottlingTest.java | 2 +- .../pulsar/broker/service/ServerCnxTest.java | 2 +- .../buffer/TransactionLowWaterMarkTest.java | 4 +- .../client/api/BrokerServiceLookupTest.java | 2 +- .../pulsar/client/api/PulsarClient.java | 22 +- .../client/impl/BinaryProtoLookupService.java | 13 +- .../client/impl/ConsumerBuilderImpl.java | 4 +- .../pulsar/client/impl/HttpLookupService.java | 6 +- .../pulsar/client/impl/LookupService.java | 17 +- .../client/impl/MultiTopicsConsumerImpl.java | 2 +- .../pulsar/client/impl/PulsarClientImpl.java | 32 +- .../TransactionCoordinatorClientImpl.java | 3 +- .../impl/MultiTopicsConsumerImplTest.java | 12 +- .../client/impl/PulsarClientImplTest.java | 3 +- .../pulsar/common/protocol/Commands.java | 7 +- .../apache/pulsar/common/util/FutureUtil.java | 4 +- pulsar-common/src/main/proto/PulsarApi.proto | 2 + .../proxy/server/LookupProxyHandler.java | 3 +- 22 files changed, 437 insertions(+), 54 deletions(-) create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/GetPartitionMetadataTest.java 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 59411aec0405f..ed2de1c6fd209 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 @@ -82,6 +82,8 @@ import org.apache.pulsar.broker.limiter.ConnectionController; import org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl; import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLookupData; +import org.apache.pulsar.broker.resources.NamespaceResources; +import org.apache.pulsar.broker.resources.TopicResources; import org.apache.pulsar.broker.service.BrokerServiceException.ConsumerBusyException; import org.apache.pulsar.broker.service.BrokerServiceException.ServerMetadataException; import org.apache.pulsar.broker.service.BrokerServiceException.ServiceUnitNotReadyException; @@ -607,6 +609,53 @@ protected void handlePartitionMetadataRequest(CommandPartitionedTopicMetadata pa isTopicOperationAllowed(topicName, TopicOperation.LOOKUP, authenticationData, originalAuthData).thenApply( isAuthorized -> { if (isAuthorized) { + // Get if exists, respond not found error if not exists. + if (!partitionMetadata.isMetadataAutoCreationEnabled()) { + final NamespaceResources namespaceResources = getBrokerService().pulsar().getPulsarResources() + .getNamespaceResources(); + final TopicResources topicResources = getBrokerService().pulsar().getPulsarResources() + .getTopicResources(); + namespaceResources.getPartitionedTopicResources() + .getPartitionedTopicMetadataAsync(topicName, false) + .thenAccept(metadata -> { + if (metadata.isPresent()) { + commandSender.sendPartitionMetadataResponse(metadata.get().partitions, requestId); + lookupSemaphore.release(); + return; + } + if (topicName.isPersistent()) { + topicResources.persistentTopicExists(topicName).thenAccept(exists -> { + if (exists) { + commandSender.sendPartitionMetadataResponse(0, requestId); + lookupSemaphore.release(); + return; + } + writeAndFlush(Commands.newPartitionMetadataResponse(ServerError.TopicNotFound, + "", requestId)); + lookupSemaphore.release(); + }).exceptionally(ex -> { + log.error("{} {} Failed to get partition metadata", topicName, + ServerCnx.this.toString(), ex); + writeAndFlush( + Commands.newPartitionMetadataResponse(ServerError.MetadataError, + "Failed to check partition metadata", + requestId)); + lookupSemaphore.release(); + return null; + }); + } + }).exceptionally(ex -> { + log.error("{} {} Failed to get partition metadata", topicName, + ServerCnx.this.toString(), ex); + writeAndFlush( + Commands.newPartitionMetadataResponse(ServerError.MetadataError, + "Failed to get partition metadata", + requestId)); + lookupSemaphore.release(); + return null; + }); + } + // Get if exists, create a new one if not exists. unsafeGetPartitionedTopicMetadataAsync(getBrokerService().pulsar(), topicName) .handle((metadata, ex) -> { if (ex == null) { 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 new file mode 100644 index 0000000000000..1b910974a1a92 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/GetPartitionMetadataTest.java @@ -0,0 +1,289 @@ +/* + * 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.pulsar.broker.admin; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertTrue; +import static org.testng.Assert.fail; +import java.util.List; +import java.util.Optional; +import lombok.extern.slf4j.Slf4j; +import org.apache.pulsar.broker.BrokerTestUtil; +import org.apache.pulsar.client.api.ProducerConsumerBase; +import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.client.impl.LookupService; +import org.apache.pulsar.client.impl.PulsarClientImpl; +import org.apache.pulsar.common.naming.TopicName; +import org.apache.pulsar.common.partition.PartitionedTopicMetadata; +import org.apache.pulsar.common.policies.data.TopicType; +import org.apache.pulsar.common.util.FutureUtil; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.DataProvider; +import org.testng.annotations.Test; + +@Test(groups = "broker-admin") +@Slf4j +public class GetPartitionMetadataTest extends ProducerConsumerBase { + + private static final String DEFAULT_NS = "public/default"; + + private PulsarClientImpl clientWithHttpLookup; + private PulsarClientImpl clientWitBinaryLookup; + + @Override + protected void setup() throws Exception { + super.internalSetup(); + super.producerBaseSetup(); + clientWithHttpLookup = + (PulsarClientImpl) PulsarClient.builder().serviceUrl(pulsar.getWebServiceAddress()).build(); + clientWitBinaryLookup = + (PulsarClientImpl) PulsarClient.builder().serviceUrl(pulsar.getBrokerServiceUrl()).build(); + } + + @Override + @AfterMethod(alwaysRun = true) + protected void cleanup() throws Exception { + super.internalCleanup(); + if (clientWithHttpLookup != null) { + clientWithHttpLookup.close(); + } + if (clientWitBinaryLookup != null) { + clientWitBinaryLookup.close(); + } + } + + @Override + protected void doInitConf() throws Exception { + super.doInitConf(); + } + + private LookupService getLookupService(boolean isUsingHttpLookup) { + if (isUsingHttpLookup) { + return clientWithHttpLookup.getLookup(); + } else { + return clientWitBinaryLookup.getLookup(); + } + } + + @Test + public void testAutoCreatingMetadataWhenCallingOldAPI() throws Exception { + conf.setAllowAutoTopicCreationType(TopicType.PARTITIONED); + conf.setDefaultNumPartitions(3); + conf.setAllowAutoTopicCreation(true); + setup(); + + // HTTP client. + final String tp1 = BrokerTestUtil.newUniqueName("persistent://" + DEFAULT_NS + "/tp"); + clientWithHttpLookup.getPartitionsForTopic(tp1).join(); + Optional metadata1 = pulsar.getPulsarResources().getNamespaceResources() + .getPartitionedTopicResources() + .getPartitionedTopicMetadataAsync(TopicName.get(tp1), true).join(); + assertTrue(metadata1.isPresent()); + assertEquals(metadata1.get().partitions, 3); + + // Binary client. + final String tp2 = BrokerTestUtil.newUniqueName("persistent://" + DEFAULT_NS + "/tp"); + clientWitBinaryLookup.getPartitionsForTopic(tp2).join(); + Optional metadata2 = pulsar.getPulsarResources().getNamespaceResources() + .getPartitionedTopicResources() + .getPartitionedTopicMetadataAsync(TopicName.get(tp2), true).join(); + assertTrue(metadata2.isPresent()); + assertEquals(metadata2.get().partitions, 3); + + // Cleanup. + admin.topics().deletePartitionedTopic(tp1, false); + admin.topics().deletePartitionedTopic(tp2, false); + } + + @DataProvider(name = "autoCreationParamsAll") + public Object[][] autoCreationParamsAll(){ + return new Object[][]{ + // configAllowAutoTopicCreation, paramCreateIfAutoCreationEnabled, isUsingHttpLookup. + {true, true, true}, + {true, true, false}, + {true, false, true}, + {true, false, false}, + {false, true, true}, + {false, true, false}, + {false, false, true}, + {false, false, false} + }; + } + + @Test(dataProvider = "autoCreationParamsAll") + public void testGetMetadataIfNonPartitionedTopicExists(boolean configAllowAutoTopicCreation, + boolean paramMetadataAutoCreationEnabled, + boolean isUsingHttpLookup) throws Exception { + conf.setAllowAutoTopicCreationType(TopicType.PARTITIONED); + conf.setDefaultNumPartitions(3); + conf.setAllowAutoTopicCreation(configAllowAutoTopicCreation); + setup(); + LookupService lookup = getLookupService(isUsingHttpLookup); + // Create topic. + final String topicNameStr = BrokerTestUtil.newUniqueName("persistent://" + DEFAULT_NS + "/tp"); + final TopicName topicName = TopicName.get(topicNameStr); + admin.topics().createNonPartitionedTopic(topicNameStr); + // Verify. + PulsarClient client = PulsarClient.builder().serviceUrl(pulsar.getBrokerServiceUrl()).build(); + PartitionedTopicMetadata response = + lookup.getPartitionedTopicMetadata(topicName, paramMetadataAutoCreationEnabled).join(); + assertEquals(response.partitions, 0); + List partitionedTopics = admin.topics().getPartitionedTopicList("public/default"); + assertFalse(partitionedTopics.contains(topicNameStr)); + List topicList = admin.topics().getList("public/default"); + for (int i = 0; i < 3; i++) { + assertFalse(topicList.contains(topicName.getPartition(i))); + } + // Cleanup. + client.close(); + admin.topics().delete(topicNameStr, false); + } + + @Test(dataProvider = "autoCreationParamsAll") + public void testGetMetadataIfPartitionedTopicExists(boolean configAllowAutoTopicCreation, + boolean paramMetadataAutoCreationEnabled, + boolean isUsingHttpLookup) throws Exception { + conf.setAllowAutoTopicCreationType(TopicType.PARTITIONED); + conf.setDefaultNumPartitions(3); + conf.setAllowAutoTopicCreation(configAllowAutoTopicCreation); + setup(); + LookupService lookup = getLookupService(isUsingHttpLookup); + // Create topic. + final String topicNameStr = BrokerTestUtil.newUniqueName("persistent://" + DEFAULT_NS + "/tp"); + final TopicName topicName = TopicName.get(topicNameStr); + admin.topics().createPartitionedTopic(topicNameStr, 3); + // Verify. + PulsarClient client = PulsarClient.builder().serviceUrl(pulsar.getBrokerServiceUrl()).build(); + PartitionedTopicMetadata response = + lookup.getPartitionedTopicMetadata(topicName, paramMetadataAutoCreationEnabled).join(); + assertEquals(response.partitions, 3); + List topicList = admin.topics().getList("public/default"); + assertFalse(topicList.contains(topicNameStr)); + // Cleanup. + client.close(); + admin.topics().deletePartitionedTopic(topicNameStr, false); + } + + @DataProvider(name = "clients") + public Object[][] clients(){ + return new Object[][]{ + // isUsingHttpLookup. + {true}, + {false} + }; + } + + @Test(dataProvider = "clients") + public void testAutoCreatePartitionedTopic(boolean isUsingHttpLookup) throws Exception { + conf.setAllowAutoTopicCreationType(TopicType.PARTITIONED); + conf.setDefaultNumPartitions(3); + conf.setAllowAutoTopicCreation(true); + setup(); + LookupService lookup = getLookupService(isUsingHttpLookup); + // Create topic. + final String topicNameStr = BrokerTestUtil.newUniqueName("persistent://" + DEFAULT_NS + "/tp"); + final TopicName topicName = TopicName.get(topicNameStr); + // Verify. + PulsarClient client = PulsarClient.builder().serviceUrl(pulsar.getBrokerServiceUrl()).build(); + PartitionedTopicMetadata response = lookup.getPartitionedTopicMetadata(topicName, true).join(); + assertEquals(response.partitions, 3); + List partitionedTopics = admin.topics().getPartitionedTopicList("public/default"); + assertTrue(partitionedTopics.contains(topicNameStr)); + List topicList = admin.topics().getList("public/default"); + assertFalse(topicList.contains(topicNameStr)); + for (int i = 0; i < 3; i++) { + // The API "getPartitionedTopicMetadata" only creates the partitioned metadata, it will not create the + // partitions. + assertFalse(topicList.contains(topicName.getPartition(i))); + } + // Cleanup. + client.close(); + admin.topics().deletePartitionedTopic(topicNameStr, false); + } + + @Test(dataProvider = "clients") + public void testAutoCreateNonPartitionedTopic(boolean isUsingHttpLookup) throws Exception { + conf.setAllowAutoTopicCreationType(TopicType.NON_PARTITIONED); + conf.setAllowAutoTopicCreation(true); + setup(); + LookupService lookup = getLookupService(isUsingHttpLookup); + // Create topic. + final String topicNameStr = BrokerTestUtil.newUniqueName("persistent://" + DEFAULT_NS + "/tp"); + final TopicName topicName = TopicName.get(topicNameStr); + // Verify. + PulsarClient client = PulsarClient.builder().serviceUrl(pulsar.getBrokerServiceUrl()).build(); + PartitionedTopicMetadata response = lookup.getPartitionedTopicMetadata(topicName, true).join(); + assertEquals(response.partitions, 0); + List partitionedTopics = admin.topics().getPartitionedTopicList("public/default"); + assertFalse(partitionedTopics.contains(topicNameStr)); + List topicList = admin.topics().getList("public/default"); + assertFalse(topicList.contains(topicNameStr)); + // Cleanup. + client.close(); + } + + @DataProvider(name = "autoCreationParamsNotAllow") + public Object[][] autoCreationParamsNotAllow(){ + return new Object[][]{ + // configAllowAutoTopicCreation, paramCreateIfAutoCreationEnabled, isUsingHttpLookup. + {true, false, true}, + {true, false, false}, + {false, true, true}, + {false, true, false}, + {false, false, true}, + {false, false, false}, + }; + } + + @Test(dataProvider = "autoCreationParamsNotAllow") + public void testGetMetadataIfNotAllowedCreate(boolean configAllowAutoTopicCreation, + boolean paramMetadataAutoCreationEnabled, + boolean isUsingHttpLookup) throws Exception { + conf.setAllowAutoTopicCreationType(TopicType.PARTITIONED); + conf.setDefaultNumPartitions(3); + conf.setAllowAutoTopicCreation(configAllowAutoTopicCreation); + setup(); + LookupService lookup = getLookupService(isUsingHttpLookup); + // Define topic. + final String topicNameStr = BrokerTestUtil.newUniqueName("persistent://" + DEFAULT_NS + "/tp"); + final TopicName topicName = TopicName.get(topicNameStr); + // Verify. + PulsarClient client = PulsarClient.builder().serviceUrl(pulsar.getBrokerServiceUrl()).build(); + try { + lookup.getPartitionedTopicMetadata(TopicName.get(topicNameStr), paramMetadataAutoCreationEnabled).join(); + fail("Expect a not found exception"); + } catch (Exception e) { + log.warn("", e); + Throwable unwrapEx = FutureUtil.unwrapCompletionException(e); + assertTrue(unwrapEx instanceof PulsarClientException.TopicDoesNotExistException + || unwrapEx instanceof PulsarClientException.NotFoundException); + } + List partitionedTopics = admin.topics().getPartitionedTopicList("public/default"); + assertFalse(partitionedTopics.contains(topicNameStr)); + List topicList = admin.topics().getList("public/default"); + assertFalse(topicList.contains(topicNameStr)); + for (int i = 0; i < 3; i++) { + assertFalse(topicList.contains(topicName.getPartition(i))); + } + // Cleanup. + client.close(); + } +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAutoCreationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAutoCreationTest.java index bb4a23bf24bd9..55601ad4c6b1d 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAutoCreationTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAutoCreationTest.java @@ -19,6 +19,7 @@ package org.apache.pulsar.broker.admin; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; import static org.testng.Assert.assertEquals; @@ -133,7 +134,7 @@ public void testPartitionedTopicAutoCreationForbiddenDuringNamespaceDeletion() // we want to skip the "lookup" phase, because it is blocked by the HTTP API LookupService mockLookup = mock(LookupService.class); ((PulsarClientImpl) pulsarClient).setLookup(mockLookup); - when(mockLookup.getPartitionedTopicMetadata(any())).thenAnswer( + when(mockLookup.getPartitionedTopicMetadata(any(), anyBoolean())).thenAnswer( i -> CompletableFuture.completedFuture(new PartitionedTopicMetadata(0))); when(mockLookup.getBroker(any())).thenAnswer(ignored -> { InetSocketAddress brokerAddress = diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java index 1818163cd340e..be1221b7fab41 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java @@ -1037,12 +1037,12 @@ protected void handlePartitionResponse(CommandPartitionedTopicMetadataResponse l // for PMR // 2 lookup will succeed long reqId1 = reqId++; - ByteBuf request1 = Commands.newPartitionMetadataRequest(topicName, reqId1); + ByteBuf request1 = Commands.newPartitionMetadataRequest(topicName, reqId1, true); CompletableFuture f1 = pool.getConnection(resolver.resolveHost()) .thenCompose(clientCnx -> clientCnx.newLookup(request1, reqId1)); long reqId2 = reqId++; - ByteBuf request2 = Commands.newPartitionMetadataRequest(topicName, reqId2); + ByteBuf request2 = Commands.newPartitionMetadataRequest(topicName, reqId2, true); CompletableFuture f2 = pool.getConnection(resolver.resolveHost()) .thenCompose(clientCnx -> { CompletableFuture future = clientCnx.newLookup(request2, reqId2); @@ -1057,17 +1057,17 @@ protected void handlePartitionResponse(CommandPartitionedTopicMetadataResponse l // 3 lookup will fail latchRef.set(new CountDownLatch(1)); long reqId3 = reqId++; - ByteBuf request3 = Commands.newPartitionMetadataRequest(topicName, reqId3); + ByteBuf request3 = Commands.newPartitionMetadataRequest(topicName, reqId3, true); f1 = pool.getConnection(resolver.resolveHost()) .thenCompose(clientCnx -> clientCnx.newLookup(request3, reqId3)); long reqId4 = reqId++; - ByteBuf request4 = Commands.newPartitionMetadataRequest(topicName, reqId4); + ByteBuf request4 = Commands.newPartitionMetadataRequest(topicName, reqId4, true); f2 = pool.getConnection(resolver.resolveHost()) .thenCompose(clientCnx -> clientCnx.newLookup(request4, reqId4)); long reqId5 = reqId++; - ByteBuf request5 = Commands.newPartitionMetadataRequest(topicName, reqId5); + ByteBuf request5 = Commands.newPartitionMetadataRequest(topicName, reqId5, true); CompletableFuture f3 = pool.getConnection(resolver.resolveHost()) .thenCompose(clientCnx -> { CompletableFuture future = clientCnx.newLookup(request5, reqId5); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceThrottlingTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceThrottlingTest.java index 312bfe0fc8ad7..c6a94833c4c62 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceThrottlingTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceThrottlingTest.java @@ -198,7 +198,7 @@ public void testLookupThrottlingForClientByBroker() throws Exception { for (int i = 0; i < totalConsumers; i++) { long reqId = 0xdeadbeef + i; Future f = executor.submit(() -> { - ByteBuf request = Commands.newPartitionMetadataRequest(topicName, reqId); + ByteBuf request = Commands.newPartitionMetadataRequest(topicName, reqId, true); pool.getConnection(resolver.resolveHost()) .thenCompose(clientCnx -> clientCnx.newLookup(request, reqId)) .get(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java index 1cb2f76c5e2b2..5387bc4998c6e 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java @@ -3571,7 +3571,7 @@ public void handlePartitionMetadataRequestWithServiceNotReady() throws Exception doReturn(false).when(pulsar).isRunning(); assertTrue(channel.isActive()); - ByteBuf clientCommand = Commands.newPartitionMetadataRequest(successTopicName, 1); + ByteBuf clientCommand = Commands.newPartitionMetadataRequest(successTopicName, 1, true); channel.writeInbound(clientCommand); Object response = getResponse(); assertTrue(response instanceof CommandPartitionedTopicMetadataResponse); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/TransactionLowWaterMarkTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/TransactionLowWaterMarkTest.java index aa7240a59f9c0..6e121aca3816f 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/TransactionLowWaterMarkTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/TransactionLowWaterMarkTest.java @@ -148,7 +148,7 @@ public void testTransactionBufferLowWaterMark() throws Exception { PartitionedTopicMetadata partitionedTopicMetadata = ((PulsarClientImpl) pulsarClient).getLookup() - .getPartitionedTopicMetadata(SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN).get(); + .getPartitionedTopicMetadata(SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN, false).get(); Transaction lowWaterMarkTxn = null; for (int i = 0; i < partitionedTopicMetadata.partitions; i++) { lowWaterMarkTxn = pulsarClient.newTransaction() @@ -253,7 +253,7 @@ public void testPendingAckLowWaterMark() throws Exception { PartitionedTopicMetadata partitionedTopicMetadata = ((PulsarClientImpl) pulsarClient).getLookup() - .getPartitionedTopicMetadata(SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN).get(); + .getPartitionedTopicMetadata(SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN, false).get(); Transaction lowWaterMarkTxn = null; for (int i = 0; i < partitionedTopicMetadata.partitions; i++) { lowWaterMarkTxn = pulsarClient.newTransaction() diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java index 0ad0b01dc1c99..336728f279eda 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java @@ -931,7 +931,7 @@ public void testMergeGetPartitionedMetadataRequests() throws Exception { // Verify the request is works after merge the requests. List> futures = new ArrayList<>(); for (int i = 0; i < 100; i++) { - futures.add(lookupService.getPartitionedTopicMetadata(TopicName.get(tpName))); + futures.add(lookupService.getPartitionedTopicMetadata(TopicName.get(tpName), false)); } for (CompletableFuture future : futures) { assertEquals(future.join().partitions, topicPartitions); diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClient.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClient.java index 78952fcaed8b3..09e6d1babaf20 100644 --- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClient.java +++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClient.java @@ -308,14 +308,32 @@ static ClientBuilder builder() { * *

This can be used to discover the partitions and create {@link Reader}, {@link Consumer} or {@link Producer} * instances directly on a particular partition. - * + * @Deprecated it is not suggested to use now; please use {@link #getPartitionsForTopic(String, boolean)}. * @param topic * the topic name * @return a future that will yield a list of the topic partitions or {@link PulsarClientException} if there was any * error in the operation. + * * @since 2.3.0 */ - CompletableFuture> getPartitionsForTopic(String topic); + @Deprecated + default CompletableFuture> getPartitionsForTopic(String topic) { + return getPartitionsForTopic(topic, true); + } + + /** + * 1. Get the partitions if the topic exists. Return "[{partition-0}, {partition-1}....{partition-n}}]" if a + * partitioned topic exists; return "[{topic}]" if a non-partitioned topic exists. + * 2. When {@param metadataAutoCreationEnabled} is "false", neither the partitioned topic nor non-partitioned + * topic does not exist. You will get an {@link PulsarClientException.NotFoundException}. + * 2-1. You will get a {@link PulsarClientException.NotSupportedException} with metadataAutoCreationEnabled=false + * on an old broker version which does not support getting partitions without partitioned metadata auto-creation. + * 3. When {@param metadataAutoCreationEnabled} is "true," it will trigger an auto-creation for this topic(using + * the default topic auto-creation strategy you set for the broker), and the corresponding result is returned. + * For the result, see case 1. + * @version 3.3.0. + */ + CompletableFuture> getPartitionsForTopic(String topic, boolean metadataAutoCreationEnabled); /** * Close the PulsarClient and release all the resources. diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BinaryProtoLookupService.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BinaryProtoLookupService.java index 8eedb3250cdf5..080a04100e904 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BinaryProtoLookupService.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BinaryProtoLookupService.java @@ -146,12 +146,14 @@ public CompletableFuture getBroker(TopicName topicName) { * calls broker binaryProto-lookup api to get metadata of partitioned-topic. * */ - public CompletableFuture getPartitionedTopicMetadata(TopicName topicName) { + @Override + public CompletableFuture getPartitionedTopicMetadata( + TopicName topicName, boolean metadataAutoCreationEnabled) { final MutableObject newFutureCreated = new MutableObject<>(); try { return partitionedMetadataInProgress.computeIfAbsent(topicName, tpName -> { - CompletableFuture newFuture = - getPartitionedTopicMetadata(serviceNameResolver.resolveHost(), topicName); + CompletableFuture newFuture = getPartitionedTopicMetadata( + serviceNameResolver.resolveHost(), topicName, metadataAutoCreationEnabled); newFutureCreated.setValue(newFuture); return newFuture; }); @@ -248,14 +250,15 @@ private CompletableFuture findBroker(InetSocketAddress socket } private CompletableFuture getPartitionedTopicMetadata(InetSocketAddress socketAddress, - TopicName topicName) { + TopicName topicName, boolean metadataAutoCreationEnabled) { long startTime = System.nanoTime(); CompletableFuture partitionFuture = new CompletableFuture<>(); client.getCnxPool().getConnection(socketAddress).thenAccept(clientCnx -> { long requestId = client.newRequestId(); - ByteBuf request = Commands.newPartitionMetadataRequest(topicName.toString(), requestId); + ByteBuf request = Commands.newPartitionMetadataRequest(topicName.toString(), requestId, + metadataAutoCreationEnabled); clientCnx.newLookup(request, requestId).whenComplete((r, t) -> { if (t != null) { histoGetTopicMetadata.recordFailure(System.nanoTime() - startTime); diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java index 7686d0072cffb..7735f66e7838a 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java @@ -136,9 +136,9 @@ public CompletableFuture> subscribeAsync() { if (deadLetterPolicy == null || StringUtils.isBlank(deadLetterPolicy.getRetryLetterTopic()) || StringUtils.isBlank(deadLetterPolicy.getDeadLetterTopic())) { CompletableFuture retryLetterTopicMetadata = - client.getPartitionedTopicMetadata(oldRetryLetterTopic); + client.getPartitionedTopicMetadata(oldRetryLetterTopic, true); CompletableFuture deadLetterTopicMetadata = - client.getPartitionedTopicMetadata(oldDeadLetterTopic); + client.getPartitionedTopicMetadata(oldDeadLetterTopic, true); applyDLQConfig = CompletableFuture.allOf(retryLetterTopicMetadata, deadLetterTopicMetadata) .thenAccept(__ -> { String retryLetterTopic = topicFirst + "-" + conf.getSubscriptionName() diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpLookupService.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpLookupService.java index 8158b6d979efd..1e568cf6eebdd 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpLookupService.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpLookupService.java @@ -139,12 +139,14 @@ public CompletableFuture getBroker(TopicName topicName) { } @Override - public CompletableFuture getPartitionedTopicMetadata(TopicName topicName) { + public CompletableFuture getPartitionedTopicMetadata( + TopicName topicName, boolean metadataAutoCreationEnabled) { long startTime = System.nanoTime(); String format = topicName.isV2() ? "admin/v2/%s/partitions" : "admin/%s/partitions"; CompletableFuture httpFuture = httpClient.get( - String.format(format, topicName.getLookupName()) + "?checkAllowAutoCreation=true", + String.format(format, topicName.getLookupName()) + "?checkAllowAutoCreation=" + + metadataAutoCreationEnabled, PartitionedTopicMetadata.class); httpFuture.thenRun(() -> { diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/LookupService.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/LookupService.java index 4d59d6591dbb8..f21ef1bffcf6f 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/LookupService.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/LookupService.java @@ -59,12 +59,19 @@ public interface LookupService extends AutoCloseable { CompletableFuture getBroker(TopicName topicName); /** - * Returns {@link PartitionedTopicMetadata} for a given topic. - * - * @param topicName topic-name - * @return + * 1.Get the partitions if the topic exists. Return "{partition: n}" if a partitioned topic exists; + * return "{partition: 0}" if a non-partitioned topic exists. + * 2. When {@param metadataAutoCreationEnabled} is "false," neither partitioned topic nor non-partitioned topic + * does not exist. You will get an {@link PulsarClientException.NotFoundException}. + * 2-1. You will get a {@link PulsarClientException.NotSupportedException} if the broker's version is an older + * one that does not support this feature and the Pulsar client is using a binary protocol "serviceUrl". + * 3.When {@param metadataAutoCreationEnabled} is "true," it will trigger an auto-creation for this topic(using + * the default topic auto-creation strategy you set for the broker), and the corresponding result is returned. + * For the result, see case 1. + * @version 3.3.0. */ - CompletableFuture getPartitionedTopicMetadata(TopicName topicName); + CompletableFuture getPartitionedTopicMetadata(TopicName topicName, + boolean metadataAutoCreationEnabled); /** * Returns current SchemaInfo {@link SchemaInfo} for a given topic. diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java index 20fd03d6a285f..8047e05351ac1 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java @@ -954,7 +954,7 @@ public CompletableFuture subscribeAsync(String topicName, boolean createTo CompletableFuture subscribeResult = new CompletableFuture<>(); - client.getPartitionedTopicMetadata(topicName) + client.getPartitionedTopicMetadata(topicName, true) .thenAccept(metadata -> subscribeTopicPartitions(subscribeResult, fullTopicName, metadata.partitions, createTopicIfDoesNotExist)) .exceptionally(ex1 -> { diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java index bd1b9564f932c..8aa16ef0e5f5f 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java @@ -386,7 +386,7 @@ private CompletableFuture> createProducerAsync(String topic, ProducerInterceptors interceptors) { CompletableFuture> producerCreatedFuture = new CompletableFuture<>(); - getPartitionedTopicMetadata(topic).thenAccept(metadata -> { + getPartitionedTopicMetadata(topic, true).thenAccept(metadata -> { if (log.isDebugEnabled()) { log.debug("[{}] Received topic metadata. partitions: {}", topic, metadata.partitions); } @@ -528,7 +528,7 @@ private CompletableFuture> doSingleTopicSubscribeAsync(ConsumerC String topic = conf.getSingleTopic(); - getPartitionedTopicMetadata(topic).thenAccept(metadata -> { + getPartitionedTopicMetadata(topic, true).thenAccept(metadata -> { if (log.isDebugEnabled()) { log.debug("[{}] Received topic metadata. partitions: {}", topic, metadata.partitions); } @@ -668,7 +668,7 @@ protected CompletableFuture> createSingleTopicReaderAsync( CompletableFuture> readerFuture = new CompletableFuture<>(); - getPartitionedTopicMetadata(topic).thenAccept(metadata -> { + getPartitionedTopicMetadata(topic, true).thenAccept(metadata -> { if (log.isDebugEnabled()) { log.debug("[{}] Received topic metadata. partitions: {}", topic, metadata.partitions); } @@ -1068,11 +1068,8 @@ public LookupService createLookup(String url) throws PulsarClientException { } } - public CompletableFuture getNumberOfPartitions(String topic) { - return getPartitionedTopicMetadata(topic).thenApply(metadata -> metadata.partitions); - } - - public CompletableFuture getPartitionedTopicMetadata(String topic) { + public CompletableFuture getPartitionedTopicMetadata( + String topic, boolean metadataAutoCreationEnabled) { CompletableFuture metadataFuture = new CompletableFuture<>(); @@ -1085,7 +1082,7 @@ public CompletableFuture getPartitionedTopicMetadata(S .setMax(conf.getMaxBackoffIntervalNanos(), TimeUnit.NANOSECONDS) .create(); getPartitionedTopicMetadata(topicName, backoff, opTimeoutMs, - metadataFuture, new ArrayList<>()); + metadataFuture, new ArrayList<>(), metadataAutoCreationEnabled); } catch (IllegalArgumentException e) { return FutureUtil.failedFuture(new PulsarClientException.InvalidConfigurationException(e.getMessage())); } @@ -1096,15 +1093,19 @@ private void getPartitionedTopicMetadata(TopicName topicName, Backoff backoff, AtomicLong remainingTime, CompletableFuture future, - List previousExceptions) { + List previousExceptions, + boolean metadataAutoCreationEnabled) { long startTime = System.nanoTime(); - lookup.getPartitionedTopicMetadata(topicName).thenAccept(future::complete).exceptionally(e -> { + CompletableFuture queryFuture = + lookup.getPartitionedTopicMetadata(topicName, metadataAutoCreationEnabled); + queryFuture.thenAccept(future::complete).exceptionally(e -> { remainingTime.addAndGet(-1 * TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTime)); long nextDelay = Math.min(backoff.next(), remainingTime.get()); // skip retry scheduler when set lookup throttle in client or server side which will lead to // `TooManyRequestsException` boolean isLookupThrottling = !PulsarClientException.isRetriableError(e.getCause()) - || e.getCause() instanceof PulsarClientException.AuthenticationException; + || e.getCause() instanceof PulsarClientException.AuthenticationException + || e.getCause() instanceof PulsarClientException.NotFoundException; if (nextDelay <= 0 || isLookupThrottling) { PulsarClientException.setPreviousExceptions(e, previousExceptions); future.completeExceptionally(e); @@ -1116,15 +1117,16 @@ private void getPartitionedTopicMetadata(TopicName topicName, log.warn("[topic: {}] Could not get connection while getPartitionedTopicMetadata -- " + "Will try again in {} ms", topicName, nextDelay); remainingTime.addAndGet(-nextDelay); - getPartitionedTopicMetadata(topicName, backoff, remainingTime, future, previousExceptions); + getPartitionedTopicMetadata(topicName, backoff, remainingTime, future, previousExceptions, + metadataAutoCreationEnabled); }, nextDelay, TimeUnit.MILLISECONDS); return null; }); } @Override - public CompletableFuture> getPartitionsForTopic(String topic) { - return getPartitionedTopicMetadata(topic).thenApply(metadata -> { + public CompletableFuture> getPartitionsForTopic(String topic, boolean metadataAutoCreationEnabled) { + return getPartitionedTopicMetadata(topic, metadataAutoCreationEnabled).thenApply(metadata -> { if (metadata.partitions > 0) { TopicName topicName = TopicName.get(topic); List partitions = new ArrayList<>(metadata.partitions); diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/transaction/TransactionCoordinatorClientImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/transaction/TransactionCoordinatorClientImpl.java index 9e79fc203c225..499627f9c73f2 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/transaction/TransactionCoordinatorClientImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/transaction/TransactionCoordinatorClientImpl.java @@ -79,7 +79,8 @@ public void start() throws TransactionCoordinatorClientException { @Override public CompletableFuture startAsync() { if (STATE_UPDATER.compareAndSet(this, State.NONE, State.STARTING)) { - return pulsarClient.getLookup().getPartitionedTopicMetadata(SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN) + return pulsarClient.getLookup() + .getPartitionedTopicMetadata(SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN, true) .thenCompose(partitionMeta -> { List> connectFutureList = new ArrayList<>(); if (LOG.isDebugEnabled()) { diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImplTest.java index febec2bff3285..191124bb7b002 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImplTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImplTest.java @@ -22,6 +22,7 @@ import static org.apache.pulsar.client.impl.ClientTestFixtures.createExceptionFuture; import static org.apache.pulsar.client.impl.ClientTestFixtures.createPulsarClientMockWithMockedClientCnx; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -153,7 +154,8 @@ private MultiTopicsConsumerImpl createMultiTopicsConsumer( int completionDelayMillis = 100; Schema schema = Schema.BYTES; PulsarClientImpl clientMock = createPulsarClientMockWithMockedClientCnx(executorProvider, internalExecutor); - when(clientMock.getPartitionedTopicMetadata(any())).thenAnswer(invocation -> createDelayedCompletedFuture( + when(clientMock.getPartitionedTopicMetadata(any(), anyBoolean())) + .thenAnswer(invocation -> createDelayedCompletedFuture( new PartitionedTopicMetadata(), completionDelayMillis)); MultiTopicsConsumerImpl impl = new MultiTopicsConsumerImpl( clientMock, consumerConfData, executorProvider, @@ -201,7 +203,8 @@ public void testConsumerCleanupOnSubscribeFailure() { int completionDelayMillis = 10; Schema schema = Schema.BYTES; PulsarClientImpl clientMock = createPulsarClientMockWithMockedClientCnx(executorProvider, internalExecutor); - when(clientMock.getPartitionedTopicMetadata(any())).thenAnswer(invocation -> createExceptionFuture( + when(clientMock.getPartitionedTopicMetadata(any(), anyBoolean())) + .thenAnswer(invocation -> createExceptionFuture( new PulsarClientException.InvalidConfigurationException("a mock exception"), completionDelayMillis)); CompletableFuture> completeFuture = new CompletableFuture<>(); MultiTopicsConsumerImpl impl = new MultiTopicsConsumerImpl(clientMock, consumerConfData, @@ -237,7 +240,8 @@ public void testDontCheckForPartitionsUpdatesOnNonPartitionedTopics() throws Exc // Simulate non partitioned topics PartitionedTopicMetadata metadata = new PartitionedTopicMetadata(0); - when(clientMock.getPartitionedTopicMetadata(any())).thenReturn(CompletableFuture.completedFuture(metadata)); + when(clientMock.getPartitionedTopicMetadata(any(), anyBoolean())) + .thenReturn(CompletableFuture.completedFuture(metadata)); CompletableFuture> completeFuture = new CompletableFuture<>(); MultiTopicsConsumerImpl impl = new MultiTopicsConsumerImpl<>( @@ -248,7 +252,7 @@ public void testDontCheckForPartitionsUpdatesOnNonPartitionedTopics() throws Exc // getPartitionedTopicMetadata should have been called only the first time, for each of the 3 topics, // but not anymore since the topics are not partitioned. - verify(clientMock, times(3)).getPartitionedTopicMetadata(any()); + verify(clientMock, times(3)).getPartitionedTopicMetadata(any(), anyBoolean()); } } diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PulsarClientImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PulsarClientImplTest.java index 274b9b4f2d572..3e897ed89f287 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PulsarClientImplTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PulsarClientImplTest.java @@ -19,6 +19,7 @@ package org.apache.pulsar.client.impl; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.ArgumentMatchers.nullable; @@ -107,7 +108,7 @@ public void testConsumerIsClosed() throws Exception { nullable(String.class))) .thenReturn(CompletableFuture.completedFuture( new GetTopicsResult(Collections.emptyList(), null, false, true))); - when(lookup.getPartitionedTopicMetadata(any(TopicName.class))) + when(lookup.getPartitionedTopicMetadata(any(TopicName.class), anyBoolean())) .thenReturn(CompletableFuture.completedFuture(new PartitionedTopicMetadata())); when(lookup.getBroker(any())) .thenReturn(CompletableFuture.completedFuture(new LookupTopicResult( diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java index 8599ec2dd3475..cbee7f354c4df 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java @@ -190,6 +190,7 @@ private static void setFeatureFlags(FeatureFlags flags) { flags.setSupportsAuthRefresh(true); flags.setSupportsBrokerEntryMetadata(true); flags.setSupportsPartialProducer(true); + flags.setSupportsGetPartitionedMetadataWithoutAutoCreation(true); } public static ByteBuf newConnect(String authMethodName, String authData, int protocolVersion, String libVersion, @@ -910,11 +911,13 @@ public static ByteBuf newPartitionMetadataResponse(ServerError error, String err return serializeWithSize(newPartitionMetadataResponseCommand(error, errorMsg, requestId)); } - public static ByteBuf newPartitionMetadataRequest(String topic, long requestId) { + public static ByteBuf newPartitionMetadataRequest(String topic, long requestId, + boolean metadataAutoCreationEnabled) { BaseCommand cmd = localCmd(Type.PARTITIONED_METADATA); cmd.setPartitionMetadata() .setTopic(topic) - .setRequestId(requestId); + .setRequestId(requestId) + .setMetadataAutoCreationEnabled(metadataAutoCreationEnabled); return serializeWithSize(cmd); } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java index f6fcb12f35939..0628d494af3af 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java @@ -199,9 +199,9 @@ public static CompletableFuture failedFuture(Throwable t) { public static Throwable unwrapCompletionException(Throwable ex) { if (ex instanceof CompletionException) { - return ex.getCause(); + return unwrapCompletionException(ex.getCause()); } else if (ex instanceof ExecutionException) { - return ex.getCause(); + return unwrapCompletionException(ex.getCause()); } else { return ex; } diff --git a/pulsar-common/src/main/proto/PulsarApi.proto b/pulsar-common/src/main/proto/PulsarApi.proto index 387e4e3ff679d..5a7eb582eb5b1 100644 --- a/pulsar-common/src/main/proto/PulsarApi.proto +++ b/pulsar-common/src/main/proto/PulsarApi.proto @@ -300,6 +300,7 @@ message FeatureFlags { optional bool supports_broker_entry_metadata = 2 [default = false]; optional bool supports_partial_producer = 3 [default = false]; optional bool supports_topic_watchers = 4 [default = false]; + optional bool supports_get_partitioned_metadata_without_auto_creation = 5 [default = false]; } message CommandConnected { @@ -413,6 +414,7 @@ message CommandPartitionedTopicMetadata { // to the proxy. optional string original_auth_data = 4; optional string original_auth_method = 5; + optional bool metadata_auto_creation_enabled = 6 [default = true]; } message CommandPartitionedTopicMetadataResponse { diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/LookupProxyHandler.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/LookupProxyHandler.java index f76adadcc3e0a..03975e153acb6 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/LookupProxyHandler.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/LookupProxyHandler.java @@ -241,7 +241,8 @@ private void handlePartitionMetadataResponse(CommandPartitionedTopicMetadata par // Connected to backend broker long requestId = proxyConnection.newRequestId(); ByteBuf command; - command = Commands.newPartitionMetadataRequest(topicName.toString(), requestId); + command = Commands.newPartitionMetadataRequest(topicName.toString(), requestId, + partitionMetadata.isMetadataAutoCreationEnabled()); clientCnx.newLookup(command, requestId).whenComplete((r, t) -> { if (t != null) { log.warn("[{}] failed to get Partitioned metadata : {}", topicName.toString(), From 481dc8a21cf291f52c20037c690bdec0c245db28 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Tue, 21 May 2024 01:18:31 +0800 Subject: [PATCH 2/9] address commenyts --- .../admin/impl/PersistentTopicsBase.java | 27 +++- .../pulsar/broker/service/ServerCnx.java | 145 +++++++++--------- .../admin/GetPartitionMetadataTest.java | 2 + .../pulsar/client/impl/LookupService.java | 10 ++ 4 files changed, 107 insertions(+), 77 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java index 924b7be0855a8..265a9df9db82b 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java @@ -537,19 +537,30 @@ protected CompletableFuture internalGetPartitionedMeta boolean checkAllowAutoCreation) { return getPartitionedTopicMetadataAsync(topicName, authoritative, checkAllowAutoCreation) .thenCompose(metadata -> { - CompletableFuture ret; - if (metadata.partitions == 0 && !checkAllowAutoCreation) { + CompletableFuture ret = new CompletableFuture<>(); + if (metadata.partitions > 1) { + // Some clients does not support partitioned topic. + return internalValidateClientVersionAsync().thenApply(__ -> metadata); + } else if (metadata.partitions == 1) { + return CompletableFuture.completedFuture(metadata); + } else { + // metadata.partitions == 0 // The topic may be a non-partitioned topic, so check if it exists here. // However, when checkAllowAutoCreation is true, the client will create the topic if // it doesn't exist. In this case, `partitions == 0` means the automatically created topic // is a non-partitioned topic so we shouldn't check if the topic exists. - ret = internalCheckTopicExists(topicName); - } else if (metadata.partitions > 1) { - ret = internalValidateClientVersionAsync(); - } else { - ret = CompletableFuture.completedFuture(null); + return pulsar().getBrokerService().isAllowAutoTopicCreationAsync(topicName) + .thenCompose(brokerAllowAutoTopicCreation -> { + if (checkAllowAutoCreation && brokerAllowAutoTopicCreation) { + // Whether it exists or not, auto create a non-partitioned topic by client. + return CompletableFuture.completedFuture(metadata); + } else { + // If it does not exist, response a Not Found error. + // Otherwise, response a non-partitioned metadata. + return internalCheckTopicExists(topicName).thenApply(__ -> metadata); + } + }); } - return ret.thenApply(__ -> metadata); }); } 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 ed2de1c6fd209..6ef7a8cda909b 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 @@ -610,81 +610,88 @@ protected void handlePartitionMetadataRequest(CommandPartitionedTopicMetadata pa isAuthorized -> { if (isAuthorized) { // Get if exists, respond not found error if not exists. - if (!partitionMetadata.isMetadataAutoCreationEnabled()) { - final NamespaceResources namespaceResources = getBrokerService().pulsar().getPulsarResources() - .getNamespaceResources(); - final TopicResources topicResources = getBrokerService().pulsar().getPulsarResources() - .getTopicResources(); - namespaceResources.getPartitionedTopicResources() - .getPartitionedTopicMetadataAsync(topicName, false) - .thenAccept(metadata -> { - if (metadata.isPresent()) { - commandSender.sendPartitionMetadataResponse(metadata.get().partitions, requestId); - lookupSemaphore.release(); - return; - } - if (topicName.isPersistent()) { - topicResources.persistentTopicExists(topicName).thenAccept(exists -> { - if (exists) { - commandSender.sendPartitionMetadataResponse(0, requestId); + getBrokerService().isAllowAutoTopicCreationAsync(topicName).thenAccept(brokerAllowAutoCreate -> { + boolean autoCreateIfNotExist = brokerAllowAutoCreate + && partitionMetadata.isMetadataAutoCreationEnabled(); + if (!autoCreateIfNotExist) { + final NamespaceResources namespaceResources = getBrokerService().pulsar().getPulsarResources() + .getNamespaceResources(); + final TopicResources topicResources = getBrokerService().pulsar().getPulsarResources() + .getTopicResources(); + namespaceResources.getPartitionedTopicResources() + .getPartitionedTopicMetadataAsync(topicName, false) + .thenAccept(metadata -> { + if (metadata.isPresent()) { + commandSender.sendPartitionMetadataResponse(metadata.get().partitions, requestId); + lookupSemaphore.release(); + return; + } + if (topicName.isPersistent()) { + topicResources.persistentTopicExists(topicName).thenAccept(exists -> { + if (exists) { + commandSender.sendPartitionMetadataResponse(0, requestId); + lookupSemaphore.release(); + return; + } + writeAndFlush(Commands.newPartitionMetadataResponse(ServerError.TopicNotFound, + "", requestId)); lookupSemaphore.release(); return; - } - writeAndFlush(Commands.newPartitionMetadataResponse(ServerError.TopicNotFound, - "", requestId)); - lookupSemaphore.release(); - }).exceptionally(ex -> { - log.error("{} {} Failed to get partition metadata", topicName, - ServerCnx.this.toString(), ex); - writeAndFlush( - Commands.newPartitionMetadataResponse(ServerError.MetadataError, - "Failed to check partition metadata", - requestId)); - lookupSemaphore.release(); - return null; - }); - } - }).exceptionally(ex -> { - log.error("{} {} Failed to get partition metadata", topicName, - ServerCnx.this.toString(), ex); - writeAndFlush( - Commands.newPartitionMetadataResponse(ServerError.MetadataError, - "Failed to get partition metadata", - requestId)); - lookupSemaphore.release(); - return null; - }); - } - // Get if exists, create a new one if not exists. - unsafeGetPartitionedTopicMetadataAsync(getBrokerService().pulsar(), topicName) - .handle((metadata, ex) -> { - if (ex == null) { - int partitions = metadata.partitions; - commandSender.sendPartitionMetadataResponse(partitions, requestId); - } else { - if (ex instanceof PulsarClientException) { - log.warn("Failed to authorize {} at [{}] on topic {} : {}", getRole(), - remoteAddress, topicName, ex.getMessage()); - commandSender.sendPartitionMetadataResponse(ServerError.AuthorizationError, - ex.getMessage(), requestId); + }).exceptionally(ex -> { + log.error("{} {} Failed to get partition metadata", topicName, + ServerCnx.this.toString(), ex); + writeAndFlush( + Commands.newPartitionMetadataResponse(ServerError.MetadataError, + "Failed to check partition metadata", + requestId)); + lookupSemaphore.release(); + return null; + }); + } + }).exceptionally(ex -> { + log.error("{} {} Failed to get partition metadata", topicName, + ServerCnx.this.toString(), ex); + writeAndFlush( + Commands.newPartitionMetadataResponse(ServerError.MetadataError, + "Failed to get partition metadata", + requestId)); + lookupSemaphore.release(); + return null; + }); + } else { + // Get if exists, create a new one if not exists. + unsafeGetPartitionedTopicMetadataAsync(getBrokerService().pulsar(), topicName) + .handle((metadata, ex) -> { + if (ex == null) { + int partitions = metadata.partitions; + commandSender.sendPartitionMetadataResponse(partitions, requestId); } else { - log.warn("Failed to get Partitioned Metadata [{}] {}: {}", remoteAddress, - topicName, ex.getMessage(), ex); - ServerError error = ServerError.ServiceNotReady; - if (ex instanceof RestException restException){ - int responseCode = restException.getResponse().getStatus(); - if (responseCode == NOT_FOUND.getStatusCode()){ - error = ServerError.TopicNotFound; - } else if (responseCode < INTERNAL_SERVER_ERROR.getStatusCode()){ - error = ServerError.MetadataError; + if (ex instanceof PulsarClientException) { + log.warn("Failed to authorize {} at [{}] on topic {} : {}", getRole(), + remoteAddress, topicName, ex.getMessage()); + 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 RestException restException){ + int responseCode = restException.getResponse().getStatus(); + if (responseCode == NOT_FOUND.getStatusCode()){ + error = ServerError.TopicNotFound; + } else if (responseCode < INTERNAL_SERVER_ERROR.getStatusCode()){ + error = ServerError.MetadataError; + } } + commandSender.sendPartitionMetadataResponse(error, ex.getMessage(), + requestId); } - commandSender.sendPartitionMetadataResponse(error, ex.getMessage(), requestId); } - } - lookupSemaphore.release(); - return null; - }); + lookupSemaphore.release(); + return null; + }); + } + }); } else { final String msg = "Client is not authorized to Get Partition Metadata"; log.warn("[{}] {} with role {} on topic {}", remoteAddress, msg, getPrincipal(), topicName); 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 1b910974a1a92..ae2584467be97 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 @@ -276,7 +276,9 @@ public void testGetMetadataIfNotAllowedCreate(boolean configAllowAutoTopicCreati assertTrue(unwrapEx instanceof PulsarClientException.TopicDoesNotExistException || unwrapEx instanceof PulsarClientException.NotFoundException); } + List partitionedTopics = admin.topics().getPartitionedTopicList("public/default"); + pulsar.getPulsarResources().getNamespaceResources().getPartitionedTopicResources().partitionedTopicExists(topicName); assertFalse(partitionedTopics.contains(topicNameStr)); List topicList = admin.topics().getList("public/default"); assertFalse(topicList.contains(topicNameStr)); diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/LookupService.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/LookupService.java index f21ef1bffcf6f..df9ee475ff6eb 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/LookupService.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/LookupService.java @@ -58,6 +58,16 @@ public interface LookupService extends AutoCloseable { */ CompletableFuture getBroker(TopicName topicName); + /** + * Returns {@link PartitionedTopicMetadata} for a given topic. + * Note: this method will try to create the topic partitioned metadata if it does not exist. + * @deprecated Please call {{@link #getPartitionedTopicMetadata(TopicName, boolean)}}. + */ + @Deprecated + default CompletableFuture getPartitionedTopicMetadata(TopicName topicName) { + return getPartitionedTopicMetadata(topicName, true); + } + /** * 1.Get the partitions if the topic exists. Return "{partition: n}" if a partitioned topic exists; * return "{partition: 0}" if a non-partitioned topic exists. From 172ed143f686cf8390a083da34a40ab430312105 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Tue, 21 May 2024 10:21:22 +0800 Subject: [PATCH 3/9] checkstyle --- .../org/apache/pulsar/broker/service/ServerCnx.java | 11 ++++++----- 1 file changed, 6 insertions(+), 5 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 6ef7a8cda909b..d01c8a8b5d501 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 @@ -614,15 +614,16 @@ protected void handlePartitionMetadataRequest(CommandPartitionedTopicMetadata pa boolean autoCreateIfNotExist = brokerAllowAutoCreate && partitionMetadata.isMetadataAutoCreationEnabled(); if (!autoCreateIfNotExist) { - final NamespaceResources namespaceResources = getBrokerService().pulsar().getPulsarResources() - .getNamespaceResources(); + final NamespaceResources namespaceResources = getBrokerService().pulsar() + .getPulsarResources().getNamespaceResources(); final TopicResources topicResources = getBrokerService().pulsar().getPulsarResources() .getTopicResources(); namespaceResources.getPartitionedTopicResources() .getPartitionedTopicMetadataAsync(topicName, false) .thenAccept(metadata -> { if (metadata.isPresent()) { - commandSender.sendPartitionMetadataResponse(metadata.get().partitions, requestId); + commandSender.sendPartitionMetadataResponse(metadata.get().partitions, + requestId); lookupSemaphore.release(); return; } @@ -633,8 +634,8 @@ protected void handlePartitionMetadataRequest(CommandPartitionedTopicMetadata pa lookupSemaphore.release(); return; } - writeAndFlush(Commands.newPartitionMetadataResponse(ServerError.TopicNotFound, - "", requestId)); + writeAndFlush(Commands.newPartitionMetadataResponse( + ServerError.TopicNotFound, "", requestId)); lookupSemaphore.release(); return; }).exceptionally(ex -> { From 486c7452312d2de729023b5ef480f65a1de0d010 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Tue, 21 May 2024 15:58:29 +0800 Subject: [PATCH 4/9] rollback the implementation of Section 2 of Goals of https://github.com/apache/pulsar/blob/master/pip/pip-344.md --- .../pulsar/broker/admin/impl/PersistentTopicsBase.java | 3 +-- .../java/org/apache/pulsar/broker/service/ServerCnx.java | 3 +-- .../pulsar/broker/admin/GetPartitionMetadataTest.java | 6 ++++-- .../broker/service/BrokerServiceAutoTopicCreationTest.java | 4 ++-- 4 files changed, 8 insertions(+), 8 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java index 265a9df9db82b..104a84d041d8a 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java @@ -537,7 +537,6 @@ protected CompletableFuture internalGetPartitionedMeta boolean checkAllowAutoCreation) { return getPartitionedTopicMetadataAsync(topicName, authoritative, checkAllowAutoCreation) .thenCompose(metadata -> { - CompletableFuture ret = new CompletableFuture<>(); if (metadata.partitions > 1) { // Some clients does not support partitioned topic. return internalValidateClientVersionAsync().thenApply(__ -> metadata); @@ -551,7 +550,7 @@ protected CompletableFuture internalGetPartitionedMeta // is a non-partitioned topic so we shouldn't check if the topic exists. return pulsar().getBrokerService().isAllowAutoTopicCreationAsync(topicName) .thenCompose(brokerAllowAutoTopicCreation -> { - if (checkAllowAutoCreation && brokerAllowAutoTopicCreation) { + if (checkAllowAutoCreation) { // Whether it exists or not, auto create a non-partitioned topic by client. return CompletableFuture.completedFuture(metadata); } else { 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 d01c8a8b5d501..2f8659dd853cf 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 @@ -611,8 +611,7 @@ protected void handlePartitionMetadataRequest(CommandPartitionedTopicMetadata pa if (isAuthorized) { // Get if exists, respond not found error if not exists. getBrokerService().isAllowAutoTopicCreationAsync(topicName).thenAccept(brokerAllowAutoCreate -> { - boolean autoCreateIfNotExist = brokerAllowAutoCreate - && partitionMetadata.isMetadataAutoCreationEnabled(); + boolean autoCreateIfNotExist = partitionMetadata.isMetadataAutoCreationEnabled(); if (!autoCreateIfNotExist) { final NamespaceResources namespaceResources = getBrokerService().pulsar() .getPulsarResources().getNamespaceResources(); 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 ae2584467be97..d625006d05529 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 @@ -246,10 +246,12 @@ public Object[][] autoCreationParamsNotAllow(){ // configAllowAutoTopicCreation, paramCreateIfAutoCreationEnabled, isUsingHttpLookup. {true, false, true}, {true, false, false}, - {false, true, true}, - {false, true, false}, {false, false, true}, {false, false, false}, + // These test cases are for the following PR. + // Which was described in the Motivation of https://github.com/apache/pulsar/pull/22206. + //{false, true, true}, + //{false, true, false}, }; } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java index 0a6cffc7685d4..ea5365bcf4b2c 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java @@ -566,13 +566,13 @@ public void testExtensibleLoadManagerImplInternalTopicAutoCreations() try { pulsarClient.newProducer().topic(ExtensibleLoadManagerImpl.BROKER_LOAD_DATA_STORE_TOPIC).create(); Assert.fail("Create should have failed."); - } catch (PulsarClientException.TopicDoesNotExistException e) { + } catch (PulsarClientException.TopicDoesNotExistException | PulsarClientException.NotFoundException e) { // expected } try { pulsarClient.newProducer().topic(ExtensibleLoadManagerImpl.TOP_BUNDLES_LOAD_DATA_STORE_TOPIC).create(); Assert.fail("Create should have failed."); - } catch (PulsarClientException.TopicDoesNotExistException e) { + } catch (PulsarClientException.TopicDoesNotExistException | PulsarClientException.NotFoundException e) { // expected } From b79edb1f3444ba3f4a41630735c491ad229d2726 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Tue, 21 May 2024 21:58:13 +0800 Subject: [PATCH 5/9] address comments --- .../admin/impl/PersistentTopicsBase.java | 11 +- .../pulsar/broker/service/ServerCnx.java | 44 ++--- .../admin/GetPartitionMetadataTest.java | 175 +++++++++++++++--- 3 files changed, 186 insertions(+), 44 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java index 104a84d041d8a..c93185e78f75b 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java @@ -556,7 +556,16 @@ protected CompletableFuture internalGetPartitionedMeta } else { // If it does not exist, response a Not Found error. // Otherwise, response a non-partitioned metadata. - return internalCheckTopicExists(topicName).thenApply(__ -> metadata); + if (topicName.isPersistent()) { + return internalCheckTopicExists(topicName).thenApply(__ -> metadata); + } else { + // Regarding non-persistent topic, we do not know whether it exists or not. + // Just return a non-partitioned metadata if partitioned metadata does not + // exist. + // Broker will respond a not found error when doing subscribing or producing if + // broker not allow to auto create topics. + return CompletableFuture.completedFuture(metadata); + } } }); } 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 2f8659dd853cf..085e5a1c52b4d 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 @@ -619,24 +619,25 @@ protected void handlePartitionMetadataRequest(CommandPartitionedTopicMetadata pa .getTopicResources(); namespaceResources.getPartitionedTopicResources() .getPartitionedTopicMetadataAsync(topicName, false) - .thenAccept(metadata -> { - if (metadata.isPresent()) { + .handle((metadata, getMetadataEx) -> { + if (getMetadataEx != null) { + log.error("{} {} Failed to get partition metadata", topicName, + ServerCnx.this.toString(), getMetadataEx); + writeAndFlush( + Commands.newPartitionMetadataResponse(ServerError.MetadataError, + "Failed to get partition metadata", + requestId)); + } else if (metadata.isPresent()) { commandSender.sendPartitionMetadataResponse(metadata.get().partitions, requestId); - lookupSemaphore.release(); - return; - } - if (topicName.isPersistent()) { + } else if (topicName.isPersistent()) { topicResources.persistentTopicExists(topicName).thenAccept(exists -> { if (exists) { commandSender.sendPartitionMetadataResponse(0, requestId); - lookupSemaphore.release(); return; } writeAndFlush(Commands.newPartitionMetadataResponse( ServerError.TopicNotFound, "", requestId)); - lookupSemaphore.release(); - return; }).exceptionally(ex -> { log.error("{} {} Failed to get partition metadata", topicName, ServerCnx.this.toString(), ex); @@ -644,24 +645,27 @@ protected void handlePartitionMetadataRequest(CommandPartitionedTopicMetadata pa Commands.newPartitionMetadataResponse(ServerError.MetadataError, "Failed to check partition metadata", requestId)); - lookupSemaphore.release(); return null; }); + } else { + // Regarding non-persistent topic, we do not know whether it exists or not. + // Just return a non-partitioned metadata if partitioned metadata does not + // exist. + // Broker will respond a not found error when doing subscribing or producing if + // broker not allow to auto create topics. + commandSender.sendPartitionMetadataResponse(0, requestId); } - }).exceptionally(ex -> { - log.error("{} {} Failed to get partition metadata", topicName, - ServerCnx.this.toString(), ex); - writeAndFlush( - Commands.newPartitionMetadataResponse(ServerError.MetadataError, - "Failed to get partition metadata", - requestId)); - lookupSemaphore.release(); return null; + }).whenComplete((ignore, ignoreEx) -> { + log.error("{} {} Failed to handle partition metadata request", topicName, + ServerCnx.this.toString(), ignoreEx); + lookupSemaphore.release(); }); } else { // Get if exists, create a new one if not exists. unsafeGetPartitionedTopicMetadataAsync(getBrokerService().pulsar(), topicName) - .handle((metadata, ex) -> { + .whenComplete((metadata, ex) -> { + lookupSemaphore.release(); if (ex == null) { int partitions = metadata.partitions; commandSender.sendPartitionMetadataResponse(partitions, requestId); @@ -687,8 +691,6 @@ protected void handlePartitionMetadataRequest(CommandPartitionedTopicMetadata pa requestId); } } - lookupSemaphore.release(); - return null; }); } }); 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 d625006d05529..147960ff03dde 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 @@ -24,6 +24,7 @@ import static org.testng.Assert.fail; import java.util.List; import java.util.Optional; +import java.util.concurrent.Semaphore; import lombok.extern.slf4j.Slf4j; import org.apache.pulsar.broker.BrokerTestUtil; import org.apache.pulsar.client.api.ProducerConsumerBase; @@ -31,10 +32,12 @@ import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.impl.LookupService; import org.apache.pulsar.client.impl.PulsarClientImpl; +import org.apache.pulsar.common.naming.TopicDomain; import org.apache.pulsar.common.naming.TopicName; import org.apache.pulsar.common.partition.PartitionedTopicMetadata; import org.apache.pulsar.common.policies.data.TopicType; import org.apache.pulsar.common.util.FutureUtil; +import org.awaitility.Awaitility; import org.testng.annotations.AfterMethod; import org.testng.annotations.DataProvider; import org.testng.annotations.Test; @@ -83,15 +86,26 @@ private LookupService getLookupService(boolean isUsingHttpLookup) { } } - @Test - public void testAutoCreatingMetadataWhenCallingOldAPI() throws Exception { + @DataProvider(name = "topicDomains") + public Object[][] topicDomains() { + return new Object[][]{ + {TopicDomain.persistent}, + {TopicDomain.non_persistent} + }; + } + + @Test(dataProvider = "topicDomains") + public void testAutoCreatingMetadataWhenCallingOldAPI(TopicDomain topicDomain) throws Exception { conf.setAllowAutoTopicCreationType(TopicType.PARTITIONED); conf.setDefaultNumPartitions(3); conf.setAllowAutoTopicCreation(true); setup(); + Semaphore semaphore = pulsar.getBrokerService().getLookupRequestSemaphore(); + int lookupPermitsBefore = semaphore.availablePermits(); + // HTTP client. - final String tp1 = BrokerTestUtil.newUniqueName("persistent://" + DEFAULT_NS + "/tp"); + final String tp1 = BrokerTestUtil.newUniqueName(topicDomain.value() + "://" + DEFAULT_NS + "/tp"); clientWithHttpLookup.getPartitionsForTopic(tp1).join(); Optional metadata1 = pulsar.getPulsarResources().getNamespaceResources() .getPartitionedTopicResources() @@ -100,7 +114,7 @@ public void testAutoCreatingMetadataWhenCallingOldAPI() throws Exception { assertEquals(metadata1.get().partitions, 3); // Binary client. - final String tp2 = BrokerTestUtil.newUniqueName("persistent://" + DEFAULT_NS + "/tp"); + final String tp2 = BrokerTestUtil.newUniqueName(topicDomain.value() + "://" + DEFAULT_NS + "/tp"); clientWitBinaryLookup.getPartitionsForTopic(tp2).join(); Optional metadata2 = pulsar.getPulsarResources().getNamespaceResources() .getPartitionedTopicResources() @@ -108,6 +122,12 @@ public void testAutoCreatingMetadataWhenCallingOldAPI() throws Exception { assertTrue(metadata2.isPresent()); assertEquals(metadata2.get().partitions, 3); + // Verify: lookup semaphore has been releases. + Awaitility.await().untilAsserted(() -> { + int lookupPermitsAfter = semaphore.availablePermits(); + assertEquals(lookupPermitsAfter, lookupPermitsBefore); + }); + // Cleanup. admin.topics().deletePartitionedTopic(tp1, false); admin.topics().deletePartitionedTopic(tp2, false); @@ -117,28 +137,41 @@ public void testAutoCreatingMetadataWhenCallingOldAPI() throws Exception { public Object[][] autoCreationParamsAll(){ return new Object[][]{ // configAllowAutoTopicCreation, paramCreateIfAutoCreationEnabled, isUsingHttpLookup. - {true, true, true}, - {true, true, false}, - {true, false, true}, - {true, false, false}, - {false, true, true}, - {false, true, false}, - {false, false, true}, - {false, false, false} + {true, true, true, TopicDomain.persistent}, + {true, true, false, TopicDomain.persistent}, + {true, false, true, TopicDomain.persistent}, + {true, false, false, TopicDomain.persistent}, + {false, true, true, TopicDomain.persistent}, + {false, true, false, TopicDomain.persistent}, + {false, false, true, TopicDomain.persistent}, + {false, false, false, TopicDomain.persistent}, + {true, true, true, TopicDomain.non_persistent}, + {true, true, false, TopicDomain.non_persistent}, + {true, false, true, TopicDomain.non_persistent}, + {true, false, false, TopicDomain.non_persistent}, + {false, true, true, TopicDomain.non_persistent}, + {false, true, false, TopicDomain.non_persistent}, + {false, false, true, TopicDomain.non_persistent}, + {false, false, false, TopicDomain.non_persistent} }; } @Test(dataProvider = "autoCreationParamsAll") public void testGetMetadataIfNonPartitionedTopicExists(boolean configAllowAutoTopicCreation, boolean paramMetadataAutoCreationEnabled, - boolean isUsingHttpLookup) throws Exception { + boolean isUsingHttpLookup, + TopicDomain topicDomain) throws Exception { conf.setAllowAutoTopicCreationType(TopicType.PARTITIONED); conf.setDefaultNumPartitions(3); conf.setAllowAutoTopicCreation(configAllowAutoTopicCreation); setup(); + + Semaphore semaphore = pulsar.getBrokerService().getLookupRequestSemaphore(); + int lookupPermitsBefore = semaphore.availablePermits(); + LookupService lookup = getLookupService(isUsingHttpLookup); // Create topic. - final String topicNameStr = BrokerTestUtil.newUniqueName("persistent://" + DEFAULT_NS + "/tp"); + final String topicNameStr = BrokerTestUtil.newUniqueName(topicDomain.value() + "://" + DEFAULT_NS + "/tp"); final TopicName topicName = TopicName.get(topicNameStr); admin.topics().createNonPartitionedTopic(topicNameStr); // Verify. @@ -152,6 +185,13 @@ public void testGetMetadataIfNonPartitionedTopicExists(boolean configAllowAutoTo for (int i = 0; i < 3; i++) { assertFalse(topicList.contains(topicName.getPartition(i))); } + + // Verify: lookup semaphore has been releases. + Awaitility.await().untilAsserted(() -> { + int lookupPermitsAfter = semaphore.availablePermits(); + assertEquals(lookupPermitsAfter, lookupPermitsBefore); + }); + // Cleanup. client.close(); admin.topics().delete(topicNameStr, false); @@ -160,14 +200,19 @@ public void testGetMetadataIfNonPartitionedTopicExists(boolean configAllowAutoTo @Test(dataProvider = "autoCreationParamsAll") public void testGetMetadataIfPartitionedTopicExists(boolean configAllowAutoTopicCreation, boolean paramMetadataAutoCreationEnabled, - boolean isUsingHttpLookup) throws Exception { + boolean isUsingHttpLookup, + TopicDomain topicDomain) throws Exception { conf.setAllowAutoTopicCreationType(TopicType.PARTITIONED); conf.setDefaultNumPartitions(3); conf.setAllowAutoTopicCreation(configAllowAutoTopicCreation); setup(); + + Semaphore semaphore = pulsar.getBrokerService().getLookupRequestSemaphore(); + int lookupPermitsBefore = semaphore.availablePermits(); + LookupService lookup = getLookupService(isUsingHttpLookup); // Create topic. - final String topicNameStr = BrokerTestUtil.newUniqueName("persistent://" + DEFAULT_NS + "/tp"); + final String topicNameStr = BrokerTestUtil.newUniqueName(topicDomain.value() + "://" + DEFAULT_NS + "/tp"); final TopicName topicName = TopicName.get(topicNameStr); admin.topics().createPartitionedTopic(topicNameStr, 3); // Verify. @@ -177,6 +222,13 @@ public void testGetMetadataIfPartitionedTopicExists(boolean configAllowAutoTopic assertEquals(response.partitions, 3); List topicList = admin.topics().getList("public/default"); assertFalse(topicList.contains(topicNameStr)); + + // Verify: lookup semaphore has been releases. + Awaitility.await().untilAsserted(() -> { + int lookupPermitsAfter = semaphore.availablePermits(); + assertEquals(lookupPermitsAfter, lookupPermitsBefore); + }); + // Cleanup. client.close(); admin.topics().deletePartitionedTopic(topicNameStr, false); @@ -186,20 +238,24 @@ public void testGetMetadataIfPartitionedTopicExists(boolean configAllowAutoTopic public Object[][] clients(){ return new Object[][]{ // isUsingHttpLookup. - {true}, - {false} + {true, TopicDomain.persistent}, + {false, TopicDomain.non_persistent} }; } @Test(dataProvider = "clients") - public void testAutoCreatePartitionedTopic(boolean isUsingHttpLookup) throws Exception { + public void testAutoCreatePartitionedTopic(boolean isUsingHttpLookup, TopicDomain topicDomain) throws Exception { conf.setAllowAutoTopicCreationType(TopicType.PARTITIONED); conf.setDefaultNumPartitions(3); conf.setAllowAutoTopicCreation(true); setup(); + + Semaphore semaphore = pulsar.getBrokerService().getLookupRequestSemaphore(); + int lookupPermitsBefore = semaphore.availablePermits(); + LookupService lookup = getLookupService(isUsingHttpLookup); // Create topic. - final String topicNameStr = BrokerTestUtil.newUniqueName("persistent://" + DEFAULT_NS + "/tp"); + final String topicNameStr = BrokerTestUtil.newUniqueName(topicDomain.value() + "://" + DEFAULT_NS + "/tp"); final TopicName topicName = TopicName.get(topicNameStr); // Verify. PulsarClient client = PulsarClient.builder().serviceUrl(pulsar.getBrokerServiceUrl()).build(); @@ -214,19 +270,30 @@ public void testAutoCreatePartitionedTopic(boolean isUsingHttpLookup) throws Exc // partitions. assertFalse(topicList.contains(topicName.getPartition(i))); } + + // Verify: lookup semaphore has been releases. + Awaitility.await().untilAsserted(() -> { + int lookupPermitsAfter = semaphore.availablePermits(); + assertEquals(lookupPermitsAfter, lookupPermitsBefore); + }); + // Cleanup. client.close(); admin.topics().deletePartitionedTopic(topicNameStr, false); } @Test(dataProvider = "clients") - public void testAutoCreateNonPartitionedTopic(boolean isUsingHttpLookup) throws Exception { + public void testAutoCreateNonPartitionedTopic(boolean isUsingHttpLookup, TopicDomain topicDomain) throws Exception { conf.setAllowAutoTopicCreationType(TopicType.NON_PARTITIONED); conf.setAllowAutoTopicCreation(true); setup(); + + Semaphore semaphore = pulsar.getBrokerService().getLookupRequestSemaphore(); + int lookupPermitsBefore = semaphore.availablePermits(); + LookupService lookup = getLookupService(isUsingHttpLookup); // Create topic. - final String topicNameStr = BrokerTestUtil.newUniqueName("persistent://" + DEFAULT_NS + "/tp"); + final String topicNameStr = BrokerTestUtil.newUniqueName(topicDomain.value() + "://" + DEFAULT_NS + "/tp"); final TopicName topicName = TopicName.get(topicNameStr); // Verify. PulsarClient client = PulsarClient.builder().serviceUrl(pulsar.getBrokerServiceUrl()).build(); @@ -236,8 +303,18 @@ public void testAutoCreateNonPartitionedTopic(boolean isUsingHttpLookup) throws assertFalse(partitionedTopics.contains(topicNameStr)); List topicList = admin.topics().getList("public/default"); assertFalse(topicList.contains(topicNameStr)); + + // Verify: lookup semaphore has been releases. + Awaitility.await().untilAsserted(() -> { + int lookupPermitsAfter = semaphore.availablePermits(); + assertEquals(lookupPermitsAfter, lookupPermitsBefore); + }); + // Cleanup. client.close(); + try { + admin.topics().delete(topicNameStr, false); + } catch (Exception ex) {} } @DataProvider(name = "autoCreationParamsNotAllow") @@ -263,6 +340,10 @@ public void testGetMetadataIfNotAllowedCreate(boolean configAllowAutoTopicCreati conf.setDefaultNumPartitions(3); conf.setAllowAutoTopicCreation(configAllowAutoTopicCreation); setup(); + + Semaphore semaphore = pulsar.getBrokerService().getLookupRequestSemaphore(); + int lookupPermitsBefore = semaphore.availablePermits(); + LookupService lookup = getLookupService(isUsingHttpLookup); // Define topic. final String topicNameStr = BrokerTestUtil.newUniqueName("persistent://" + DEFAULT_NS + "/tp"); @@ -287,6 +368,56 @@ public void testGetMetadataIfNotAllowedCreate(boolean configAllowAutoTopicCreati for (int i = 0; i < 3; i++) { assertFalse(topicList.contains(topicName.getPartition(i))); } + + // Verify: lookup semaphore has been releases. + Awaitility.await().untilAsserted(() -> { + int lookupPermitsAfter = semaphore.availablePermits(); + assertEquals(lookupPermitsAfter, lookupPermitsBefore); + }); + + // Cleanup. + client.close(); + } + + @Test(dataProvider = "autoCreationParamsNotAllow") + public void testGetNonPersistentMetadataIfNotAllowedCreate(boolean configAllowAutoTopicCreation, + boolean paramMetadataAutoCreationEnabled, + boolean isUsingHttpLookup) throws Exception { + conf.setAllowAutoTopicCreationType(TopicType.PARTITIONED); + conf.setDefaultNumPartitions(3); + conf.setAllowAutoTopicCreation(configAllowAutoTopicCreation); + setup(); + + Semaphore semaphore = pulsar.getBrokerService().getLookupRequestSemaphore(); + int lookupPermitsBefore = semaphore.availablePermits(); + + LookupService lookup = getLookupService(isUsingHttpLookup); + // Define topic. + final String topicNameStr = BrokerTestUtil.newUniqueName("non-persistent://" + DEFAULT_NS + "/tp"); + final TopicName topicName = TopicName.get(topicNameStr); + // Verify. + // Regarding non-persistent topic, we do not know whether it exists or not. + // Broker will return a non-partitioned metadata if partitioned metadata does not exist. + PulsarClient client = PulsarClient.builder().serviceUrl(pulsar.getBrokerServiceUrl()).build(); + PartitionedTopicMetadata metadata = lookup + .getPartitionedTopicMetadata(TopicName.get(topicNameStr), paramMetadataAutoCreationEnabled).join(); + assertEquals(metadata.partitions, 0); + + List partitionedTopics = admin.topics().getPartitionedTopicList("public/default"); + pulsar.getPulsarResources().getNamespaceResources().getPartitionedTopicResources().partitionedTopicExists(topicName); + assertFalse(partitionedTopics.contains(topicNameStr)); + List topicList = admin.topics().getList("public/default"); + assertFalse(topicList.contains(topicNameStr)); + for (int i = 0; i < 3; i++) { + assertFalse(topicList.contains(topicName.getPartition(i))); + } + + // Verify: lookup semaphore has been releases. + Awaitility.await().untilAsserted(() -> { + int lookupPermitsAfter = semaphore.availablePermits(); + assertEquals(lookupPermitsAfter, lookupPermitsBefore); + }); + // Cleanup. client.close(); } From 087ccb5a97e81090fad56f780db22f8ec8cf3078 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Tue, 21 May 2024 22:04:05 +0800 Subject: [PATCH 6/9] address comments --- .../java/org/apache/pulsar/client/api/PulsarClient.java | 3 ++- .../java/org/apache/pulsar/client/impl/LookupService.java | 7 ++++--- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClient.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClient.java index 09e6d1babaf20..6c46bce254f6f 100644 --- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClient.java +++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClient.java @@ -325,7 +325,8 @@ default CompletableFuture> getPartitionsForTopic(String topic) { * 1. Get the partitions if the topic exists. Return "[{partition-0}, {partition-1}....{partition-n}}]" if a * partitioned topic exists; return "[{topic}]" if a non-partitioned topic exists. * 2. When {@param metadataAutoCreationEnabled} is "false", neither the partitioned topic nor non-partitioned - * topic does not exist. You will get an {@link PulsarClientException.NotFoundException}. + * topic does not exist. You will get an {@link PulsarClientException.NotFoundException} or a + * {@link PulsarClientException.TopicDoesNotExistException}. * 2-1. You will get a {@link PulsarClientException.NotSupportedException} with metadataAutoCreationEnabled=false * on an old broker version which does not support getting partitions without partitioned metadata auto-creation. * 3. When {@param metadataAutoCreationEnabled} is "true," it will trigger an auto-creation for this topic(using diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/LookupService.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/LookupService.java index df9ee475ff6eb..b83289b1e3fbd 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/LookupService.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/LookupService.java @@ -72,9 +72,10 @@ default CompletableFuture getPartitionedTopicMetadata( * 1.Get the partitions if the topic exists. Return "{partition: n}" if a partitioned topic exists; * return "{partition: 0}" if a non-partitioned topic exists. * 2. When {@param metadataAutoCreationEnabled} is "false," neither partitioned topic nor non-partitioned topic - * does not exist. You will get an {@link PulsarClientException.NotFoundException}. - * 2-1. You will get a {@link PulsarClientException.NotSupportedException} if the broker's version is an older - * one that does not support this feature and the Pulsar client is using a binary protocol "serviceUrl". + * does not exist. You will get a {@link PulsarClientException.NotFoundException} or + * a {@link PulsarClientException.TopicDoesNotExistException}. + * 2-1. You will get a {@link PulsarClientException.NotSupportedException} with metadataAutoCreationEnabled=false + * on an old broker version which does not support getting partitions without partitioned metadata auto-creation. * 3.When {@param metadataAutoCreationEnabled} is "true," it will trigger an auto-creation for this topic(using * the default topic auto-creation strategy you set for the broker), and the corresponding result is returned. * For the result, see case 1. From fe6f54bacc424a96a0a73b515f661a1cd64fc3c3 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Tue, 21 May 2024 22:59:33 +0800 Subject: [PATCH 7/9] checkstyle --- .../main/java/org/apache/pulsar/client/impl/LookupService.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/LookupService.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/LookupService.java index b83289b1e3fbd..ccd1f6b23f2f3 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/LookupService.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/LookupService.java @@ -75,7 +75,8 @@ default CompletableFuture getPartitionedTopicMetadata( * does not exist. You will get a {@link PulsarClientException.NotFoundException} or * a {@link PulsarClientException.TopicDoesNotExistException}. * 2-1. You will get a {@link PulsarClientException.NotSupportedException} with metadataAutoCreationEnabled=false - * on an old broker version which does not support getting partitions without partitioned metadata auto-creation. + * on an old broker version which does not support getting partitions without partitioned metadata + * auto-creation. * 3.When {@param metadataAutoCreationEnabled} is "true," it will trigger an auto-creation for this topic(using * the default topic auto-creation strategy you set for the broker), and the corresponding result is returned. * For the result, see case 1. From 2c6e82668dff6c767066a39ee4d9640bd7f50999 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Wed, 22 May 2024 12:22:19 +0800 Subject: [PATCH 8/9] guarantees the behavior of non-persistent topic is the same as before --- .../admin/impl/PersistentTopicsBase.java | 11 +-- .../admin/GetPartitionMetadataTest.java | 73 ++++++++++++++++--- 2 files changed, 62 insertions(+), 22 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java index c93185e78f75b..104a84d041d8a 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java @@ -556,16 +556,7 @@ protected CompletableFuture internalGetPartitionedMeta } else { // If it does not exist, response a Not Found error. // Otherwise, response a non-partitioned metadata. - if (topicName.isPersistent()) { - return internalCheckTopicExists(topicName).thenApply(__ -> metadata); - } else { - // Regarding non-persistent topic, we do not know whether it exists or not. - // Just return a non-partitioned metadata if partitioned metadata does not - // exist. - // Broker will respond a not found error when doing subscribing or producing if - // broker not allow to auto create topics. - return CompletableFuture.completedFuture(metadata); - } + return internalCheckTopicExists(topicName).thenApply(__ -> metadata); } }); } 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 147960ff03dde..51f643d2b7823 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 @@ -27,6 +27,7 @@ import java.util.concurrent.Semaphore; import lombok.extern.slf4j.Slf4j; import org.apache.pulsar.broker.BrokerTestUtil; +import org.apache.pulsar.broker.ServiceConfiguration; import org.apache.pulsar.client.api.ProducerConsumerBase; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; @@ -38,6 +39,7 @@ import org.apache.pulsar.common.policies.data.TopicType; import org.apache.pulsar.common.util.FutureUtil; import org.awaitility.Awaitility; +import org.testng.Assert; import org.testng.annotations.AfterMethod; import org.testng.annotations.DataProvider; import org.testng.annotations.Test; @@ -325,10 +327,8 @@ public Object[][] autoCreationParamsNotAllow(){ {true, false, false}, {false, false, true}, {false, false, false}, - // These test cases are for the following PR. - // Which was described in the Motivation of https://github.com/apache/pulsar/pull/22206. - //{false, true, true}, - //{false, true, false}, + {false, true, true}, + {false, true, false}, }; } @@ -336,6 +336,11 @@ public Object[][] autoCreationParamsNotAllow(){ public void testGetMetadataIfNotAllowedCreate(boolean configAllowAutoTopicCreation, boolean paramMetadataAutoCreationEnabled, boolean isUsingHttpLookup) throws Exception { + if (!configAllowAutoTopicCreation && paramMetadataAutoCreationEnabled) { + // These test cases are for the following PR. + // Which was described in the Motivation of https://github.com/apache/pulsar/pull/22206. + return; + } conf.setAllowAutoTopicCreationType(TopicType.PARTITIONED); conf.setDefaultNumPartitions(3); conf.setAllowAutoTopicCreation(configAllowAutoTopicCreation); @@ -379,7 +384,34 @@ public void testGetMetadataIfNotAllowedCreate(boolean configAllowAutoTopicCreati client.close(); } - @Test(dataProvider = "autoCreationParamsNotAllow") + @DataProvider(name = "autoCreationParamsForNonPersistentTopic") + public Object[][] autoCreationParamsForNonPersistentTopic(){ + return new Object[][]{ + // configAllowAutoTopicCreation, paramCreateIfAutoCreationEnabled, isUsingHttpLookup. + {true, true, true}, + {true, true, false}, + {false, true, true}, + {false, true, false}, + {false, false, true} + }; + } + + /** + * Regarding the API "get partitioned metadata" about non-persistent topic. + * The original behavior is: + * param-auto-create = true, broker-config-auto-create = true + * HTTP API: default configuration {@link ServiceConfiguration#getDefaultNumPartitions()} + * binary API: default configuration {@link ServiceConfiguration#getDefaultNumPartitions()} + * param-auto-create = true, broker-config-auto-create = false + * HTTP API: {partitions: 0} + * binary API: {partitions: 0} + * param-auto-create = false + * HTTP API: not found error + * binary API: not support + * This test only guarantees that the behavior is the same as before. The following separated PR will fix the + * incorrect behavior. + */ + @Test(dataProvider = "autoCreationParamsForNonPersistentTopic") public void testGetNonPersistentMetadataIfNotAllowedCreate(boolean configAllowAutoTopicCreation, boolean paramMetadataAutoCreationEnabled, boolean isUsingHttpLookup) throws Exception { @@ -399,17 +431,34 @@ public void testGetNonPersistentMetadataIfNotAllowedCreate(boolean configAllowAu // Regarding non-persistent topic, we do not know whether it exists or not. // Broker will return a non-partitioned metadata if partitioned metadata does not exist. PulsarClient client = PulsarClient.builder().serviceUrl(pulsar.getBrokerServiceUrl()).build(); + + if (!configAllowAutoTopicCreation && !paramMetadataAutoCreationEnabled && isUsingHttpLookup) { + try { + lookup.getPartitionedTopicMetadata(TopicName.get(topicNameStr), paramMetadataAutoCreationEnabled) + .join(); + Assert.fail("Expected a not found ex"); + } catch (Exception ex) { + // Cleanup. + client.close(); + return; + } + } + PartitionedTopicMetadata metadata = lookup .getPartitionedTopicMetadata(TopicName.get(topicNameStr), paramMetadataAutoCreationEnabled).join(); - assertEquals(metadata.partitions, 0); + if (configAllowAutoTopicCreation && paramMetadataAutoCreationEnabled) { + assertEquals(metadata.partitions, 3); + } else { + assertEquals(metadata.partitions, 0); + } List partitionedTopics = admin.topics().getPartitionedTopicList("public/default"); - pulsar.getPulsarResources().getNamespaceResources().getPartitionedTopicResources().partitionedTopicExists(topicName); - assertFalse(partitionedTopics.contains(topicNameStr)); - List topicList = admin.topics().getList("public/default"); - assertFalse(topicList.contains(topicNameStr)); - for (int i = 0; i < 3; i++) { - assertFalse(topicList.contains(topicName.getPartition(i))); + pulsar.getPulsarResources().getNamespaceResources().getPartitionedTopicResources() + .partitionedTopicExists(topicName); + if (configAllowAutoTopicCreation && paramMetadataAutoCreationEnabled) { + assertTrue(partitionedTopics.contains(topicNameStr)); + } else { + assertFalse(partitionedTopics.contains(topicNameStr)); } // Verify: lookup semaphore has been releases. From 403f4d94934ba39eb8e75e0ee97530e9604b5b19 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Wed, 22 May 2024 16:12:59 +0800 Subject: [PATCH 9/9] fix in-correct error log --- .../java/org/apache/pulsar/broker/service/ServerCnx.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 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 085e5a1c52b4d..926ca13c05a20 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 @@ -657,9 +657,11 @@ protected void handlePartitionMetadataRequest(CommandPartitionedTopicMetadata pa } return null; }).whenComplete((ignore, ignoreEx) -> { - log.error("{} {} Failed to handle partition metadata request", topicName, - ServerCnx.this.toString(), ignoreEx); lookupSemaphore.release(); + if (ignoreEx != null) { + log.error("{} {} Failed to handle partition metadata request", topicName, + ServerCnx.this.toString(), ignoreEx); + } }); } else { // Get if exists, create a new one if not exists.