[PIP-54] Support acknowledgment at batch index level #6052

codelipenghui · 2020-01-14T13:34:57Z

Master issue: #6253
Fixes #5969

Motivation

Add support for ack batch message local index. Can be disabled at broker side by set batchIndexAcknowledgeEnable=false at broker.conf

PIP-54 documentation will be created soon.

Modifications

Managed cursor support track and persistent local index of batch message.
Client support send batch index ack to broker.
The batch messages with index ack information dispatched to the client.
Client skip the acked index.

Verifying this change

New unit tests added

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: (yes)
The rest endpoints: (no)
The admin cli options: (no)
Anything that affects deployment: (no)

Documentation

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

codelipenghui · 2020-01-17T07:21:46Z

run java8 tests

sijie

@codelipenghui thank you for your work for introducing this great feature. Before going into details, I would like to first discuss the "ack set" introduced in the cusor storage information and wire protocol. I have some suggestions there. once we agree on how the data structure would look like, I will review the implementation.

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java

managed-ledger/src/main/proto/MLDataFormats.proto

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

zzzming

@codelipenghui Thanks for implementing batch acknowledgement. This is a complex task. Would you expand this PR to C++ library? I would suggest to add more documentation to explain the expected haviour and supported or non-supported consumer subscriptions; also the performance considerations like whether re-transmission of the entire batch of partially acked messages. Thanks!

zzzming · 2020-02-13T14:24:32Z

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java

+                .newBuilder();
+            MLDataFormats.BatchDeletedIndexInfo.Builder batchDeletedIndexInfoBuilder = MLDataFormats.BatchDeletedIndexInfo
+                .newBuilder();
+            List<MLDataFormats.BatchDeletedIndexInfo> result = Lists.newArrayList();


It seems that a large code section has been locked/unlocked. You might want to consider to move these initialization outside of protect section.
Isn't ConcurrentSkipListMap already thread safe when you read? Probably only the iteration should be protected. I think the goal is avoid lock a large section of code not to let another thread to wait on.

zzzming · 2020-02-13T14:27:53Z

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java

