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

heesung-sn · 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-sn#53

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

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

...-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java

    /**
     * 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.

...test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java

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

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java

+                    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

...src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java

+                                }
+                        );
+            } 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

...ava/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java

+                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

...ava/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java

@@ -732,15 +770,20 @@ private void handleOwnEvent(String serviceUnit, ServiceUnitStateData data) {
        if (getOwnerRequest != null) {
            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

...ava/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java

+        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

...-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java

    /**
     * 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

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java

+        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.

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java

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

...-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java

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


+1, disconnectingClients is better.

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java

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

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java

+                    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-sn requested a review from Demogorgon314 October 20, 2023 21:57

Demogorgon314 reviewed Oct 26, 2023

View reviewed changes

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java Show resolved Hide resolved

Demogorgon314 reviewed Oct 26, 2023

View reviewed changes

heesung-sn 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-sn 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-sn force-pushed the pip-307 branch from 2d42d6e to 5bc4a23 Compare November 3, 2023 22:08

resolved comments

7f0b981

heesung-sn 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-sn merged commit c30eae1 into apache:master Nov 6, 2023
48 checks passed

heesung-sn 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-sn 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

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

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

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

heesung-sn commented Oct 20, 2023

Demogorgon314 Oct 26, 2023

heesung-sn Oct 27, 2023

dragosvictor Nov 3, 2023

gaoran10 Nov 3, 2023 •

edited

Loading

heesung-sn Nov 3, 2023

dragosvictor Oct 31, 2023

heesung-sn Nov 1, 2023

dragosvictor left a comment

dragosvictor Nov 3, 2023

heesung-sn Nov 3, 2023

dragosvictor Nov 3, 2023

heesung-sn Nov 3, 2023

dragosvictor Nov 3, 2023

heesung-sn Nov 3, 2023

dragosvictor Nov 3, 2023

heesung-sn Nov 3, 2023

dragosvictor Nov 3, 2023

dragosvictor Nov 3, 2023

heesung-sn Nov 3, 2023

gaoran10 left a comment

gaoran10 Nov 3, 2023 •

edited

Loading

dragosvictor Nov 4, 2023

heesung-sn Nov 4, 2023

dragosvictor commented Nov 5, 2023

	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));

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

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

Conversation

heesung-sn 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gaoran10 Nov 3, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dragosvictor left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gaoran10 left a comment

Choose a reason for hiding this comment

gaoran10 Nov 3, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dragosvictor commented Nov 5, 2023

gaoran10 Nov 3, 2023 •

edited

Loading

gaoran10 Nov 3, 2023 •

edited

Loading