Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[PIP 95][Issue 12040][broker] Improved multi-listener in standalone mode #12066

Merged
merged 3 commits into from
Oct 12, 2021
+43 −14
Merged

[PIP 95][Issue 12040][broker] Improved multi-listener in standalone mode #12066

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
26eaddd
Improved multi-listener in standalone mode.
EronWright Sep 16, 2021
8d907bb
Improved multi-listener in standalone mode
EronWright Sep 20, 2021
afa0d52
Merge remote-tracking branch 'origin/master' into ewright/pip95-stand…
EronWright Oct 11, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 11 additions & 7 deletions pulsar-broker/src/main/java/org/apache/pulsar/PulsarStandalone.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
import org.apache.logging.log4j.LogManager;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.ServiceConfigurationUtils;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminBuilder;
import org.apache.pulsar.client.admin.PulsarAdminException;
Expand Down Expand Up @@ -296,11 +297,12 @@ public void start() throws Exception {
broker.start();

final String cluster = config.getClusterName();

if (!config.isTlsEnabled()) {
URL webServiceUrl = new URL(
String.format("http://%s:%d", config.getAdvertisedAddress(), config.getWebServicePort().get()));
String brokerServiceUrl = String.format("pulsar://%s:%d", config.getAdvertisedAddress(),
URL webServiceUrl = new URL(String.format("http://%s:%d",
ServiceConfigurationUtils.getAppliedAdvertisedAddress(config, true),
config.getWebServicePort().get()));
String brokerServiceUrl = String.format("pulsar://%s:%d",
ServiceConfigurationUtils.getAppliedAdvertisedAddress(config, true),
config.getBrokerServicePort().get());
admin = PulsarAdmin.builder().serviceHttpUrl(
webServiceUrl.toString()).authentication(
Expand All @@ -312,9 +314,11 @@ public void start() throws Exception {
.build();
createSampleNameSpace(clusterData, cluster);
} else {
URL webServiceUrlTls = new URL(
String.format("https://%s:%d", config.getAdvertisedAddress(), config.getWebServicePortTls().get()));
String brokerServiceUrlTls = String.format("pulsar+ssl://%s:%d", config.getAdvertisedAddress(),
URL webServiceUrlTls = new URL(String.format("https://%s:%d",
ServiceConfigurationUtils.getAppliedAdvertisedAddress(config, true),
config.getWebServicePortTls().get()));
String brokerServiceUrlTls = String.format("pulsar+ssl://%s:%d",
ServiceConfigurationUtils.getAppliedAdvertisedAddress(config, true),
config.getBrokerServicePortTls().get());
PulsarAdminBuilder builder = PulsarAdmin.builder()
.serviceHttpUrl(webServiceUrlTls.toString())
Expand Down
2 changes: 1 addition & 1 deletion pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/NoopLoadManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,7 @@ public void start() throws PulsarServerException {

LocalBrokerData localData = new LocalBrokerData(pulsar.getSafeWebServiceAddress(),
pulsar.getWebServiceAddressTls(),
pulsar.getSafeBrokerServiceUrl(), pulsar.getBrokerServiceUrlTls());
pulsar.getSafeBrokerServiceUrl(), pulsar.getBrokerServiceUrlTls(), pulsar.getAdvertisedListeners());
localData.setProtocols(pulsar.getProtocolDataToAdvertise());
String brokerReportPath = LoadManager.LOADBALANCE_BROKERS_ROOT + "/" + lookupServiceAddress;

Expand Down
8 changes: 5 additions & 3 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/TopicLookupBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
import javax.ws.rs.core.Response;
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
import org.apache.pulsar.broker.namespace.LookupOptions;
import org.apache.pulsar.broker.web.PulsarWebResource;
Expand Down Expand Up @@ -294,9 +295,10 @@ public static CompletableFuture<ByteBuf> lookupTopicAsync(PulsarService pulsarSe
newAuthoritative, LookupType.Redirect, requestId, false));
} else {
// When running in standalone mode we want to redirect the client through the service
// url, so that the advertised address configuration is not relevant anymore.
boolean redirectThroughServiceUrl = pulsarService.getConfiguration()
.isRunningStandalone();
// url, if the advertised address is localhost (per PulsarStandaloneStarter).
ServiceConfiguration conf = pulsarService.getConfiguration();
boolean isLocalhost = StringUtils.equals("localhost", conf.getAdvertisedAddress());
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In standalone mode, what if we set advertisedAddress to 127.0.0.1?
Or the domain name host is set on the machine, but it points to the local.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

+1

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Java has an API in order to understand if the address is local, for instance we should support also IPv6 localhost addresses

Copy link
Contributor Author

@EronWright EronWright Sep 20, 2021 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good idea @315157973 and @eolivelli, I'll use InetAddress::isLoopbackAddress to test whether the advertised address is a loopback address, in which case we'll force the proxy mode (standalone only).

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I updated the code to use isLoopbackAddress against the resolved LookupData. The logic can be understood as "don't return a LookupData that is pointing to a loopback address; force a proxy mode in that case".

boolean redirectThroughServiceUrl = conf.isRunningStandalone() && isLocalhost;

lookupfuture.complete(newLookupResponse(lookupData.getBrokerUrl(),
lookupData.getBrokerUrlTls(), true /* authoritative */, LookupType.Connect,
Expand Down