[improve][broker] PIP-307 Added assignedBrokerUrl to CloseProducerCmd to skip lookups upon producer reconnections during unloading by heesung-sohn · Pull Request #21408 · apache/pulsar

heesung-sohn · 2023-10-20T21:50:47Z

PIP: #20748

Motivation

Please refer to the pip #20748

There will be a separate PR for the similar changes on CloseConsumerCmd.

Modifications

Please refer to the pip #20748

Added assignedBrokerUrls to CloseProducerCmd to skip client lookups upon producer reconnections during topic unloading
Upon bundle transfer, topic.close(..) will be called at the source broker two times:
- At Releasing, without disconnecting producers
- At Owned, to fully close the producers with assigned broker information

Verifying this change

Make sure that the change passes the CI checks.

Added new tests In ExtensibleLoadManagerTest to cover this logic.

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

Documentation

doc
doc-required
doc-not-needed
doc-complete

Matching PR in forked repository

PR in forked repository: heesung-sohn#53

Demogorgon314 · 2023-10-26T02:28:56Z

    /**
     * Close this topic - close all producers and subscriptions associated with this topic.
     *
+     * @param closeWithoutDisconnectingClients don't disconnect clients


nit: I think the parameter should be disconnectingClients, it will be clearer to other developers. But it is not this PR's responsibility.

yes, I was following the closeWithoutWaitingClientDisconnect's convention. Probably we need a minor PR to fix this naming, if we want to refactor it.

Agree, the naming is confusing at first.

+1, disconnectingClients is better.

dragosvictor · 2023-10-31T23:30:51Z

+                    final URI uri = new URI(producer.client.conf.isUseTls()
+                            ? closeProducer.getAssignedBrokerServiceUrlTls()
+                            : closeProducer.getAssignedBrokerServiceUrl());
+                    log.info("[{}] Broker notification of Closed producer: {}. Redirecting to {}.",
+                            remoteAddress, closeProducer.getProducerId(), uri);
+                    producer.getConnectionHandler().connectionClosed(this, 0L, Optional.of(uri));
+                } catch (URISyntaxException e) {
+                    log.error("[{}] Invalid redirect url {}/{} for {}", remoteAddress,
+                            closeProducer.getAssignedBrokerServiceUrl(),
+                            closeProducer.getAssignedBrokerServiceUrlTls(),


GetAssignedBrokerServiceUrl or GetAssignedBrokerServiceUrlTls could throw an InvalidStateException if the respective value is not present.

Good point. I will update this part.

dragosvictor

Looks good overall, left a couple of cosmetic suggestions and questions.

dragosvictor · 2023-11-03T03:39:48Z

+                                }
+                        );
+            } catch (Throwable e) {
+                log.error("Failed to DestinationBrokerLookupData for topic:{}", topic, e);


Nit: can we improve the error message here?

Suggested change

log.error("Failed to DestinationBrokerLookupData for topic:{}", topic, e);

log.error("Failed to lookup destination broker for topic:{}", topic, e);

sure. I will update this.

dragosvictor · 2023-11-03T03:45:02Z

+                if (data.dstBroker() != null) {
+                    return Optional.of(data.dstBroker());
+                }
+                return Optional.empty();


Nit: we can simplify this to:

Suggested change

if (data.dstBroker() != null) {

return Optional.of(data.dstBroker());

}

return Optional.empty();

return Optional.ofNullable(data.dstBroker());

sure. I will update this.

dragosvictor · 2023-11-03T03:57:07Z

            getOwnerRequest.complete(data.dstBroker());
        }
-        stateChangeListeners.notify(serviceUnit, data, null);
+        CompletableFuture<Integer> ownFuture = null;


The null assignment doesn't seem necessary here.

sure. I will update this.

dragosvictor · 2023-11-03T04:26:10Z

+        stateChangeListeners.notifyOnCompletion(ownFuture, serviceUnit, data)
+                .whenComplete((__, e) -> log(e, serviceUnit, data, null));


Question: why this is needed now?

Good point.
I think we need to log the first and second if cases only.

Let me update this.

dragosvictor · 2023-11-03T04:32:13Z

    /**
     * Close this topic - close all producers and subscriptions associated with this topic.
     *
+     * @param closeWithoutDisconnectingClients don't disconnect clients


Agree, the naming is confusing at first.

dragosvictor · 2023-11-03T04:54:15Z

+        connectionClosed(cnx, null, Optional.empty());
+    }
+
+    public void connectionClosed(ClientCnx cnx, Long initialConnectionDelayMs, Optional<URI> hostUrl) {


Just an observation: it's not clear to me when we're using a raw reference and when an Optional. Seems like we could've used Optional<Long> initialConnectionDelayMs just as well.

I agree. I will make this Optional<Long> initialConnectionDelayMs

… to skip lookups upon producer reconnections during unloading

gaoran10

Great work! Left some trival comments.

gaoran10 · 2023-11-03T17:44:12Z

    /**
     * Close this topic - close all producers and subscriptions associated with this topic.
     *
+     * @param closeWithoutDisconnectingClients don't disconnect clients


+1, disconnectingClients is better.

dragosvictor · 2023-11-04T03:48:16Z

+                    final URI uri = new URI(producer.client.conf.isUseTls()
+                            ? closeProducer.getAssignedBrokerServiceUrlTls()
+                            : closeProducer.getAssignedBrokerServiceUrl());


Thank you for fixing the similar issue below, but this one is still present.

Thank you for checking this.

I changed the catch clause to catch IllegalStateException as well in case the broker does not return both urls.

dragosvictor · 2023-11-05T23:37:17Z

Nice, looks good!

github-actions Bot added the doc-not-needed Your PR changes do not impact docs label Oct 20, 2023

heesung-sohn requested a review from Demogorgon314 October 20, 2023 21:57

Demogorgon314 reviewed Oct 26, 2023

View reviewed changes

Comment thread pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

Demogorgon314 reviewed Oct 26, 2023

View reviewed changes

heesung-sohn force-pushed the pip-307 branch 2 times, most recently from 025aac7 to 2d42d6e Compare October 27, 2023 22:11

Demogorgon314 approved these changes Oct 31, 2023

View reviewed changes

Demogorgon314 requested review from Technoboy-, codelipenghui and gaoran10 October 31, 2023 03:45

dragosvictor reviewed Oct 31, 2023

View reviewed changes

dragosvictor reviewed Nov 3, 2023

View reviewed changes

Heesung Sohn added 2 commits November 3, 2023 10:49

[improve][broker] PIP-307 Added assignedBrokerUrl to CloseProducerCmd…

e14968b

… to skip lookups upon producer reconnections during unloading

Resolved comments

a97e2e0

gaoran10 reviewed Nov 3, 2023

View reviewed changes

heesung-sohn force-pushed the pip-307 branch from 2d42d6e to 5bc4a23 Compare November 3, 2023 22:08

resolved comments

7f0b981

heesung-sohn force-pushed the pip-307 branch from 5bc4a23 to 7f0b981 Compare November 3, 2023 22:13

gaoran10 approved these changes Nov 4, 2023

View reviewed changes

dragosvictor reviewed Nov 4, 2023

View reviewed changes

resolved comment

9af980d

heesung-sohn merged commit c30eae1 into apache:master Nov 6, 2023

heesung-sohn mentioned this pull request Nov 6, 2023

[improve][broker] PIP-307 Added assignedBrokerUrl to CloseProducerCmd to skip lookups upon producer reconnections during unloading #21516

Closed

15 tasks

dragosvictor mentioned this pull request Dec 6, 2023

[improve][broker] PIP-307: Implement broker and client consumer changes when topic is unloading #21682

Merged

15 tasks

dragosvictor mentioned this pull request Dec 22, 2023

[improve][broker] PIP-307: Add proxy support for Java client #21789

Merged

15 tasks

coderzc mentioned this pull request Dec 27, 2023

Bump pulsar version to 3.2.0-SNAPSHOT & fix incompatibility with master streamnative/mop#1194

Closed

3 tasks

heesung-sohn deleted the pip-307 branch April 2, 2024 17:45

dragosvictor mentioned this pull request Apr 18, 2024

[improve] PIP-307: Use assigned broker URL hints during broker reconnection apache/pulsar-client-go#1208

Merged

1 task

lhotari added this to the 3.2.0 milestone Oct 21, 2024

codelipenghui assigned heesung-sohn Jun 4, 2025

	log.error("Failed to DestinationBrokerLookupData for topic:{}", topic, e);
	log.error("Failed to lookup destination broker for topic:{}", topic, e);

		stateChangeListeners.notifyOnCompletion(ownFuture, serviceUnit, data)
		.whenComplete((__, e) -> log(e, serviceUnit, data, null));

Conversation

heesung-sohn commented Oct 20, 2023

Motivation

Modifications

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Matching PR in forked repository

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gaoran10 Nov 3, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dragosvictor left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gaoran10 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

gaoran10 Nov 3, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dragosvictor commented Nov 5, 2023

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

gaoran10 Nov 3, 2023 •

edited

Loading

gaoran10 Nov 3, 2023 •

edited

Loading