Add epoch for connection handler to handle create producer timeout. #5571

codelipenghui · 2019-11-06T09:11:36Z

Motivation

Currently, if user create producer timeout, the connection handler of producer will reconnect to the broker later, but if in broker already done the previous create producer request, the reconnection will failed with "producer with name xxx is already connected".

So this PR will introduce epoch for connection handler and add a field named isGeneratedName for producer to handle above problem.

This PR only handle the generated producer name scenario, so many users occur errors such like
#5535, so we need to fix the generated producer name scenario first.

For the scenario of user specified producer name, we can discuss later and find a simple approach to handle it, i left my idea here: using producer id and producer name as the identity of producer, producer name used for EO producer and producer id can used by the producer reconnect, but this approach depends on globally unique producer id generator.

Modifications

If the producer with generated producer name and epoch of the producer is bigger than the exists producer, the new producer will overwrite the old producer, so the reconnect producer will create succeed.

Verifying this change

Add unit tests to simulate producer timeout and reconnection

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

If yes was chosen, please highlight the changes

Dependencies (does it add or upgrade a dependency): (no)
The public API: (no)
The schema: (no)
The default values of configurations: (no)
The wire protocol: (no)
The rest endpoints: (no)
The admin cli options: (no)
Anything that affects deployment: (no)

Documentation

Does this pull request introduce a new feature? (no)

codelipenghui · 2019-11-06T09:13:12Z

@wolfstudy if this PR can complete before cut 2.4.2, please considering include it, thanks.

sijie · 2019-11-06T09:35:24Z

@wolfstudy @codelipenghui This issue has been reported by many users. so let's include it in 2.4.2

codelipenghui · 2019-11-06T09:58:16Z

run java8 tests

sijie

@codelipenghui the idea looks good to me. left a few comments. PTAL

sijie · 2019-11-06T09:37:12Z

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Producer.java

+    /**
+     * @return the name of producer is generated or user specified
+     */
+    boolean isGeneratedName();


do we need to expose this to producer api? I think this is an implementation detail, which should be hidden in the implementation.

ok, currently no need to expose to the interface

sijie · 2019-11-06T09:37:38Z

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

 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.impl.HandlerState.State;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

