diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LoadManager.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LoadManager.java index 2cce68b60cb494..0dd5d948480ab9 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LoadManager.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LoadManager.java @@ -31,6 +31,7 @@ import org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerWrapper; import org.apache.pulsar.broker.loadbalance.impl.SimpleLoadManagerImpl; import org.apache.pulsar.broker.lookup.LookupResult; +import org.apache.pulsar.broker.namespace.LookupOptions; import org.apache.pulsar.common.naming.ServiceUnitId; import org.apache.pulsar.common.stats.Metrics; import org.apache.pulsar.common.util.Reflections; @@ -63,7 +64,7 @@ public interface LoadManager { Optional getLeastLoaded(ServiceUnitId su) throws Exception; default CompletableFuture> findBrokerServiceUrl( - Optional topic, ServiceUnitId bundle) { + Optional topic, ServiceUnitId bundle, LookupOptions options) { throw new UnsupportedOperationException(); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManager.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManager.java index b7da70d1cf1de1..eabf6005b439b1 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManager.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManager.java @@ -60,9 +60,12 @@ public interface ExtensibleLoadManager extends Closeable { * (e.g. {@link NamespaceService#internalGetWebServiceUrl(NamespaceBundle, LookupOptions)}), * So the topic is optional. * @param serviceUnit service unit (e.g. bundle). + * @param options The lookup options. * @return The broker lookup data. */ - CompletableFuture> assign(Optional topic, ServiceUnitId serviceUnit); + CompletableFuture> assign(Optional topic, + ServiceUnitId serviceUnit, + LookupOptions options); /** * Check the incoming service unit is owned by the current broker. diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java index 677ca2ddae0d82..2b657f8cb9cba9 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java @@ -87,6 +87,7 @@ import org.apache.pulsar.broker.loadbalance.extensions.strategy.LeastResourceUsageWithWeight; import org.apache.pulsar.broker.loadbalance.impl.LoadManagerShared; import org.apache.pulsar.broker.loadbalance.impl.SimpleResourceAllocationPolicies; +import org.apache.pulsar.broker.namespace.LookupOptions; import org.apache.pulsar.broker.namespace.NamespaceEphemeralData; import org.apache.pulsar.broker.namespace.NamespaceService; import org.apache.pulsar.client.admin.PulsarAdminException; @@ -496,7 +497,8 @@ public void initialize(PulsarService pulsar) { @Override public CompletableFuture> assign(Optional topic, - ServiceUnitId serviceUnit) { + ServiceUnitId serviceUnit, + LookupOptions options) { final String bundle = serviceUnit.toString(); @@ -510,7 +512,7 @@ public CompletableFuture> assign(Optional getHeartbeatOrSLAMonitorBrokerId(ServiceUnitId } private CompletableFuture getOrSelectOwnerAsync(ServiceUnitId serviceUnit, - String bundle) { + String bundle, + LookupOptions options) { return serviceUnitStateChannel.getOwnerAsync(bundle).thenCompose(broker -> { // If the bundle not assign yet, select and publish assign event to channel. if (broker.isEmpty()) { - return this.selectAsync(serviceUnit).thenCompose(brokerOpt -> { + return this.selectAsync(serviceUnit, Collections.emptySet(), options).thenCompose(brokerOpt -> { if (brokerOpt.isPresent()) { assignCounter.incrementSuccess(); log.info("Selected new owner broker: {} for bundle: {}.", brokerOpt.get(), bundle); return serviceUnitStateChannel.publishAssignEventAsync(bundle, brokerOpt.get()); } - throw new IllegalStateException( - "Failed to select the new owner broker for bundle: " + bundle); + return CompletableFuture.completedFuture(null); }); } assignCounter.incrementSkip(); @@ -548,22 +550,19 @@ private CompletableFuture> getBrokerLookupData( String bundle) { return owner.thenCompose(broker -> { if (broker.isEmpty()) { - String errorMsg = String.format( - "Failed to get or assign the owner for bundle:%s", bundle); - log.error(errorMsg); - throw new IllegalStateException(errorMsg); - } - return CompletableFuture.completedFuture(broker.get()); - }).thenCompose(broker -> this.getBrokerRegistry().lookupAsync(broker).thenCompose(brokerLookupData -> { - if (brokerLookupData.isEmpty()) { - String errorMsg = String.format( - "Failed to lookup broker:%s for bundle:%s, the broker has not been registered.", - broker, bundle); - log.error(errorMsg); - throw new IllegalStateException(errorMsg); + return CompletableFuture.completedFuture(Optional.empty()); } - return CompletableFuture.completedFuture(brokerLookupData); - })); + return this.getBrokerRegistry().lookupAsync(broker.get()).thenCompose(brokerLookupData -> { + if (brokerLookupData.isEmpty()) { + String errorMsg = String.format( + "Failed to lookup broker:%s for bundle:%s, the broker has not been registered.", + broker, bundle); + log.error(errorMsg); + throw new IllegalStateException(errorMsg); + } + return CompletableFuture.completedFuture(brokerLookupData); + }); + }); } /** @@ -576,7 +575,7 @@ private CompletableFuture> getBrokerLookupData( public CompletableFuture tryAcquiringOwnership(NamespaceBundle namespaceBundle) { log.info("Try acquiring ownership for bundle: {} - {}.", namespaceBundle, brokerRegistry.getBrokerId()); final String bundle = namespaceBundle.toString(); - return assign(Optional.empty(), namespaceBundle) + return assign(Optional.empty(), namespaceBundle, LookupOptions.builder().readOnly(false).build()) .thenApply(brokerLookupData -> { if (brokerLookupData.isEmpty()) { String errorMsg = String.format( @@ -609,12 +608,12 @@ private CompletableFuture> dedupeLookupRequest( } } - public CompletableFuture> selectAsync(ServiceUnitId bundle) { - return selectAsync(bundle, Collections.emptySet()); - } - public CompletableFuture> selectAsync(ServiceUnitId bundle, - Set excludeBrokerSet) { + Set excludeBrokerSet, + LookupOptions options) { + if (options.isReadOnly()) { + return CompletableFuture.completedFuture(Optional.empty()); + } BrokerRegistry brokerRegistry = getBrokerRegistry(); return brokerRegistry.getAvailableBrokerLookupDataAsync() .thenComposeAsync(availableBrokers -> { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerWrapper.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerWrapper.java index cd1561cb70e2d8..25eb27bc58d274 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerWrapper.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerWrapper.java @@ -28,10 +28,11 @@ import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.loadbalance.LoadManager; import org.apache.pulsar.broker.loadbalance.ResourceUnit; -import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLookupData; import org.apache.pulsar.broker.lookup.LookupResult; +import org.apache.pulsar.broker.namespace.LookupOptions; import org.apache.pulsar.common.naming.ServiceUnitId; import org.apache.pulsar.common.stats.Metrics; +import org.apache.pulsar.common.util.FutureUtil; import org.apache.pulsar.policies.data.loadbalancer.LoadManagerReport; public class ExtensibleLoadManagerWrapper implements LoadManager { @@ -62,9 +63,15 @@ public boolean isCentralized() { @Override public CompletableFuture> findBrokerServiceUrl( - Optional topic, ServiceUnitId bundle) { - return loadManager.assign(topic, bundle) - .thenApply(lookupData -> lookupData.map(BrokerLookupData::toLookupResult)); + Optional topic, ServiceUnitId bundle, LookupOptions options) { + return loadManager.assign(topic, bundle, options) + .thenApply(lookupData -> lookupData.map(data -> { + try { + return data.toLookupResult(options); + } catch (PulsarServerException ex) { + throw FutureUtil.wrapToCompletionException(ex); + } + })); } @Override diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java index 25173f61c0648e..4fa9c1db3c19eb 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java @@ -83,6 +83,7 @@ import org.apache.pulsar.broker.loadbalance.extensions.models.Split; import org.apache.pulsar.broker.loadbalance.extensions.models.Unload; import org.apache.pulsar.broker.loadbalance.impl.LoadManagerShared; +import org.apache.pulsar.broker.namespace.LookupOptions; import org.apache.pulsar.broker.namespace.NamespaceService; import org.apache.pulsar.broker.service.BrokerServiceException; import org.apache.pulsar.client.api.CompressionType; @@ -1456,7 +1457,8 @@ private synchronized void doCleanup(String broker) { private Optional selectBroker(String serviceUnit, String inactiveBroker) { try { return loadManager.selectAsync( - LoadManagerShared.getNamespaceBundle(pulsar, serviceUnit), Set.of(inactiveBroker)) + LoadManagerShared.getNamespaceBundle(pulsar, serviceUnit), + Set.of(inactiveBroker), LookupOptions.builder().build()) .get(inFlightStateWaitingTimeInMillis, MILLISECONDS); } catch (Throwable e) { log.error("Failed to select a broker for serviceUnit:{}", serviceUnit); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/data/BrokerLookupData.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/data/BrokerLookupData.java index 41f5b18e321e84..50a2b704040394 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/data/BrokerLookupData.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/data/BrokerLookupData.java @@ -18,9 +18,12 @@ */ package org.apache.pulsar.broker.loadbalance.extensions.data; +import java.net.URI; import java.util.Map; import java.util.Optional; +import org.apache.pulsar.broker.PulsarServerException; import org.apache.pulsar.broker.lookup.LookupResult; +import org.apache.pulsar.broker.namespace.LookupOptions; import org.apache.pulsar.broker.namespace.NamespaceEphemeralData; import org.apache.pulsar.policies.data.loadbalancer.AdvertisedListener; import org.apache.pulsar.policies.data.loadbalancer.ServiceLookupData; @@ -79,7 +82,19 @@ public long getStartTimestamp() { return this.startTimestamp; } - public LookupResult toLookupResult() { + public LookupResult toLookupResult(LookupOptions options) throws PulsarServerException { + if (options.hasAdvertisedListenerName()) { + AdvertisedListener listener = advertisedListeners.get(options.getAdvertisedListenerName()); + if (listener == null) { + throw new PulsarServerException("the broker do not have " + + options.getAdvertisedListenerName() + " listener"); + } + URI url = listener.getBrokerServiceUrl(); + URI urlTls = listener.getBrokerServiceUrlTls(); + return new LookupResult(webServiceUrl, webServiceUrlTls, + url == null ? null : url.toString(), + urlTls == null ? null : urlTls.toString(), LookupResult.Type.BrokerUrl, false); + } return new LookupResult(webServiceUrl, webServiceUrlTls, pulsarServiceUrl, pulsarServiceUrlTls, LookupResult.Type.BrokerUrl, false); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java index 2f75e364ea58a2..539f899393a4a6 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java @@ -197,7 +197,7 @@ public CompletableFuture> getBrokerServiceUrlAsync(TopicN return CompletableFuture.completedFuture(optResult); } if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) { - return loadManager.get().findBrokerServiceUrl(Optional.of(topic), bundle); + return loadManager.get().findBrokerServiceUrl(Optional.of(topic), bundle, options); } else { // TODO: Add unit tests cover it. return findBrokerServiceUrl(bundle, options); @@ -313,7 +313,7 @@ private CompletableFuture> internalGetWebServiceUrl(@Nullable Serv } CompletableFuture> future = ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar) - ? loadManager.get().findBrokerServiceUrl(Optional.ofNullable(topic), bundle) : + ? loadManager.get().findBrokerServiceUrl(Optional.ofNullable(topic), bundle, options) : findBrokerServiceUrl(bundle, options); return future.thenApply(lookupResult -> { diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/AntiAffinityNamespaceGroupExtensionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/AntiAffinityNamespaceGroupExtensionTest.java index d77490e1b8210a..cd653a964be360 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/AntiAffinityNamespaceGroupExtensionTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/AntiAffinityNamespaceGroupExtensionTest.java @@ -38,6 +38,7 @@ import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateData; import org.apache.pulsar.broker.loadbalance.extensions.filter.AntiAffinityGroupPolicyFilter; import org.apache.pulsar.broker.loadbalance.extensions.policies.AntiAffinityGroupPolicyHelper; +import org.apache.pulsar.broker.namespace.LookupOptions; import org.apache.pulsar.client.admin.PulsarAdminException; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.PulsarClient; @@ -61,7 +62,8 @@ protected String getLoadManagerClassName() { protected String selectBroker(ServiceUnitId serviceUnit, Object loadManager) { try { - return ((ExtensibleLoadManagerImpl) loadManager).assign(Optional.empty(), serviceUnit).get() + return ((ExtensibleLoadManagerImpl) loadManager) + .assign(Optional.empty(), serviceUnit, LookupOptions.builder().build()).get() .get().getPulsarServiceUrl(); } catch (Throwable e) { throw new RuntimeException(e); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplBaseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplBaseTest.java index 3b611ef62c45bf..ea3eff702b4e03 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplBaseTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplBaseTest.java @@ -21,6 +21,8 @@ import static org.mockito.Mockito.reset; import static org.mockito.Mockito.spy; import com.google.common.collect.Sets; + +import java.util.Optional; import java.util.concurrent.CompletableFuture; import org.apache.commons.lang3.reflect.FieldUtils; import org.apache.commons.lang3.tuple.Pair; @@ -76,6 +78,8 @@ protected ServiceConfiguration updateConfig(ServiceConfiguration conf) { conf.setLoadBalancerLoadSheddingStrategy(TransferShedder.class.getName()); conf.setLoadBalancerSheddingEnabled(false); conf.setLoadBalancerDebugModeEnabled(true); + conf.setWebServicePortTls(Optional.of(0)); + conf.setBrokerServicePortTls(Optional.of(0)); return conf; } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java index 3bc7d73294e0e9..b629513a1994b0 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java @@ -156,10 +156,12 @@ public ExtensibleLoadManagerImplTest() { public void testAssignInternalTopic() throws Exception { Optional brokerLookupData1 = primaryLoadManager.assign( Optional.of(TopicName.get(ServiceUnitStateChannelImpl.TOPIC)), - getBundleAsync(pulsar1, TopicName.get(ServiceUnitStateChannelImpl.TOPIC)).get()).get(); + getBundleAsync(pulsar1, TopicName.get(ServiceUnitStateChannelImpl.TOPIC)).get(), + LookupOptions.builder().build()).get(); Optional brokerLookupData2 = secondaryLoadManager.assign( Optional.of(TopicName.get(ServiceUnitStateChannelImpl.TOPIC)), - getBundleAsync(pulsar1, TopicName.get(ServiceUnitStateChannelImpl.TOPIC)).get()).get(); + getBundleAsync(pulsar1, TopicName.get(ServiceUnitStateChannelImpl.TOPIC)).get(), + LookupOptions.builder().build()).get(); assertEquals(brokerLookupData1, brokerLookupData2); assertTrue(brokerLookupData1.isPresent()); @@ -167,7 +169,7 @@ public void testAssignInternalTopic() throws Exception { FieldUtils.readField(channel1, "leaderElectionService", true); Optional currentLeader = leaderElectionService.getCurrentLeader(); assertTrue(currentLeader.isPresent()); - assertEquals(brokerLookupData1.get().getWebServiceUrl(), currentLeader.get().getServiceUrl()); + assertEquals(brokerLookupData1.get().getWebServiceUrlTls(), currentLeader.get().getServiceUrl()); } @Test @@ -175,15 +177,17 @@ public void testAssign() throws Exception { Pair topicAndBundle = getBundleIsNotOwnByChangeEventTopic("test-assign"); TopicName topicName = topicAndBundle.getLeft(); NamespaceBundle bundle = topicAndBundle.getRight(); - Optional brokerLookupData = primaryLoadManager.assign(Optional.empty(), bundle).get(); + Optional brokerLookupData = primaryLoadManager.assign(Optional.empty(), bundle, + LookupOptions.builder().build()).get(); assertTrue(brokerLookupData.isPresent()); log.info("Assign the bundle {} to {}", bundle, brokerLookupData); // Should get owner info from channel. - Optional brokerLookupData1 = secondaryLoadManager.assign(Optional.empty(), bundle).get(); + Optional brokerLookupData1 = secondaryLoadManager.assign(Optional.empty(), bundle, + LookupOptions.builder().build()).get(); assertEquals(brokerLookupData, brokerLookupData1); Optional lookupResult = pulsar2.getNamespaceService() - .getBrokerServiceUrlAsync(topicName, null).get(); + .getBrokerServiceUrlAsync(topicName, LookupOptions.builder().build()).get(); assertTrue(lookupResult.isPresent()); assertEquals(lookupResult.get().getLookupData().getHttpUrl(), brokerLookupData.get().getWebServiceUrl()); @@ -193,6 +197,43 @@ public void testAssign() throws Exception { assertEquals(webServiceUrl.get().toString(), brokerLookupData.get().getWebServiceUrl()); } + @Test + public void testLookupOptions() throws Exception { + Pair topicAndBundle = + getBundleIsNotOwnByChangeEventTopic("test-lookup-options"); + TopicName topicName = topicAndBundle.getLeft(); + NamespaceBundle bundle = topicAndBundle.getRight(); + + admin.topics().createPartitionedTopic(topicName.toString(), 1); + + // Test LookupOptions.readOnly = true when the bundle is not owned by any broker. + Optional webServiceUrlReadOnlyTrue = pulsar1.getNamespaceService() + .getWebServiceUrl(bundle, LookupOptions.builder().readOnly(true).requestHttps(false).build()); + assertTrue(webServiceUrlReadOnlyTrue.isEmpty()); + + // Test LookupOptions.readOnly = false and the bundle assign to some broker. + Optional webServiceUrlReadOnlyFalse = pulsar1.getNamespaceService() + .getWebServiceUrl(bundle, LookupOptions.builder().readOnly(false).requestHttps(false).build()); + assertTrue(webServiceUrlReadOnlyFalse.isPresent()); + + // Test LookupOptions.requestHttps = true + Optional webServiceUrlHttps = pulsar2.getNamespaceService() + .getWebServiceUrl(bundle, LookupOptions.builder().requestHttps(true).build()); + assertTrue(webServiceUrlHttps.isPresent()); + assertTrue(webServiceUrlHttps.get().toString().startsWith("https")); + + // TODO: Support LookupOptions.loadTopicsInBundle = true + + // Test LookupOptions.advertisedListenerName = internal but the broker do not have internal listener. + try { + pulsar2.getNamespaceService() + .getWebServiceUrl(bundle, LookupOptions.builder().advertisedListenerName("internal").build()); + fail(); + } catch (Exception e) { + assertTrue(e.getMessage().contains("the broker do not have internal listener")); + } + } + @Test public void testCheckOwnershipAsync() throws Exception { Pair topicAndBundle = getBundleIsNotOwnByChangeEventTopic("test-check-ownership"); @@ -210,7 +251,7 @@ public void testCheckOwnershipAsync() throws Exception { assertFalse(secondaryLoadManager.checkOwnershipAsync(Optional.empty(), bundle).get()); // 2. Assign the bundle to a broker. - Optional lookupData = primaryLoadManager.assign(Optional.empty(), bundle).get(); + Optional lookupData = primaryLoadManager.assign(Optional.empty(), bundle, LookupOptions.builder().build()).get(); assertTrue(lookupData.isPresent()); if (lookupData.get().getPulsarServiceUrl().equals(pulsar1.getBrokerServiceUrl())) { assertTrue(primaryLoadManager.checkOwnershipAsync(Optional.empty(), bundle).get()); @@ -243,7 +284,7 @@ public CompletableFuture> filterAsync(Map brokerLookupData = primaryLoadManager.assign(Optional.empty(), bundle).get(); + Optional brokerLookupData = primaryLoadManager.assign(Optional.empty(), bundle, LookupOptions.builder().build()).get(); assertTrue(brokerLookupData.isPresent()); assertEquals(brokerLookupData.get().getWebServiceUrl(), pulsar2.getWebServiceAddress()); } @@ -263,7 +304,7 @@ public CompletableFuture> filterAsync(Map brokerLookupData = primaryLoadManager.assign(Optional.empty(), bundle).get(); + Optional brokerLookupData = primaryLoadManager.assign(Optional.empty(), bundle, LookupOptions.builder().build()).get(); assertTrue(brokerLookupData.isPresent()); } @@ -272,7 +313,7 @@ public void testUnloadUponTopicLookupFailure() throws Exception { TopicName topicName = TopicName.get("public/test/testUnloadUponTopicLookupFailure"); NamespaceBundle bundle = pulsar1.getNamespaceService().getBundle(topicName); - primaryLoadManager.assign(Optional.empty(), bundle).get(); + primaryLoadManager.assign(Optional.empty(), bundle, LookupOptions.builder().build()).get(); CompletableFuture future1 = new CompletableFuture(); CompletableFuture future2 = new CompletableFuture(); @@ -853,7 +894,7 @@ public CompletableFuture> filterAsync(Map brokerLookupData = primaryLoadManager.assign(Optional.empty(), bundle).get(); + Optional brokerLookupData = primaryLoadManager.assign(Optional.empty(), bundle, LookupOptions.builder().build()).get(); Awaitility.waitAtMost(5, TimeUnit.SECONDS).untilAsserted(() -> { assertTrue(brokerLookupData.isPresent()); assertEquals(brokerLookupData.get().getWebServiceUrl(), pulsar2.getWebServiceAddress()); @@ -1543,7 +1584,7 @@ public void testGetOwnedServiceUnitsAndGetOwnedNamespaceStatus() throws Exceptio String topic = "persistent://" + defaultTestNamespace + "/test-get-owned-service-units"; admin.topics().createPartitionedTopic(topic, 1); NamespaceBundle bundle = getBundleAsync(pulsar1, TopicName.get(topic)).join(); - CompletableFuture> owner = primaryLoadManager.assign(Optional.empty(), bundle); + CompletableFuture> owner = primaryLoadManager.assign(Optional.empty(), bundle, LookupOptions.builder().build()); assertFalse(owner.join().isEmpty()); BrokerLookupData brokerLookupData = owner.join().get(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelTest.java index e060148ceefbdf..837aceca1416fe 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelTest.java @@ -502,7 +502,7 @@ public void transferTestWhenDestBrokerFails() // recovered, check the monitor update state : Assigned -> Owned doReturn(CompletableFuture.completedFuture(Optional.of(brokerId1))) - .when(loadManager).selectAsync(any(), any()); + .when(loadManager).selectAsync(any(), any(), any()); FieldUtils.writeDeclaredField(channel2, "producer", producer, true); FieldUtils.writeDeclaredField(channel1, "inFlightStateWaitingTimeInMillis", 1 , true); @@ -724,7 +724,7 @@ public void handleBrokerDeletionEventTest() var owner1 = channel1.getOwnerAsync(bundle1); var owner2 = channel2.getOwnerAsync(bundle2); doReturn(CompletableFuture.completedFuture(Optional.of(brokerId2))) - .when(loadManager).selectAsync(any(), any()); + .when(loadManager).selectAsync(any(), any(), any()); assertTrue(owner1.get().isEmpty()); assertTrue(owner2.get().isEmpty()); @@ -1126,7 +1126,7 @@ public void assignTestWhenDestBrokerProducerFails() FieldUtils.writeDeclaredField(channel2, "inFlightStateWaitingTimeInMillis", 3 * 1000, true); doReturn(CompletableFuture.completedFuture(Optional.of(brokerId2))) - .when(loadManager).selectAsync(any(), any()); + .when(loadManager).selectAsync(any(), any(), any()); channel1.publishAssignEventAsync(bundle, brokerId2); // channel1 is broken. the assign won't be complete. waitUntilState(channel1, bundle); @@ -1527,7 +1527,7 @@ public void testOverrideInactiveBrokerStateData() // test stable metadata state doReturn(CompletableFuture.completedFuture(Optional.of(brokerId2))) - .when(loadManager).selectAsync(any(), any()); + .when(loadManager).selectAsync(any(), any(), any()); leaderChannel.handleMetadataSessionEvent(SessionReestablished); followerChannel.handleMetadataSessionEvent(SessionReestablished); FieldUtils.writeDeclaredField(leaderChannel, "lastMetadataSessionEventTimestamp", @@ -1592,7 +1592,7 @@ public void testOverrideOrphanStateData() // test stable metadata state doReturn(CompletableFuture.completedFuture(Optional.of(brokerId2))) - .when(loadManager).selectAsync(any(), any()); + .when(loadManager).selectAsync(any(), any(), any()); FieldUtils.writeDeclaredField(leaderChannel, "inFlightStateWaitingTimeInMillis", -1, true); FieldUtils.writeDeclaredField(followerChannel, "inFlightStateWaitingTimeInMillis", @@ -1678,7 +1678,7 @@ public void testActiveGetOwner() throws Exception { // case 7: the ownership cleanup(no new owner) by the leader channel doReturn(CompletableFuture.completedFuture(Optional.empty())) - .when(loadManager).selectAsync(any(), any()); + .when(loadManager).selectAsync(any(), any(), any()); var leaderChannel = channel1; String leader1 = channel1.getChannelOwnerAsync().get(2, TimeUnit.SECONDS).get(); String leader2 = channel2.getChannelOwnerAsync().get(2, TimeUnit.SECONDS).get(); @@ -1705,7 +1705,7 @@ public void testActiveGetOwner() throws Exception { overrideTableViews(bundle, new ServiceUnitStateData(Owned, broker, null, 1)); doReturn(CompletableFuture.completedFuture(Optional.of(brokerId1))) - .when(loadManager).selectAsync(any(), any()); + .when(loadManager).selectAsync(any(), any(), any()); leaderChannel.handleMetadataSessionEvent(SessionReestablished); FieldUtils.writeDeclaredField(leaderChannel, "lastMetadataSessionEventTimestamp", System.currentTimeMillis() - (MAX_CLEAN_UP_DELAY_TIME_IN_SECS * 1000 + 1000), true); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/data/BrokerLookupDataTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/data/BrokerLookupDataTest.java index 0d874e0f77117c..66e8c917d1fc5e 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/data/BrokerLookupDataTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/data/BrokerLookupDataTest.java @@ -18,13 +18,19 @@ */ package org.apache.pulsar.broker.loadbalance.extensions.data; +import static org.testng.Assert.fail; import static org.testng.AssertJUnit.assertEquals; import static org.testng.AssertJUnit.assertTrue; + +import java.net.URI; +import java.net.URISyntaxException; import java.util.HashMap; import java.util.Map; import java.util.Optional; +import org.apache.pulsar.broker.PulsarServerException; import org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl; import org.apache.pulsar.broker.lookup.LookupResult; +import org.apache.pulsar.broker.namespace.LookupOptions; import org.apache.pulsar.policies.data.loadbalancer.AdvertisedListener; import org.testng.annotations.Test; @@ -32,12 +38,20 @@ public class BrokerLookupDataTest { @Test - public void testConstructors() { + public void testConstructors() throws PulsarServerException, URISyntaxException { String webServiceUrl = "http://localhost:8080"; String webServiceUrlTls = "https://localhoss:8081"; String pulsarServiceUrl = "pulsar://localhost:6650"; String pulsarServiceUrlTls = "pulsar+ssl://localhost:6651"; - Map advertisedListeners = new HashMap<>(); + final String listenerUrl = "pulsar://gateway:7000"; + final String listenerUrlTls = "pulsar://gateway:8000"; + final String listener = "internal"; + Map advertisedListeners = new HashMap<>(){{ + put(listener, AdvertisedListener.builder() + .brokerServiceUrl(new URI(listenerUrl)) + .brokerServiceUrlTls(new URI(listenerUrlTls)) + .build()); + }}; Map protocols = new HashMap<>(){{ put("kafka", "9092"); }}; @@ -56,10 +70,22 @@ public void testConstructors() { assertEquals("3.0", lookupData.brokerVersion()); - LookupResult lookupResult = lookupData.toLookupResult(); + LookupResult lookupResult = lookupData.toLookupResult(LookupOptions.builder().build()); assertEquals(webServiceUrl, lookupResult.getLookupData().getHttpUrl()); assertEquals(webServiceUrlTls, lookupResult.getLookupData().getHttpUrlTls()); assertEquals(pulsarServiceUrl, lookupResult.getLookupData().getBrokerUrl()); assertEquals(pulsarServiceUrlTls, lookupResult.getLookupData().getBrokerUrlTls()); + + try { + lookupData.toLookupResult(LookupOptions.builder().advertisedListenerName("others").build()); + fail(); + } catch (PulsarServerException ex) { + assertTrue(ex.getMessage().contains("the broker do not have others listener")); + } + lookupResult = lookupData.toLookupResult(LookupOptions.builder().advertisedListenerName(listener).build()); + assertEquals(listenerUrl, lookupResult.getLookupData().getBrokerUrl()); + assertEquals(listenerUrlTls, lookupResult.getLookupData().getBrokerUrlTls()); + assertEquals(webServiceUrl, lookupResult.getLookupData().getHttpUrl()); + assertEquals(webServiceUrlTls, lookupResult.getLookupData().getHttpUrlTls()); } } diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/loadbalance/ExtensibleLoadManagerTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/loadbalance/ExtensibleLoadManagerTest.java index 1f29e19f018734..ee7497010adfcc 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/loadbalance/ExtensibleLoadManagerTest.java +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/loadbalance/ExtensibleLoadManagerTest.java @@ -403,9 +403,10 @@ public void testIsolationPolicy() throws Exception { () -> { try { admin.lookups().lookupTopicAsync(topic).get(5, TimeUnit.SECONDS); + fail(); } catch (Exception ex) { log.error("Failed to lookup topic: ", ex); - assertThat(ex.getMessage()).contains("Failed to select the new owner broker for bundle"); + assertThat(ex.getMessage()).contains("Service Unavailable"); } } );