+    public long[] getDeletedBatchIndexesLongArray(PositionImpl position) {
+        lock.readLock().lock();
+        try {
+            BitSet bitSet = batchDeletedIndexes.get(position);


Ditto. I think only batchDeleteIndexes.get() part should be locked. Again, double check with concurrentSkipList get() if it is already a thread safe data struct in terms of read. I think it should although I have never used this map collection.

zzzming · 2020-02-13T15:30:56Z

pulsar-common/src/main/java/org/apache/pulsar/common/util/SafeCollectionUtils.java

+            List<Long> result = new ArrayList<>(array.length);
+            for (long l : array) {
+                result.add(l);
+            }
+            return result;


How about using stream in Java 8?
return Arrays.stream(a).boxed().collect(Collectors.toList())

Actually the entire function can be
return array == null ? Collections.emptyList() : Arrays.stream(a).boxed().collect(Collectors.toList())

Collections.emptyList() returns an immutable list however new ArrayList is mutable. So I do not know whether you would like to have a mutable or immutable list. This needs to be clarified.

zzzming · 2020-02-13T15:49:42Z

pulsar-common/src/main/java/org/apache/pulsar/common/util/SafeCollectionUtils.java

+        if (list == null || list.size() == 0) {
+            return new long[0];
+        } else {
+            long[] array = new long[list.size()];
+            for (int i = 0; i < list.size(); i++) {
+                array[i] = list.get(i);
+            }
+            return array;
+        }


Again consider using stream
return list == null ? new long[0] : list.stream().mapToLong(l->l).toArray()

Probably there is no need to have a file to deal with primitive type array and list conversion. Does Apache common lang or Guava have library for that?

zzzming · 2020-02-13T16:00:53Z

pulsar-common/src/main/java/org/apache/pulsar/common/api/proto/PulsarApi.java

    private void initFields() {
      ledgerId_ = 0L;
      entryId_ = 0L;
      partition_ = -1;
      batchIndex_ = -1;
+      ackSet_ = java.util.Collections.emptyList();;


Double ;; You might want to do a global replacement since these are in multiple places.

The PulsarApi.java is generated by a proto tool. So I won’t do it manually here, it's better to improve the generation tool.

zzzming · 2020-02-13T16:12:34Z

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

@@ -219,7 +226,8 @@ public ChannelPromise sendMessages(final List<Entry> entries, EntryBatchSizes ba
        }

        // reduce permit and increment unackedMsg count with total number of messages in batch-msgs
-        MESSAGE_PERMITS_UPDATER.addAndGet(this, -totalMessages);
+        int ackedCount = batchIndexesAcks == null ? 0 : batchIndexesAcks.getTotalAckedIndexCount();
+        MESSAGE_PERMITS_UPDATER.addAndGet(this, ackedCount - totalMessages);


Would this case be covered if the total number of messages in a batch is greater the allowed number of permits? Or is this not even possible?

The batch size can greater than available permits and already handled at the client size.

zzzming · 2020-02-13T16:19:35Z

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

@@ -171,6 +171,9 @@
            + " affecting the accuracy of the delivery time compared to the scheduled time. Default is 1 second.")
    private long delayedDeliveryTickTimeMillis = 1000;

+    @FieldContext(category = CATEGORY_SERVER, doc = "Whether to enable the acknowledge of batch local index")
+    private boolean batchIndexAcknowledgeEnable = true;


If this is a feature has to enabled on the broker, is there a way for consumer to know or set such feature is supported? I think the problem comes how to help troubleshooting when both broker and consumer need to support this.

Should we consider adding a flag to enable or disable at consumer side? I think this feature does not introduce too much overhead on the consumer side.

For troubleshooting, users just to ensure that their brokers are enabled this feature and the client version is 2.6.0

sijie

@codelipenghui

Can you also create a master issue for this PIP for tracking the support for other language clients?

sijie · 2020-02-13T22:03:17Z

conf/broker.conf

@@ -312,6 +312,9 @@ delayedDeliveryEnabled=true
 # Default is 1 second.
 delayedDeliveryTickTimeMillis=1000

+# Whether to enable acknowledge of batch local index.
+batchIndexAcknowledgeEnable = true


Suggested change

batchIndexAcknowledgeEnable = true

batchIndexAcknowledgementEnabled = true

a better name for this feature would be "acknowledgment at batch index level". So we can name this setting "acknowledgmentAtBatchIndexLevelEnabled".

sijie · 2020-02-13T22:07:57Z

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java

+    /**
+     * Get deleted batch indexes list for a batch message.
+     */
+    long[] getDeletedBatchIndexesLongArray(PositionImpl position);


Suggested change

long[] getDeletedBatchIndexesLongArray(PositionImpl position);

long[] getDeletedBatchIndexesAsLongArray(PositionImpl position);

sijie · 2020-02-13T22:08:22Z

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java

@@ -39,6 +39,8 @@

    private boolean createIfMissing = true;
    private int maxUnackedRangesToPersist = 10000;
+    private int maxBatchDeletedIndexToPersist = 10000;
+    private boolean batchIndexDeleteEnabled = true;


Suggested change

private boolean batchIndexDeleteEnabled = true;

private boolean deletionAtBatchIndexLevelEnabled = true;

sijie · 2020-02-13T22:10:14Z

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java

@@ -148,6 +151,10 @@
        return position;
    };
    private final LongPairRangeSet<PositionImpl> individualDeletedMessages;
+
+    // Maintain the indexes deleted status of batch messages that not deleted completely


Suggested change

// Maintain the indexes deleted status of batch messages that not deleted completely

// Maintain the deletion status for batch messages

sijie · 2020-02-13T22:10:32Z

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java

@@ -148,6 +151,10 @@
        return position;
    };
    private final LongPairRangeSet<PositionImpl> individualDeletedMessages;