-class ConnectionHandler {
+public class ConnectionHandler {


nit: add @VisiableForTesting ?

Ok, let me try

sijie · 2019-11-06T09:43:46Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ProducerConfigurationData.java

@@ -57,6 +57,7 @@

    private String topicName = null;
    private String producerName = null;
+    private boolean isGeneratedName = true;


do we really need adding another field in ProducerConfigurationData?

sijie · 2019-11-06T09:47:03Z

pulsar-common/src/main/proto/PulsarApi.proto

+
+    // Indicate the name of the producer is generated or not(user specified)
+    // Use default false here is in order to be forward compatible with the client
+    optional bool is_generated_name = 9 [default = false];


is_generated_name => user_provided_producer_name ?

sijie · 2019-11-06T10:01:41Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

-                throw new NamingException(
-                        "Producer with name '" + producer.getProducerName() + "' is already connected to topic");
+                boolean canOverwrite = false;
+                for (Producer existProducer : producers.values()) {


Can we avoid iterating over the producers set? I.e. can you change the hash set to a hash map?

codelipenghui · 2019-11-06T11:06:43Z

@sijie I have addressed your comments, please take a look again.

sijie · 2019-11-06T13:27:20Z

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

@@ -139,7 +140,9 @@ public String toString() {

    @Override
    public ConcurrentOpenHashSet<Producer> getProducers() {
-        return producers;
+        ConcurrentOpenHashSet<Producer> result = new ConcurrentOpenHashSet<>(16, 1);


oops. I didn't realize that there is a getProducers here. It is used in a lot of places. hence it might be a performance problem if we changed to a hash map.

I have checked the used places of method getProducers()

Looks ok, two places used it, handle backlog exceeded backlog and prometheus metrics aggregator, others are unit tests.

@sijie i will try change the getProducers method to return a map.

codelipenghui · 2019-11-07T02:21:00Z

run java8 tests

codelipenghui · 2019-11-07T06:00:16Z

run java8 tests

sijie · 2019-11-07T15:25:43Z

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

+                    canOverwrite = true;
+                }
+                if (canOverwrite) {
+                    producers.put(producer.getProducerName(), producer);


use producers.replace(producer.getProducerName(), existingProducer, producer) to make sure one can successfully add the producer.

sijie · 2019-11-07T15:26:19Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

+                    canOverwrite = true;
+                }
+                if (canOverwrite) {
+                    producers.put(producer.getProducerName(), producer);


use replace and check the return result

sijie · 2019-11-08T07:15:50Z

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

+            oldProducer.close();
+            canOverwrite = true;
+        }
+        if (canOverwrite) {


nit: this can be simplified with

if (canOverwrite && !producers.replace(newProducer.getProducerName(), oldProducer, newProducer)) { throw new BrokerServiceException.NamingException( "Producer with name '" + newProducer.getProducerName() + "' is already connected }

May be if can simplified with

if (!canOverwrite || !producers.replace(newProducer.getProducerName(), oldProducer, newProducer)) { throw new BrokerServiceException.NamingException( "Producer with name '" + newProducer.getProducerName() + "' is already connected") }

Right?

sijie · 2019-11-08T07:17:02Z

@merlimat @rdhabalia PTAL

codelipenghui · 2019-11-08T10:26:17Z

run java8 tests

codelipenghui · 2019-11-09T01:16:38Z

run cpp tests
run java8 tests

codelipenghui · 2019-11-10T00:43:54Z

run java8 tests

codelipenghui · 2019-11-10T06:48:19Z

run java8 tests

codelipenghui · 2019-11-10T14:16:02Z

run java8 tests

codelipenghui · 2019-11-11T01:59:42Z

run java8 tests

sijie · 2019-11-11T05:44:31Z

@merlimat @rdhabalia can you review this pull request?

sijie · 2019-11-11T05:45:16Z

@codelipenghui can you also create github issues for tracking adding similar fixes for other language clients?

codelipenghui · 2019-11-11T06:14:08Z

@sijie I have added a task tracker #5606 for other language catch up.

sijie · 2019-11-12T08:57:01Z

@merlimat @rdhabalia can you please review this?

wolfstudy · 2019-11-13T03:19:24Z

@codelipenghui l will change the Milestone to 2.4.3. So we can cut 2.4.2 and if needed
2.4.3 in a few weeks.

wolfstudy · 2019-11-19T13:03:22Z

@codelipenghui @jiazhai The pr changes the proto file, will move the milestone to 2.5.0

…5701) Fixes #5698 Motivation Since #5571 handle the generated producer name, the replicator producer was created by broker, only one producer for a replicated topic. So, we can handle it simple by considered replicator producer as generated name producer.

Add epoch for connection handler to handle create producer timeout.

08fd3ed

codelipenghui requested review from sijie, merlimat, wolfstudy and jiazhai November 6, 2019 09:11

codelipenghui self-assigned this Nov 6, 2019

sijie added the area/client label Nov 6, 2019

sijie added this to the 2.4.2 milestone Nov 6, 2019

sijie requested changes Nov 6, 2019

View reviewed changes

Fix comments

edae5d1

sijie reviewed Nov 6, 2019

View reviewed changes

codelipenghui added 2 commits November 7, 2019 08:16

Avoid iterating over the producers map

760b5b3

Fix unit tests issue

38ebf2a

sijie requested changes Nov 7, 2019

View reviewed changes

Add concurrent update check.

56764ce

sijie approved these changes Nov 8, 2019

View reviewed changes

sijie requested a review from rdhabalia November 8, 2019 07:16

Fix unit tests.

7fc26ea

codelipenghui mentioned this pull request Nov 11, 2019

Handle create producer timeout on other language client #5606

Closed

wolfstudy modified the milestones: 2.4.2, 2.4.3 Nov 13, 2019

jiazhai approved these changes Nov 13, 2019

View reviewed changes

jiazhai merged commit 75c7229 into apache:master Nov 19, 2019

jiazhai modified the milestones: 2.4.3, 2.4.2 Nov 19, 2019

codelipenghui mentioned this pull request Nov 19, 2019

Geo replicator has ProducerBusyException #5698

Closed

wolfstudy modified the milestones: 2.4.2, 2.5.0 Nov 19, 2019

weishuisheng mentioned this pull request Nov 19, 2019

[issues 5698]Handle replicator producer as generated name producer #5701

Merged

sijie mentioned this pull request Dec 27, 2019

ISSUE-5606: Handle create producer timeout on other language client streamnative/pulsar-archived#311

Open

addisonj mentioned this pull request Feb 25, 2020

In some instances, producers fail to create after a topic is scheduled onto previously used broker #6416

Closed

sijie mentioned this pull request Feb 25, 2020

ISSUE-6416: In some instances, producers fail to create after a topic is scheduled onto previously used broker streamnative/pulsar-archived#747

Closed

jiazhai mentioned this pull request Sep 28, 2020

Failed to add producer to topic（C language Client） #8124

Closed

codelipenghui deleted the issue-5535 branch March 4, 2021 03:07

codelipenghui mentioned this pull request Mar 4, 2021

Producer cannot connect after broker load shedding #9792

Closed

lhotari mentioned this pull request Apr 20, 2021

[Java Client] Fix concurrency issue in client's producer epoch handling #10278

Merged

freeznet mentioned this pull request Aug 10, 2021

Duplicate producer when create producer timeout apache/pulsar-client-go#584

Closed

sijie mentioned this pull request Aug 10, 2021

ISSUE-584: Duplicate producer when create producer timeout streamnative/pulsar-client-go#205

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add epoch for connection handler to handle create producer timeout. #5571

Add epoch for connection handler to handle create producer timeout. #5571

codelipenghui commented Nov 6, 2019

codelipenghui commented Nov 6, 2019

sijie commented Nov 6, 2019

codelipenghui commented Nov 6, 2019

sijie left a comment

sijie Nov 6, 2019

codelipenghui Nov 6, 2019

sijie Nov 6, 2019

codelipenghui Nov 6, 2019

sijie Nov 6, 2019

sijie Nov 6, 2019

sijie Nov 6, 2019

codelipenghui commented Nov 6, 2019

sijie Nov 6, 2019

codelipenghui Nov 6, 2019 •

edited

codelipenghui Nov 6, 2019

codelipenghui commented Nov 7, 2019

codelipenghui commented Nov 7, 2019

sijie Nov 7, 2019

sijie Nov 7, 2019

sijie Nov 8, 2019

codelipenghui Nov 8, 2019 •

edited

sijie commented Nov 8, 2019

codelipenghui commented Nov 8, 2019

codelipenghui commented Nov 9, 2019

codelipenghui commented Nov 10, 2019

codelipenghui commented Nov 10, 2019

codelipenghui commented Nov 10, 2019

codelipenghui commented Nov 11, 2019

sijie commented Nov 11, 2019

sijie commented Nov 11, 2019

codelipenghui commented Nov 11, 2019

sijie commented Nov 12, 2019

wolfstudy commented Nov 13, 2019

wolfstudy commented Nov 19, 2019

Add epoch for connection handler to handle create producer timeout. #5571

Add epoch for connection handler to handle create producer timeout. #5571

Conversation

codelipenghui commented Nov 6, 2019

Motivation

Modifications

Verifying this change

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

Documentation

codelipenghui commented Nov 6, 2019

sijie commented Nov 6, 2019

codelipenghui commented Nov 6, 2019

sijie 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

codelipenghui commented Nov 6, 2019

Choose a reason for hiding this comment

codelipenghui Nov 6, 2019 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codelipenghui commented Nov 7, 2019

codelipenghui commented Nov 7, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codelipenghui Nov 8, 2019 • edited

Choose a reason for hiding this comment

sijie commented Nov 8, 2019

codelipenghui commented Nov 8, 2019

codelipenghui commented Nov 9, 2019

codelipenghui commented Nov 10, 2019

codelipenghui commented Nov 10, 2019

codelipenghui commented Nov 10, 2019

codelipenghui commented Nov 11, 2019

sijie commented Nov 11, 2019

sijie commented Nov 11, 2019

codelipenghui commented Nov 11, 2019

sijie commented Nov 12, 2019

wolfstudy commented Nov 13, 2019

wolfstudy commented Nov 19, 2019

codelipenghui Nov 6, 2019 •

edited

codelipenghui Nov 8, 2019 •

edited