[improve][txn][PIP-298] Consumer supports specifying consumption isolation level #21246

hzh0425 · 2023-09-25T12:59:46Z

Motivation

Consumer supports specifying consumption isolation level #21114

Modifications

Add a configuration 'SubscriptionIsolationLevel' in consumer builder:

PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://localhost:6650").build();
Consumer<String> consumer = client.newConsumer(Schema.STRING)
        .topic("persistent://my-tenant/my-namespace/my-topic")
        .subscriptionName("my-subscription")
        .subscriptionType(SubscriptionType.Shared)
        .subscriptionIsolationLevel(IsolationLevel.READ_COMMITTED) // Adding the isolation level configuration
        .subscribe();

Verifying this change

Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

Added integration tests for end-to-end deployment with large payloads (10MB)
Extended integration test for recovery after broker failure

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:

hzh0425 · 2023-09-25T13:00:39Z

@liangyepianzhou pls cc

liangyepianzhou · 2023-10-24T08:02:11Z

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

@@ -399,6 +399,14 @@ message CommandSubscribe {

    // The consumer epoch, when exclusive and failover consumer redeliver unack message will increase the epoch
    optional uint64 consumer_epoch = 19;
+
+    enum IsolationLevel {


@codelipenghui Should we increase ProtocolVersion?

Yes. The broker should only read the isolation type from a client version that has isolation-level support. Otherwise, the broker will get an incompatibility issue.

We should also add an integration test to https://github.com/apache/pulsar/tree/master/tests/integration
And we should also add a compatibility test like https://github.com/apache/pulsar/tree/master/tests/bc_2_6_0

.../java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

...va/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java

…bscription isolation level is ReadUnCommitted.

hzh0425 · 2023-10-26T13:38:29Z

@codelipenghui

Hi, could you please help with the review as soon as possible? We are currently very urgent
Thank you for your time

poorbarcode · 2023-10-26T14:27:52Z

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

+     *
+     * @param subscriptionIsolationLevel If READ_COMMITTED is selected, the Consumer can only consume all transactional messages which have been committed,
+     *                                   else if READ_UNCOMMITTED is selected, the Consumer can consume all messages, even transactional messages which have been aborted.
+     *                                   Note that this is a subscription dimension configuration, and all consumers under the same subscription need to be configured with the same IsolationLevel.


These lines are too long, so the check of check style could not pass. Could you run mvn clean install -Dskiptests first to ensure the check can passed? Thanks

@codelipenghui
Hi, could you please help with the review as soon as possible? We are currently very urgent.
Thank you for your time

Since you are urgent, you should also focus on the failed CIs.

You can set up your IDE by following this doc: https://pulsar.apache.org/contribute/setup-ide/

The doc of Kafka also can be referred to.

@codelipenghui @hzh0425 @liangyepianzhou

In the current PR, the implementation is different with Kafka:

Kafka: the consumer will receive all messages, even if the messages have been aborted.

Current PR:

the consumer will receive committed messages;

the consumer will receive the messages that state transaction on-gonging, once the transaction was aborted, these messages could not be received.

We have discussed this question in the meeting. At that time, the consensus was that not receiving aborted messages was a better solution.

The original intention of using read uncommitted is to reduce read and write latency, not to read error messages

There are differences between Kafka and Pulsar in filtering the aborted message.

Kafka filters the aborted message on the client side, so if the client uses READ_UNCOMMITED, the client can't feel the ongoing transaction, the client is only able to filter the markers. If we need to filter the abort message on the kafka client, it requires to response of the abortedTransactions data each fetch, which will reduce the consumption performance.

Pulsar has different implements for filter parts, the filter logic is on the Broker side. If it is to reduce read and write latency, it is okay.

But I think reading all messages just like Kafka is beneficial in the following cases:
In the event of issues or during development and testing phases, having access to all messages, including uncommitted ones, can be valuable for debugging and troubleshooting.

poorbarcode

Left some comments. Please address them. Thanks

poorbarcode · 2023-10-26T15:18:09Z

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

+        }
+
+        if (properties.containsKey(SUBSCRIPTION_ISOLATION_LEVEL_PROPERTY)) {
+            IsolationLevel isolationLevel = IsolationLevel.valueOf(Integer.parseInt(properties.get(SUBSCRIPTION_ISOLATION_LEVEL_PROPERTY)));


Could we catch the NumberFormatException here? And please add a test for this case.

poorbarcode · 2023-10-26T15:21:44Z

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

    protected final LongAdder bytesOutFromRemovedConsumers = new LongAdder();
    protected final LongAdder msgOutFromRemovedConsumer = new LongAdder();

+    public static void wrapIsolationLevelToProperties(Map<String, String> properties, IsolationLevel isolationLevel) {
+        if (properties != null) {
+            properties.put(SUBSCRIPTION_ISOLATION_LEVEL_PROPERTY, String.valueOf(isolationLevel.getValue()));


Once the subscription was created, the prop SUBSCRIPTION_ISOLATION_LEVEL_PROPERTY can not be modified, right?

If yes, we should prevent modifying it by the API pulsar-admin topics update-subscription-properties. And please add a test for this case.

Sorry, why do we need to persist the isolation level to properties?
It should follow the subscription type. The subscription will use the isolation type that the first consumer provided.
After all consumer disconnected, a new consumer will get a chance to change the isolation level.

@codelipenghui

Sorry, why do we need to persist the isolation level to properties?
It should follow the subscription type. The subscription will use the isolation type that the first consumer provided.
After all consumer disconnected, a new consumer will get a chance to change the isolation level.

For example:

there are 10 messages

3:0~3:4(committed)

3:5~3:9(transaction on-going)]

create a subscription with reading un-committed isolation level

the consumer committed 3:0~3~6

close all consumers

create a new consumer on the old subscription with read committed isolation level, (Highlight) the read position and marked deleted position will go to a strange value

It should follow the subscription type. The subscription will use the isolation type that the first consumer provided.
After all consumer disconnected, a new consumer will get a chance to change the isolation level.

I agree that the subscription use the isolation type of the first consumer provided. But the isolation level should not be changed. Just like schema.

poorbarcode · 2023-10-26T15:28:52Z

...broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

@@ -160,6 +162,7 @@ public PersistentSubscription(PersistentTopic topic, String subscriptionName, Ma
            this.pendingAckHandle = new PendingAckHandleDisabled();
        }
        IS_FENCED_UPDATER.set(this, FALSE);
+        this.isolationLevel = fetchIsolationLevelFromProperties(subscriptionProperties);


Do we support users using a message that is read from a subscription typed READ_UNCOMMITTED for a new transaction?

If false, we should init the component pendingAckHandle typed PendingAckHandleDisabled when the isolationLevel is READ_UNCOMMITTED, right?

If yes, how can we skip the recover task for this pending ack?

And please add a test for this case.

poorbarcode · 2023-10-26T15:37:05Z

.../src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java

@@ -236,6 +240,11 @@ public String getTypeString() {
        return "Null";
    }

+    @Override
+    public IsolationLevel getIsolationLevel() {


Maybe use @lombok.Getter instead of this method is better?

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

poorbarcode · 2023-10-26T15:58:42Z

pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionConsumeTest.java

@@ -228,6 +229,89 @@ public void sortedTest() throws Exception {
        log.info("TransactionConsumeTest sortedTest finish.");
    }

+    @Test
+    public void testConsumeMessageWithDifferentIsolationLevel() throws Exception {


Could you also add an E2E test ( do not start & commit transaction by PersistentTopic, just use a producer to do it). and another two test cases:

abort this transaction before receiving messages

unload topic before receiving message

poorbarcode · 2023-10-26T15:59:39Z

pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionConsumeTest.java

+        }
+
+        // Now commit the transaction
+        persistentTopic.endTxn(txnID, TxnAction.COMMIT_VALUE, 0L).get();


Could you also add another test case: abort this transaction before receiving messages?

poorbarcode · 2023-10-26T16:03:30Z

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

@@ -45,6 +45,7 @@ public class SubscriptionOption {
    private Map<String, String> metadata;
    private boolean readCompacted;
    private CommandSubscribe.InitialPosition initialPosition;
+    private CommandSubscribe.IsolationLevel isolationLevel;


Can you modify this variable with final?

poorbarcode · 2023-10-26T16:05:06Z

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

 import java.util.Optional;
 import java.util.concurrent.atomic.LongAdder;
 import java.util.function.ToLongFunction;

 public abstract class AbstractSubscription implements Subscription {
+    protected static final String SUBSCRIPTION_ISOLATION_LEVEL_PROPERTY = "pulsar.subscription.isolation.level";


Since it is a prop for Transaction, we should make this prop name contain the keyword transaction, right?

And for all the system properties, please add __ as a prefix.

poorbarcode · 2023-10-26T16:07:08Z

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

    protected final LongAdder bytesOutFromRemovedConsumers = new LongAdder();
    protected final LongAdder msgOutFromRemovedConsumer = new LongAdder();

+    public static void wrapIsolationLevelToProperties(Map<String, String> properties, IsolationLevel isolationLevel) {
+        if (properties != null) {


If the argument properties is null, you discarded the prop isolationLevel, it is wrong. please add a test for this case, thanks

codelipenghui · 2023-10-27T07:20:22Z

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

 import java.util.Optional;
 import java.util.concurrent.atomic.LongAdder;
 import java.util.function.ToLongFunction;

 public abstract class AbstractSubscription implements Subscription {
+    protected static final String SUBSCRIPTION_ISOLATION_LEVEL_PROPERTY = "pulsar.subscription.isolation.level";


And for all the system properties, please add __ as a prefix.

codelipenghui · 2023-10-27T07:23:06Z

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

    protected final LongAdder bytesOutFromRemovedConsumers = new LongAdder();
    protected final LongAdder msgOutFromRemovedConsumer = new LongAdder();

+    public static void wrapIsolationLevelToProperties(Map<String, String> properties, IsolationLevel isolationLevel) {
+        if (properties != null) {
+            properties.put(SUBSCRIPTION_ISOLATION_LEVEL_PROPERTY, String.valueOf(isolationLevel.getValue()));


Sorry, why do we need to persist the isolation level to properties?
It should follow the subscription type. The subscription will use the isolation type that the first consumer provided.
After all consumer disconnected, a new consumer will get a chance to change the isolation level.

codelipenghui · 2023-10-27T07:29:46Z

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

+     *
+     * @param subscriptionIsolationLevel If READ_COMMITTED is selected, the Consumer can only consume all transactional messages which have been committed,
+     *                                   else if READ_UNCOMMITTED is selected, the Consumer can consume all messages, even transactional messages which have been aborted.
+     *                                   Note that this is a subscription dimension configuration, and all consumers under the same subscription need to be configured with the same IsolationLevel.


The doc of Kafka also can be referred to.

codelipenghui · 2023-10-27T07:37:48Z

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

@@ -399,6 +399,14 @@ message CommandSubscribe {

    // The consumer epoch, when exclusive and failover consumer redeliver unack message will increase the epoch
    optional uint64 consumer_epoch = 19;
+
+    enum IsolationLevel {


Yes. The broker should only read the isolation type from a client version that has isolation-level support. Otherwise, the broker will get an incompatibility issue.

We should also add an integration test to https://github.com/apache/pulsar/tree/master/tests/integration
And we should also add a compatibility test like https://github.com/apache/pulsar/tree/master/tests/bc_2_6_0

codelipenghui · 2024-05-28T01:12:15Z

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

+ */
+@InterfaceAudience.Public
+@InterfaceStability.Stable
+public enum SubscriptionIsolationLevel {


We should use TransactionIsolationLevel. We don't have any concept of SubscriptionIsolationLevel.

hzh0425 added 2 commits September 25, 2023 20:48

Add isolationLevel in consumer side.

e13370d

Add isolationLevel logic in broker side.

05a5c46

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

Add stats 'subscriptionIsolationLevel' in SubscriptionStatsImpl

7eee90c

hzh0425 force-pushed the pip-298-code branch from a4c4db2 to 7eee90c Compare October 7, 2023 15:31

liangyepianzhou approved these changes Oct 24, 2023

View reviewed changes

liangyepianzhou requested review from codelipenghui and poorbarcode October 24, 2023 08:25

return 'topic.getManagedLedger().getLastConfirmedEntry()' when the su…

4712979

…bscription isolation level is ReadUnCommitted.

poorbarcode reviewed Oct 26, 2023

View reviewed changes

poorbarcode assigned hzh0425 Oct 26, 2023

poorbarcode added this to the 3.2.0 milestone Oct 26, 2023

poorbarcode reviewed Oct 26, 2023

View reviewed changes

poorbarcode added type/feature The PR added a new feature or issue requested a new feature area/transaction labels Oct 26, 2023

liangyepianzhou changed the title ~~[improve][broker/consumer][PIP-298] Consumer supports specifying consumption isolation level~~ [improve][txn][PIP-298] Consumer supports specifying consumption isolation level Oct 27, 2023

codelipenghui requested changes Oct 27, 2023

View reviewed changes

Technoboy- modified the milestones: 3.2.0, 3.3.0 Dec 22, 2023

coderzc modified the milestones: 3.3.0, 3.4.0 May 8, 2024

visortelle mentioned this pull request May 27, 2024

[improve][doc]Add introduction of transaction isolation level in the txn-advanced-features.md apache/pulsar-site#906

Draft

1 task

shibd mentioned this pull request May 28, 2024

[improve][cli] PIP-353: Improve transaction message visibility for peek-message #22762

Merged

14 tasks

codelipenghui reviewed May 28, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[improve][txn][PIP-298] Consumer supports specifying consumption isolation level #21246

[improve][txn][PIP-298] Consumer supports specifying consumption isolation level #21246

hzh0425 commented Sep 25, 2023 •

edited

Loading

hzh0425 commented Sep 25, 2023

liangyepianzhou Oct 24, 2023

codelipenghui Oct 27, 2023

hzh0425 commented Oct 26, 2023

poorbarcode Oct 26, 2023 •

edited

Loading

poorbarcode Oct 26, 2023

poorbarcode Oct 26, 2023

codelipenghui Oct 27, 2023

poorbarcode Oct 27, 2023 •

edited

Loading

liangyepianzhou Oct 27, 2023

liangyepianzhou Oct 27, 2023

Demogorgon314 Oct 27, 2023

poorbarcode left a comment

poorbarcode Oct 26, 2023

poorbarcode Oct 26, 2023

codelipenghui Oct 27, 2023

poorbarcode Oct 27, 2023

liangyepianzhou Oct 27, 2023

poorbarcode Oct 26, 2023

poorbarcode Oct 26, 2023

poorbarcode Oct 26, 2023

poorbarcode Oct 26, 2023

poorbarcode Oct 26, 2023

poorbarcode Oct 26, 2023 •

edited

Loading

codelipenghui Oct 27, 2023

poorbarcode Oct 26, 2023 •

edited

Loading

codelipenghui Oct 27, 2023

codelipenghui Oct 27, 2023

codelipenghui Oct 27, 2023

codelipenghui Oct 27, 2023

codelipenghui May 28, 2024

[improve][txn][PIP-298] Consumer supports specifying consumption isolation level #21246

Are you sure you want to change the base?

[improve][txn][PIP-298] Consumer supports specifying consumption isolation level #21246

Conversation

hzh0425 commented Sep 25, 2023 • edited Loading

Motivation

Modifications

Verifying this change

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

Documentation

Matching PR in forked repository

hzh0425 commented Sep 25, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hzh0425 commented Oct 26, 2023

poorbarcode Oct 26, 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

poorbarcode Oct 27, 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

poorbarcode 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

poorbarcode Oct 26, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

poorbarcode Oct 26, 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hzh0425 commented Sep 25, 2023 •

edited

Loading

poorbarcode Oct 26, 2023 •

edited

Loading

poorbarcode Oct 27, 2023 •

edited

Loading

poorbarcode Oct 26, 2023 •

edited

Loading

poorbarcode Oct 26, 2023 •

edited

Loading