[fix] [broker] Make specified producer could override the previous one #21155

poorbarcode · 2023-09-11T01:02:57Z

Motivation

2023-09-12T07:17:05,388+0000 [pulsar-web-37-4] INFO  org.apache.pulsar.broker.service.Producer - Disconnecting producer: Producer{topic=PersistentTopic{topic=persistent://public/default/tp1}, client=/127.0.0.6:36433, producerName=st-0-5, producerId=87}
2023-09-12T07:38:55,707+0000 [pulsar-io-4-1] INFO  org.apache.pulsar.broker.service.ServerCnx - [/127.0.0.6:37543][persistent://public/default/tp1] Creating producer. producerId=87
2023-09-12T07:38:55,707+0000 [pulsar-io-4-1] WARN  org.apache.pulsar.broker.service.ServerCnx - [/127.0.0.6:37543] Failed to add producer to topic persistent://public/default/tp1: producerId=87, Producer with name 'st-0-5' is already connected to topic
2023-09-12T07:38:55,916+0000 [pulsar-io-4-1] INFO  org.apache.pulsar.broker.service.ServerCnx - [/127.0.0.6:37543][persistent://public/default/tp1] Creating producer. producerId=87
2023-09-12T07:38:55,916+0000 [pulsar-io-4-1] WARN  org.apache.pulsar.broker.service.ServerCnx - [/127.0.0.6:37543] Failed to add producer to topic persistent://public/default/tp1: producerId=87, Producer with name 'st-0-5' is already connected to topic
...
...
...
2023-09-12T07:39:47,395+0000 [pulsar-io-4-1] INFO  org.apache.pulsar.broker.service.ServerCnx - Closed connection from /127.0.0.6:36433
2023-09-12T07:39:47,395+0000 [pulsar-io-4-1] WARN  org.apache.pulsar.common.protocol.PulsarHandler - [[id: 0x0f0484da, L:/10.48.3.67:6652 - R:/127.0.0.6:36433]] Forcing connection to close after keep-alive timeout
2023-09-12T07:40:12,161+0000 [pulsar-io-4-1] INFO  org.apache.pulsar.broker.service.ServerCnx - [/127.0.0.6:37543][persistent://public/default/tp1] Creating producer. producerId=87
2023-09-12T07:40:12,161+0000 [pulsar-io-4-1] INFO  org.apache.pulsar.broker.service.ServerCnx - [/127.0.0.6:37543] Created new producer: Producer{topic=PersistentTopic{topic=persistent://public/default/tp1}, client=/127.0.0.6:37543, producerName=st-0-5, producerId=87}

Issue 1

The client assumed the connection was inactive, but the Broker assumed the connection was fine. The Client tried to use a new connection to reconnect a producer, then got an error Producer with name 'st-0-5' is already connected to topic.

Issue 2

In a connection, the second connection waits for the first connection to complete^[1]. But there is a bug that causes this mechanism to fail^[2].
If a producer uses a default name, the second registration will override the first one. But it can not override the first one if it uses a specified producer name^[3]. I think this mechanism is to prevent a client from creating two producers with the same name. However, method Producer.isSuccessorTo has checked the producer-id, and the producer-id of multiple producers created by the same client are different. So this mechanism can be deleted.

[1]
https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java#L1398-L1423

[2]
https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java#L1397-L1425

CompletableFuture<Producer> existingProducerFuture = producers.putIfAbsent(producerId, producerFuture);

if (existingProducerFuture != null) {
    if (existingProducerFuture.isDone() && !existingProducerFuture.isCompletedExceptionally()) {
        // step-1: If the existing producer's future is done and succeded responds success
       sendProducerSuccessResponse(...);
       return null;
    } else {
        if (!existingProducerFuture.isDone()) {
             // step-2: If the existing producer's future is not, responds fail
            error = ServerError.ServiceNotReady;
            sendErrorResponse(...);
            return null;
        } else {          
             // step-3: else: responds fail
            producers.remove(producerId, existingProducerFuture);
            sendErrorResponse(...);
            return null;
        }
    }
}

If the future is successful after the check step-1, the pulsar will remove a completed future from serverCnc.producers.

[3]
https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java#L983-L999

private void tryOverwriteOldProducer(Producer oldProducer, Producer newProducer)
            throws BrokerServiceException {
    if (newProducer.isSuccessorTo(oldProducer) && !isUserProvidedProducerName(oldProducer)
            && !isUserProvidedProducerName(newProducer)) {
        oldProducer.close(false);
        if (!producers.replace(newProducer.getProducerName(), oldProducer, newProducer)) {
            // Met concurrent update, throw exception here so that client can try reconnect later.
            throw new BrokerServiceException.NamingException("Producer with name '" + newProducer.getProducerName()
                    + "' replace concurrency error");
        } else {
            handleProducerRemoved(oldProducer);
        }
    } else {
        throw new BrokerServiceException.NamingException(
                "Producer with name '" + newProducer.getProducerName() + "' is already connected to topic");
    }
}

Modifications

For issue 1: If a producer with the same name tries to use a new connection, async checks the old connection is available. The producers related to the connection that is not available are automatically cleaned up.
For issue 2:
- Fix the bug that causes a complete producer future will be removed from ServerCnx.
- Remove the mechanism that prevents a producer with a specified name from overriding the previous producer.

Documentation

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

Matching PR in forked repository

PR in forked repository: x

codelipenghui · 2023-09-11T02:08:15Z

If the question from this comment is not a problem, I think this fix will work because cnx and producer ID is enough to determine the producer instances from one logic producer or not.

poorbarcode · 2023-09-11T03:25:48Z

@codelipenghui

If the question from this #21144 (comment) is not a problem, I think this fix will work because cnx and producer ID is enough to determine the producer instances from one logic producer or not.

I added a new mechanism: If a producer with the same name tries to use a new connection, Pulsar will async check whether the old connection is available. The producers will be automatically cleaned up if the old connection is unavailable.

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

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

#21155) The client assumed the connection was inactive, but the Broker assumed the connection was fine. The Client tried to use a new connection to reconnect a producer, then got an error `Producer with name 'st-0-5' is already connected to topic`. - In a connection, the second connection waits for the first connection to complete\. But there is a bug that causes this mechanism to fail\. - If a producer uses a default name, the second registration will override the first one. But it can not override the first one if it uses a specified producer name\. I think this mechanism is to prevent a client from creating two producers with the same name. However, method `Producer.isSuccessorTo` has checked the `producer-id`, and the `producer-id` of multiple producers created by the same client are different. So this mechanism can be deleted. - For `issue 1`: If a producer with the same name tries to use a new connection, async checks the old connection is available. The producers related to the connection that is not available are automatically cleaned up. - For `issue 2`: - Fix the bug that causes a complete producer future will be removed from `ServerCnx`. - Remove the mechanism that prevents a producer with a specified name from overriding the previous producer. (cherry picked from commit bda16b6)

poorbarcode · 2023-09-14T04:13:48Z

I want to cherry-pick the tool method Servercnx.checkConnectionLiveness( introduced in the PR
#20026) into branch-2.10 and branch-2.11 along with the current PR.

See the discussion for details.

#21155) The client assumed the connection was inactive, but the Broker assumed the connection was fine. The Client tried to use a new connection to reconnect a producer, then got an error `Producer with name 'st-0-5' is already connected to topic`. - In a connection, the second connection waits for the first connection to complete\. But there is a bug that causes this mechanism to fail\. - If a producer uses a default name, the second registration will override the first one. But it can not override the first one if it uses a specified producer name\. I think this mechanism is to prevent a client from creating two producers with the same name. However, method `Producer.isSuccessorTo` has checked the `producer-id`, and the `producer-id` of multiple producers created by the same client are different. So this mechanism can be deleted. - For `issue 1`: If a producer with the same name tries to use a new connection, async checks the old connection is available. The producers related to the connection that is not available are automatically cleaned up. - For `issue 2`: - Fix the bug that causes a complete producer future will be removed from `ServerCnx`. - Remove the mechanism that prevents a producer with a specified name from overriding the previous producer. (cherry picked from commit bda16b6)

poorbarcode · 2023-09-18T08:58:10Z

I want to cherry-pick the tool method Servercnx.checkConnectionLiveness( introduced in the PR
#20026) into branch-2.10 and branch-2.11 along with the current PR.
See the discussion for details.

Cherry-picked

branch-2.10: fbf87df
branch-2.11: f6b43b3

michaeljmarshall · 2023-09-18T15:20:18Z

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

+            // If a producer with the same name tries to use a new connection, async check the old connection is
+            // available. The producers related the connection that not available are automatically cleaned up.
+            if (!Objects.equals(oldProducer.getCnx(), newProducer.getCnx())) {
+                oldProducer.getCnx().checkConnectionLiveness();


This returns a future that can be used to prevent unnecessary exceptions. Did you consider wiring that up?

@michaeljmarshall

I submitted a new PR to improve this behavior. Please help to review it

[improve] [broker] Let the producer request success at the first time if the previous one is inactive #21220

BewareMyPower · 2023-10-16T01:46:12Z

I think this PR should be cherry-picked into branch-3.1, right?

#21155) The client assumed the connection was inactive, but the Broker assumed the connection was fine. The Client tried to use a new connection to reconnect a producer, then got an error `Producer with name 'st-0-5' is already connected to topic`. - In a connection, the second connection waits for the first connection to complete\. But there is a bug that causes this mechanism to fail\. - If a producer uses a default name, the second registration will override the first one. But it can not override the first one if it uses a specified producer name\. I think this mechanism is to prevent a client from creating two producers with the same name. However, method `Producer.isSuccessorTo` has checked the `producer-id`, and the `producer-id` of multiple producers created by the same client are different. So this mechanism can be deleted. - For `issue 1`: If a producer with the same name tries to use a new connection, async checks the old connection is available. The producers related to the connection that is not available are automatically cleaned up. - For `issue 2`: - Fix the bug that causes a complete producer future will be removed from `ServerCnx`. - Remove the mechanism that prevents a producer with a specified name from overriding the previous producer. (cherry picked from commit bda16b6)

poorbarcode · 2023-10-16T03:38:33Z

I think this PR should be cherry-picked into branch-3.1, right?

Done

…ne faster (#21183) ### Motivation There is an issue similar to the #21155 fixed one. The client assumed the connection was inactive, but the Broker assumed the connection was fine. The Client tried to use a new connection to reconnect an exclusive consumer, then got an error `Exclusive consumer is already connected` ### Modifications - Check the connection of the old consumer is available when the new one tries to subscribe

…ne faster (#21183) ### Motivation There is an issue similar to the #21155 fixed one. The client assumed the connection was inactive, but the Broker assumed the connection was fine. The Client tried to use a new connection to reconnect an exclusive consumer, then got an error `Exclusive consumer is already connected` ### Modifications - Check the connection of the old consumer is available when the new one tries to subscribe (cherry picked from commit 29db8f8)

…ne faster (#21183) There is an issue similar to the #21155 fixed one. The client assumed the connection was inactive, but the Broker assumed the connection was fine. The Client tried to use a new connection to reconnect an exclusive consumer, then got an error `Exclusive consumer is already connected` - Check the connection of the old consumer is available when the new one tries to subscribe (cherry picked from commit 29db8f8)

apache#21155) The client assumed the connection was inactive, but the Broker assumed the connection was fine. The Client tried to use a new connection to reconnect a producer, then got an error `Producer with name 'st-0-5' is already connected to topic`. - In a connection, the second connection waits for the first connection to complete\. But there is a bug that causes this mechanism to fail\. - If a producer uses a default name, the second registration will override the first one. But it can not override the first one if it uses a specified producer name\. I think this mechanism is to prevent a client from creating two producers with the same name. However, method `Producer.isSuccessorTo` has checked the `producer-id`, and the `producer-id` of multiple producers created by the same client are different. So this mechanism can be deleted. - For `issue 1`: If a producer with the same name tries to use a new connection, async checks the old connection is available. The producers related to the connection that is not available are automatically cleaned up. - For `issue 2`: - Fix the bug that causes a complete producer future will be removed from `ServerCnx`. - Remove the mechanism that prevents a producer with a specified name from overriding the previous producer. (cherry picked from commit bda16b6)

…ne faster (apache#21183) There is an issue similar to the apache#21155 fixed one. The client assumed the connection was inactive, but the Broker assumed the connection was fine. The Client tried to use a new connection to reconnect an exclusive consumer, then got an error `Exclusive consumer is already connected` - Check the connection of the old consumer is available when the new one tries to subscribe (cherry picked from commit 29db8f8) (cherry picked from commit b796f56)

[fix] [broker] Make specified producer could override the previous one

5f74e48

github-actions bot added the doc-not-needed Your PR changes do not impact docs label Sep 11, 2023

poorbarcode self-assigned this Sep 11, 2023

poorbarcode added release/3.0.2 release/2.11.3 release/2.10.6 category/reliability The function does not work properly in certain specific environments or failures. e.g. data lost labels Sep 11, 2023

poorbarcode added this to the 3.2.0 milestone Sep 11, 2023

Close connections that not available

a5ba726

codelipenghui mentioned this pull request Sep 11, 2023

[fix] [client] fix same producer/consumer use more than one connection per broker #21144

Merged

4 tasks

codelipenghui reviewed Sep 11, 2023

View reviewed changes

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

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

liangyepianzhou approved these changes Sep 12, 2023

View reviewed changes

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

poorbarcode added 2 commits September 12, 2023 09:19

fix wrong logic

8916452

add test

2136266

poorbarcode requested a review from codelipenghui September 12, 2023 02:02

add test testHandleProducerAfterClientChannelInactive

66491fe

codelipenghui approved these changes Sep 13, 2023

View reviewed changes

remove unnecassary import

3b968ad

poorbarcode merged commit bda16b6 into apache:master Sep 13, 2023
44 of 45 checks passed

poorbarcode added the cherry-picked/branch-3.0 label Sep 14, 2023

poorbarcode mentioned this pull request Sep 14, 2023

[improve][broker] Prevent range conflicts with Key Shared sticky consumers when TCP/IP connections get orphaned #20026

Merged

4 tasks

poorbarcode mentioned this pull request Sep 14, 2023

[fix] [broker] Make the new exclusive consumer instead the inactive one faster #21183

Merged

4 tasks

poorbarcode added cherry-picked/branch-2.10 cherry-picked/branch-2.11 labels Sep 18, 2023

michaeljmarshall reviewed Sep 18, 2023

View reviewed changes

poorbarcode mentioned this pull request Sep 21, 2023

[improve] [broker] Let the producer request success at the first time if the previous one is inactive #21220

Merged

4 tasks

poorbarcode added cherry-picked/branch-3.1 release/3.1.2 labels Oct 16, 2023

Technoboy- mentioned this pull request Jan 3, 2024

Handle already have connected replicate produce with same name #5104

Closed

Technoboy- mentioned this pull request Jan 10, 2024

[Bug] [worker] ProducerFencedException：Topic has an existing exclusive producer #21809

Closed

2 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[fix] [broker] Make specified producer could override the previous one #21155

[fix] [broker] Make specified producer could override the previous one #21155

poorbarcode commented Sep 11, 2023 •

edited

Loading

codelipenghui commented Sep 11, 2023

poorbarcode commented Sep 11, 2023

poorbarcode commented Sep 14, 2023

poorbarcode commented Sep 18, 2023

michaeljmarshall Sep 18, 2023

poorbarcode Sep 21, 2023

BewareMyPower commented Oct 16, 2023

poorbarcode commented Oct 16, 2023

[fix] [broker] Make specified producer could override the previous one #21155

[fix] [broker] Make specified producer could override the previous one #21155

Conversation

poorbarcode commented Sep 11, 2023 • edited Loading

Motivation

Issue 1

Issue 2

Modifications

Documentation

Matching PR in forked repository

codelipenghui commented Sep 11, 2023

poorbarcode commented Sep 11, 2023

poorbarcode commented Sep 14, 2023

poorbarcode commented Sep 18, 2023

michaeljmarshall Sep 18, 2023

Choose a reason for hiding this comment

poorbarcode Sep 21, 2023

Choose a reason for hiding this comment

BewareMyPower commented Oct 16, 2023

poorbarcode commented Oct 16, 2023

poorbarcode commented Sep 11, 2023 •

edited

Loading