+
+    // Maintain the indexes deleted status of batch messages that not deleted completely
+    // (ledgerId, entryId) -> deleted indexes


Suggested change

// (ledgerId, entryId) -> deleted indexes

// (ledgerId, entryId) -> deletion indexes

sijie · 2020-02-13T22:27:50Z

...nt/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java

+        if (acknowledgementGroupTimeMicros == 0 || !properties.isEmpty()) {
+            doImmediateBatchIndexAck(msgId, batchIndex, batchSize, ackType, properties);
+        } else if (ackType == AckType.Cumulative) {
+            ConcurrentBitSet bitSet = new ConcurrentBitSet(batchSize);


Why do we need a concurrent bit set?

sijie · 2020-02-13T22:34:27Z

...nt/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java


    /**
     * This is a set of all the individual acks that the application has issued and that were not already sent to
     * broker.
     */
    private final ConcurrentSkipListSet<MessageIdImpl> pendingIndividualAcks;
+    private final ConcurrentHashMap<MessageIdImpl, ConcurrentBitSet> pendingIndividualBatchIndexAcks;


bookkeeper has a pretty efficient AckSet / WriteSet implementation that can avoid producing a lot of garbages. I think we should leverage the implementation there - https://github.com/apache/bookkeeper/blob/master/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/RoundRobinDistributionSchedule.java#L76

sijie · 2020-02-13T22:37:11Z

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java

+
+    // Maintain the indexes deleted status of batch messages that not deleted completely
+    // (ledgerId, entryId) -> deleted indexes
+    private final ConcurrentSkipListMap<PositionImpl, BitSet> batchDeletedIndexes;


Same comment as I made to client side change. We should consider re-using some of the AckSet implementation in bookkeeper client.

sijie · 2020-02-13T22:40:51Z

managed-ledger/src/main/proto/MLDataFormats.proto

@@ -78,6 +81,11 @@ message MessageRange {
    required NestedPositionInfo upperEndpoint = 2;
 }

+message BatchDeletedIndexInfo {
+    required NestedPositionInfo position = 1;
+    repeated int64 ackBitSet = 2;


A few comments here:

I would prefer calling it delete_set. "Acknowledgement" is the concept at the client & broker side. In managed ledger level, we use "delete". So it is better to keep the name consistent.

sijie · 2020-02-13T22:43:15Z

managed-ledger/src/main/proto/MLDataFormats.proto

@@ -78,6 +81,11 @@ message MessageRange {
    required NestedPositionInfo upperEndpoint = 2;
 }

+message BatchDeletedIndexInfo {


The name is not very clear. I think it is keeping the deletion index for a batched entry. So I would suggest naming it "BatchedEntryDeletionIndexInfo"

sijie · 2020-02-13T22:48:51Z

@codelipenghui I think the change looks very good in general. However, there are a few major comments I would like to clarify here.

we should be consistent using the name across the client, broker and storage layers. I think we should call the bitmap ack_set at client and broker layers. but we should call it delete_set at the storage layer (ML part).
bookkeeper has a good implementation of WriteSet and AckSet that leverage netty recycle objects. we should leverage that implementation.

sijie · 2020-02-13T22:53:48Z

Also, I feel we should clarify that the feature is implementing "acknowledgment at batch index level". It is confusing when we are saying "batch acknowledgment". So we'd better rename the PIP to "Support acknowledgment at batch index level"

codelipenghui · 2020-02-18T11:58:11Z

@sijie I will change the default value of batchIndexAcknowledgementEnabled to false since this feature brings some resource overhead.

codelipenghui · 2020-02-18T13:48:34Z

@zzzming Thanks for your comments. I have added an issue #6353 to track implementation for other language clients. I'm not familiar with C++.

I would suggest to add more documentation to explain the expected haviour and supported or non-supported consumer subscriptions;

This feature has no restrictions on subscription types.

also the performance considerations like whether re-transmission of the entire batch of partially acked messages

This is not introduced by this feature and we really need to do some tradeoffs. If users care about re-transmission of the entire batch, they can disable the batch message. But in my opinion, to disabling batch message could be worse.

If the broker splits and sends batch messages, this will introduce more overhead on the broker.

codelipenghui · 2020-02-20T09:02:12Z

/pulsarbot run-failure-checks

codelipenghui · 2020-02-20T10:26:28Z

/pulsarbot run-failure-checks

codelipenghui · 2020-02-22T03:39:27Z

/pulsarbot run-failure-checks

codelipenghui · 2020-02-22T06:35:33Z

/pulsarbot run-failure-checks

codelipenghui · 2020-02-26T02:58:26Z

/pulsarbot run-failure-checks

codelipenghui · 2020-05-25T02:02:34Z

/pulsarbot run-failure-checks

codelipenghui · 2020-05-26T13:50:13Z

/pulsarbot run-failure-checks

codelipenghui · 2020-05-28T02:37:20Z

/pulsarbot run-failure-checks

codelipenghui · 2020-05-28T03:30:04Z

/pulsarbot run-failure-checks

codelipenghui · 2020-05-28T05:08:28Z

/pulsarbot run-failure-checks

codelipenghui · 2020-05-28T15:30:16Z

/pulsarbot run-failure-checks

codelipenghui · 2020-05-29T01:56:54Z

/pulsarbot run-failure-checks

codelipenghui · 2020-05-29T08:28:55Z

/pulsarbot run-failure-checks

codelipenghui · 2020-05-29T15:14:50Z

/pulsarbot run-failure-checks

codelipenghui · 2020-05-30T14:48:28Z

/pulsarbot run-failure-checks

Anonymitaet · 2020-06-08T03:36:07Z

Doc has been added.
#7018

Master issue: apache#6253 Fixes apache#5969 ### Motivation Add support for ack batch message local index. Can be disabled at broker side by set batchIndexAcknowledgeEnable=false at broker.conf PIP-54 documentation will be created soon. ### Modifications 1. Managed cursor support track and persistent local index of batch message. 2. Client support send batch index ack to broker. 3. The batch messages with index ack information dispatched to the client. 4. Client skip the acked index. ### Verifying this change New unit tests added

startjava · 2022-09-22T11:58:57Z

@codelipenghui penghui please see the url:
https://forum.streamnative.io/t/f-noack/487
thank you
You are familiar with this

codelipenghui requested review from sijie, merlimat, rdhabalia and jiazhai January 14, 2020 13:35

codelipenghui self-assigned this Jan 14, 2020

sijie requested changes Jan 17, 2020

View reviewed changes

sijie added this to the 2.6.0 milestone Jan 17, 2020

sijie added area/broker type/feature The PR added a new feature or issue requested a new feature labels Jan 17, 2020

codelipenghui force-pushed the batch_index_ack branch from c92f2dc to 94fba9e Compare January 22, 2020 03:21

codelipenghui force-pushed the batch_index_ack branch from 51f599e to 201bc3e Compare February 13, 2020 12:12

zzzming reviewed Feb 13, 2020

View reviewed changes

sijie requested changes Feb 13, 2020

View reviewed changes

sijie added the doc-required Your PR changes impact docs and you will update later. label Feb 13, 2020

codelipenghui changed the title ~~[PIP-54] Acknowledgement for batch message local index~~ [PIP-54] Support acknowledgment at batch index level Feb 18, 2020

codelipenghui mentioned this pull request Feb 18, 2020

Clients support acknowledgment at batch index level. #6353

Closed

sijie mentioned this pull request Feb 18, 2020

ISSUE-6353: Clients support acknowledgment at batch index level. streamnative/pulsar-archived#719

Open

codelipenghui force-pushed the batch_index_ack branch from 335cdff to 00f0865 Compare February 19, 2020 12:15

codelipenghui added 2 commits April 24, 2020 15:15

Add support for batch local index ack

f69fc22

Fix comments

0bd8eab

codelipenghui added 2 commits May 27, 2020 17:31

Fix tests

f248dc8

Merge remote-tracking branch 'apache/master' into batch_index_ack

e3f9e72

Fix tests.

15f3979

Merge remote-tracking branch 'apache/master' into batch_index_ack

4d65aa6

Merge branch 'master' into batch_index_ack

9789f39

sijie approved these changes Jun 2, 2020

View reviewed changes

sijie merged commit aa0fbb0 into apache:master Jun 2, 2020

codelipenghui deleted the batch_index_ack branch June 2, 2020 01:01

codelipenghui mentioned this pull request Jun 2, 2020

Acknowledgement for batch message local index #6253

Closed

Anonymitaet removed the doc-required Your PR changes impact docs and you will update later. label Jun 8, 2020

omnilight mentioned this pull request Mar 10, 2021

Support for PIP-54 apache/pulsar-client-go#486

Open

sijie mentioned this pull request Mar 10, 2021

ISSUE-486: Support for PIP-54 streamnative/pulsar-client-go#163

Open

shibd mentioned this pull request Oct 28, 2022

[feature request] C++ and Python support batch index acknowledgment apache/pulsar-client-cpp#87

Closed

2 tasks

sijie mentioned this pull request Aug 19, 2022

ISSUE-17184: [feature request] C++ and Python support batch index acknowledgment streamnative/pulsar-archived#4764

Open

2 tasks

tisonkun mentioned this pull request Dec 10, 2022

[Issue 5969] prevent redelivery of acked batch message at the client api #5990

Closed

shibd mentioned this pull request Jan 3, 2023

[feature request] support batch index acknowledgment apache/pulsar-client-go#934

Closed

1 task

	batchIndexAcknowledgeEnable = true
	batchIndexAcknowledgementEnabled = true

	long[] getDeletedBatchIndexesLongArray(PositionImpl position);
	long[] getDeletedBatchIndexesAsLongArray(PositionImpl position);

	private boolean batchIndexDeleteEnabled = true;
	private boolean deletionAtBatchIndexLevelEnabled = true;

	// Maintain the indexes deleted status of batch messages that not deleted completely
	// Maintain the deletion status for batch messages

	// (ledgerId, entryId) -> deleted indexes
	// (ledgerId, entryId) -> deletion indexes

[PIP-54] Support acknowledgment at batch index level #6052

[PIP-54] Support acknowledgment at batch index level #6052

Conversation

codelipenghui commented Jan 14, 2020 • edited Loading

Motivation

Modifications

Verifying this change

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

Documentation

codelipenghui commented Jan 17, 2020

sijie left a comment

Choose a reason for hiding this comment

zzzming 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

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sijie Feb 13, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sijie commented Feb 13, 2020

sijie commented Feb 13, 2020

codelipenghui commented Feb 18, 2020

codelipenghui commented Feb 18, 2020

codelipenghui commented Feb 20, 2020

codelipenghui commented Feb 20, 2020

codelipenghui commented Feb 22, 2020

codelipenghui commented Feb 22, 2020

codelipenghui commented Feb 26, 2020

codelipenghui commented May 25, 2020

codelipenghui commented May 26, 2020

codelipenghui commented May 28, 2020

codelipenghui commented May 28, 2020

codelipenghui commented May 28, 2020

codelipenghui commented May 28, 2020

codelipenghui commented May 29, 2020

codelipenghui commented May 29, 2020

codelipenghui commented May 29, 2020

codelipenghui commented May 30, 2020

Anonymitaet commented Jun 8, 2020

startjava commented Sep 22, 2022

codelipenghui commented Jan 14, 2020 •

edited

Loading

sijie Feb 13, 2020 •

edited

Loading