Ack response implementation #8996

congbobo184 · 2020-12-18T06:00:04Z

Motivation

in order to handle ack response implementation. When this PR commit, I will handle #8997.

implement

we implement a new PersistentAcknowledgmentsWithResponseGroupingTracker.
we will add two ackRequests struct for async and sync flush.
add a timer to handle timeout and the timeout task don't need to lock, because the timeout is sequential.

Verifying this change

Add the tests for it

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)

congbobo184 · 2020-12-18T07:23:30Z

/pulsarbot run-failure-checks

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

congbobo184 · 2020-12-23T01:01:54Z

/pulsarbot run-failure-checks

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

eolivelli · 2020-12-27T19:03:18Z

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

-                    pendingIndividualTransactionAcks
-                            .add(Triple.of(txn.getTxnIdMostBits(), txn.getTxnIdLeastBits(), msgId));
+                consumer.onAcknowledgeCumulative(msgId, null);
+                if (((BatchMessageIdImpl) msgId).ackCumulative()) {


What happens if we don't have this class? Should we add an instanceof test?

line 183 is checked, I think @congbobo184 you can use batchMessageId since you already convert to BatchMessageIdImpl in 184

eolivelli · 2020-12-27T19:04:29Z

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

+                    // when flush the ack, we should bind the this ack in the currentFuture, during this time we can't
+                    // change currentFuture. but we can lock by the read lock, because the currentFuture is not change
+                    // any ack operation is allowed.
+                    this.lock.readLock().lock();


Move this line out of the finally block

congbobo184 · 2020-12-30T01:18:42Z

/pulsarbot run-failure-checks

congbobo184 · 2020-12-30T09:15:12Z

/pulsarbot run-failure-checks

congbobo184 · 2020-12-31T01:15:04Z

/pulsarbot run-failure-checks

congbobo184 · 2021-01-05T00:55:36Z

/pulsarbot run-failure-checks

sijie · 2021-01-06T10:58:40Z

@codelipenghui Can you review it?

# Conflicts: # pulsar-client/src/main/java/org/apache/pulsar/client/impl/AcknowledgmentsGroupingTracker.java # pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java # pulsar-client/src/test/java/org/apache/pulsar/client/impl/AcknowledgementsGroupingTrackerTest.java # pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java

# Conflicts: # pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingTest.java

congbobo184 · 2021-01-11T05:43:50Z

/pulsarbot run-failure-checks

congbobo184 · 2021-01-12T03:59:49Z

/pulsarbot run-failure-checks

codelipenghui

Need to consider the compatibility that the new client connects the old version broker which does not support enable ack response. If a broker does not write the ack response to the client, the client might be stuck on the message acknowledgment?

And this should be covered by the compatibility test in the tests modules.

codelipenghui · 2021-01-14T09:45:49Z

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

+    /**
+     * Consumer ack for response timeout.
+     */
+    public static class AckResponseTimeoutException extends PulsarClientException {


Is the PulsarClientException.TimeoutException works?

No, it can be deleted.

codelipenghui · 2021-01-14T09:49:39Z

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

+     * @param ackResponseEnabled {@link Boolean} is enable ack for response
+     * @return the consumer builder instance
+     */
+    ConsumerBuilder<T> enableAckResponse(boolean ackResponseEnabled);


Suggested change

ConsumerBuilder<T> enableAckResponse(boolean ackResponseEnabled);

ConsumerBuilder<T> isAckReceiptEnabled(boolean ackReceiptEnabled);

I think response here a little bit ambiguous since the client also can get the returned future without this feature.

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

codelipenghui · 2021-01-14T10:03:27Z

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

-                    pendingIndividualTransactionAcks
-                            .add(Triple.of(txn.getTxnIdMostBits(), txn.getTxnIdLeastBits(), msgId));
+                consumer.onAcknowledgeCumulative(msgId, null);
+                if (((BatchMessageIdImpl) msgId).ackCumulative()) {


line 183 is checked, I think @congbobo184 you can use batchMessageId since you already convert to BatchMessageIdImpl in 184

codelipenghui · 2021-01-14T10:58:53Z

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

+    private MessageIdImpl modifyBatchMessageIdAndStatusInConsumer(BatchMessageIdImpl batchMessageId) {
+        MessageIdImpl messageId = new MessageIdImpl(batchMessageId.getLedgerId(),
+                batchMessageId.getEntryId(), batchMessageId.getPartitionIndex());
+        consumer.getStats().incrementNumAcksSent(batchMessageId.getBatchSize());
+        modifyMessageIdStatusInConsumerCommon(messageId);
+        return messageId;
+    }
+
+    private void modifyMessageIdStatusInConsumer(MessageIdImpl messageId) {
+        consumer.getStats().incrementNumAcksSent(1);
+        modifyMessageIdStatusInConsumerCommon(messageId);
+    }
+
+    private void modifyMessageIdStatusInConsumerCommon(MessageIdImpl messageId) {
+        consumer.getUnAckedMessageTracker().remove(messageId);
+        if (consumer.getPossibleSendToDeadLetterTopicMessages() != null) {
+            consumer.getPossibleSendToDeadLetterTopicMessages().remove(messageId);
+        }
+    }


These methods not clear here. Stats, not status. And should split the update stats method and cleanup consumer method stay independent, this will improve the code readability

codelipenghui · 2021-01-14T11:01:57Z

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

+                ByteBuf cmd = Commands.newAck(consumer.consumerId, lastCumulativeAck.messageId.ledgerId,
+                        lastCumulativeAck.messageId.getEntryId(), lastCumulativeAck.bitSetRecyclable,
+                        AckType.Cumulative, null, Collections.emptyMap(), requestId);
+                cnx.newAckForResponseWithFuture(cmd, requestId, currentCumulativeAckFuture);
+                this.currentCumulativeAckFuture = new TimedCompletableFuture<>();
+            } else {
+                ByteBuf cmd = Commands.newAck(consumer.consumerId, lastCumulativeAck.messageId.ledgerId,
+                        lastCumulativeAck.messageId.getEntryId(), lastCumulativeAck.bitSetRecyclable,
+                        AckType.Cumulative, null, Collections.emptyMap(), -1);
+                cnx.ctx().write(cmd, cnx.ctx().voidPromise());


Please optimize duplicate code

if we optimize the code, we will judge the ackResponseEnabled twice, because get requestId then new cmd and then write. get requestId and write are different from normal ack and ack response. the middle operation newAckComand can't be optimized.

codelipenghui · 2021-01-14T11:21:23Z

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

+                    long requestId = consumer.getClient().newRequestId();
+                    ByteBuf cmd = Commands.newMultiMessageAck(consumer.consumerId, entriesToAck, requestId);
+                    completableFuture = cnx.newAckForResponse(cmd, requestId);


Should write the command to the broker? And please consider reducing the duplicate code

because newImmediateAckAndFlush so we should write the command to broker immediately.

codelipenghui · 2021-01-14T11:23:46Z

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

+                // if don't support multi message ack, it also support ack response, so we should not think about the
+                // ack response in this logic


Why only support for muti message ack. Single message acknowledge also can enable ack receipt?

this is the chunk message, so if want to use ack response, broker must support multi message ack, we don't need to support broker don't support multi message ack and then we return ack response.

codelipenghui · 2021-01-14T11:26:06Z

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

+            if (ackResponseEnabled) {
+                long requestId = consumer.getClient().newRequestId();
+                ByteBuf cmd = Commands.newAck(consumerId, msgId.getLedgerId(), msgId.getEntryId(), bitSet,
+                        ackType, null, map, requestId);
+                completableFuture = cnx.newAckForResponse(cmd, requestId);
            } else {
-                cnx.ctx().write(cmd, cnx.ctx().voidPromise());
+                ByteBuf cmd = Commands.newAck(consumerId, msgId.getLedgerId(), msgId.getEntryId(), bitSet,
+                        ackType, null, map, -1);
+                cnx.ctx().writeAndFlush(cmd, cnx.ctx().voidPromise());
+                completableFuture = CompletableFuture.completedFuture(null);
+            }
+        }
+        return completableFuture;


I think we can use a method getRequestId() and a method getCompletableFuture(). I think this will make the logic simpler

codelipenghui · 2021-01-14T11:32:33Z

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

+    private <T> CompletableFuture<T> sendRequestAndHandleTimeout(ByteBuf requestMessage, long requestId,
+                                                                 RequestType requestType, boolean flush,
+                                                                 TimedCompletableFuture<T> future) {


Do we need to return the future again?

I think you can use 2 methods to handle this case

private <T> void sendRequestAndHandleTimeout(ByteBuf requestMessage, long requestId, RequestType requestType, boolean flush, TimedCompletableFuture<T> future) { }

private <T> CompletableFuture<T> sendRequestAndHandleTimeout(ByteBuf requestMessage, long requestId, RequestType requestType, boolean flush) { TimedCompletableFuture<T> future = new TimedCompletableFuture<>(); sendRequestAndHandleTimeout(... future); return future; }

lhotari · 2021-07-07T09:41:34Z

@congbobo184 @codelipenghui @Anonymitaet @merlimat Has the "ackReceipt" feature added by this PR been documented? The only documentation that I can find is in the javadoc of the isAckReceiptEnabled method "Ack will return receipt but does not mean that the message will not be resent after get receipt." .
Would it be possible to clarify this in the Pulsar documentation?
Usually PIP documents are created for changes which change the wire protocol. I didn't find and PIP documents or mailing list discussions about this change. What problem is this change solving? The issue and PR description aren't very informative. Please elaborate more about the context behind this change.

Would it make sense to document this change in the "Pulsar binary protocol specification" document, https://pulsar.apache.org/docs/en/develop-binary-protocol/ https://github.com/apache/pulsar/blob/master/site2/docs/developing-binary-protocol.md ?

congbobo184 · 2021-07-08T01:18:38Z

@lhotari hi, this pr only means server receive client ack request and then send response to client. "Ack will return receipt but does not mean that the message will not be resent after get receipt." . can add to the Pulsar documentation.

lhotari · 2021-07-08T09:01:53Z

Hi @Anonymitaet , could you help @congbobo184 in documenting this feature?

Since this PR also changed the wire protocol (there were protobuf changes), I think that this requires creating a PIP document. This is a common rule in Pulsar development that there must be a PIP for wire protocol changes. @codelipenghui Could you help with a PIP document for the "ack receipt" feature? /cc @merlimat @sijie

codelipenghui · 2021-07-08T10:07:37Z

@lhotari The ack response proposal contained in the transaction proposal, you can find here https://docs.google.com/document/d/145VYp09JKTw9jAT-7yNyFU255FptB2_B2Fye100ZXDI/edit#bookmark=id.r370pcotoarj

For the documentation, you can just create an issue to track it.

Anonymitaet · 2021-07-09T03:42:44Z

I will work with @congbobo184 on the docs and track the issue here: #11272

lhotari · 2021-07-20T08:53:03Z

@congbobo184 I recently came across #7683 . Does this PR somehow help resolve that issue?

Anonymitaet · 2022-05-09T05:22:48Z

Doc status update: I've discussed w/ @liangyepianzhou, he will add docs and @momo-jun will help review.

@congbobo184 @codelipenghui @Anonymitaet @merlimat Has the "ackReceipt" feature added by this PR been documented? The only documentation that I can find is in the javadoc of the isAckReceiptEnabled method "Ack will return receipt but does not mean that the message will not be resent after get receipt." . Would it be possible to clarify this in the Pulsar documentation? Usually PIP documents are created for changes which change the wire protocol. I didn't find and PIP documents or mailing list discussions about this change. What problem is this change solving? The issue and PR description aren't very informative. Please elaborate more about the context behind this change.

Would it make sense to document this change in the "Pulsar binary protocol specification" document, https://pulsar.apache.org/docs/en/develop-binary-protocol/ https://github.com/apache/pulsar/blob/master/site2/docs/developing-binary-protocol.md ?

Anonymitaet · 2022-05-10T07:42:04Z

Doc is added #15497

### Motivation There were several issues caused by the thread safe issue of `LastCumulativeAck`, see: - apache#10586 - apache#12343 The root cause is that `LastCumulativeAck` could be accessed by different threads, especially in `flushAsync` method. But the fields are accessed directly and no thread safety can be guaranteed. In addition, the current `LastCumulativeAck` class was added in apache#8996 to hold two object references, but this modification is wrong. Before apache#8996, there are two CAS operations in `doCumulativeAck` method in case it's called concurretly. Though the composite CAS operation is not atomic. However, after apache#8996, only CAS operation was performed but it's compared with a `LastCumulativeAck` object, not the two fields (`messageId` and `bitSetRecyclable`). ### Modifications To solve the thread safety issue, this PR move the `LastCumulativeAck` out of the `PersistentAcknowledgmentsGroupingTracker` to disable directly access to the internal fields. Then, two synchronized methods were added to guarantee the thread safety: - `update`: Guarantee the safe write operations. It also recycles the `BitSetRecyclable` object before assigning new values. - `moveOwnershipTo`: This method moves the ownership to another `LastCumulativeAck` object, which will be responsible to recycle the `BitSetRecyclable` field after that. With the methods above, each time `flushAsync` is called, move the ownership of `lastCumulativeAck` field to another thread local field to send the ACK command and recycle the `BitSetRecyclable` field. - `lastCumulativeAck` updates the latest message ID and bit set, the update operations can be performed by multiple threads and `lastCumulativeAck` saves the latest value. - `threadLocalLastCumulativeAckToFlush` only acts as a temporary cache to the latest value in `flushAsync`.

### Motivation There were several issues caused by the thread safe issue of `LastCumulativeAck`, see: - apache#10586 - apache#12343 The root cause is that `LastCumulativeAck` could be accessed by different threads, especially in `flushAsync` method. But the fields are accessed directly and no thread safety can be guaranteed. In addition, the current `LastCumulativeAck` class was added in apache#8996 to hold two object references, but this modification is wrong. Before apache#8996, there are two CAS operations in `doCumulativeAck` method in case it's called concurretly. Though the composite CAS operation is not atomic. However, after apache#8996, only CAS operation was performed but it's compared with a `LastCumulativeAck` object, not the two fields (`messageId` and `bitSetRecyclable`). ### Modifications To solve the thread safety issue, this PR move the `LastCumulativeAck` out of the `PersistentAcknowledgmentsGroupingTracker` to disable directly access to the internal fields. Then, the following synchronized methods were added to guarantee the thread safety: - `update`: Guarantee the safe write operations. It also recycles the `BitSetRecyclable` object before assigning new values. - `moveOwnershipTo`: This method moves the ownership to another `LastCumulativeAck` object. After that, the `update` operation on this object won't recycle the `BitSetRecyclable` field. - `restoreOwnershipIfEmpty`: Restore the ownership from another `LastCumulativeAck` object. With the methods above, each time `flushAsync` is called, move the ownership of `lastCumulativeAck` field to another thread local field to send the ACK command. After that, restore the ownership to `lastCumulativeAck` unless it has been updated in other threads.

### Motivation There were several issues caused by the thread safe issue of `LastCumulativeAck`, see: - apache#10586 - apache#12343 The root cause is that `LastCumulativeAck` could be accessed by different threads, especially in `flushAsync` method. But the fields are accessed directly and no thread safety can be guaranteed. In addition, the current `LastCumulativeAck` class was added in apache#8996 to hold two object references, but this modification is wrong. Before apache#8996, there are two CAS operations in `doCumulativeAck` method in case it's called concurretly. Though the composite CAS operation is not atomic. However, after apache#8996, only CAS operation was performed but it's compared with a `LastCumulativeAck` object, not the two fields (`messageId` and `bitSetRecyclable`). There is another issue that it uses a flag `cumulativeAckFlushRequired` to mark if `lastCumulativeAck` should flush. However, if `flushAsync` was called concurrently, both would send ACK commands to broker. ### Modifications To solve the thread safety issue, this PR move the `LastCumulativeAck` out of the `PersistentAcknowledgmentsGroupingTracker` to disable directly access to the internal fields. Then, the following synchronized methods were added to guarantee the thread safety: - `update`: Guarantee the safe write operations. It also recycles the `BitSetRecyclable` object before assigning new values and indicates itself can be flushed. - `flush`: If it can be flushed, return a thread local `LastCumulativeAck` instance that contains the message ID and the bit set. Then mark it as no need to flush. In addition, since the `messageId` field is volatile, the `getMessageId` method can always retrieve the latest reference. Based on the new design, we can only maintain a `LastCumulativeAck` field in `PersistentAcknowledgmentsGroupingTracker` and call the related methods in `doCumulativeAck` and `flushAsync`. It also fixes the problem that two concurrent `flushAsync` calls might send the same ACK command twice.

### Motivation There were several issues caused by the thread safe issue of `LastCumulativeAck`, see: - apache#10586 - apache#12343 The root cause is that `LastCumulativeAck` could be accessed by different threads, especially in `flushAsync` method. But the fields are accessed directly and no thread safety can be guaranteed. In addition, the current `LastCumulativeAck` class was added in apache#8996 to hold two object references, but this modification is wrong. Before apache#8996, there are two CAS operations in `doCumulativeAck` method in case it's called concurretly. Though the composite CAS operation is not atomic. However, after apache#8996, only CAS operation was performed but it's compared with a `LastCumulativeAck` object, not the two fields (`messageId` and `bitSetRecyclable`). There is another issue that it uses a flag `cumulativeAckFlushRequired` to mark if `lastCumulativeAck` should flush. However, if `flushAsync` was called concurrently, both would send ACK commands to broker. ### Modifications To solve the thread safety issue, this PR move the `LastCumulativeAck` out of the `PersistentAcknowledgmentsGroupingTracker` to disable directly access to the internal fields. Then, the following synchronized methods were added to guarantee the thread safety: - `update`: Guarantee the safe write operations. It also recycles the `BitSetRecyclable` object before assigning new values and indicates itself can be flushed. - `flush`: If it can be flushed, return a thread local `LastCumulativeAck` instance that contains the message ID and the bit set. The bit set is deep copied to avoid the original reference being recycled in another `update` call. In addition, since the `messageId` field is volatile, the `getMessageId` method can always retrieve the latest reference. `LastCumulativeAckTest` is added to verify the sematics above. Based on the new design, we can only maintain a `LastCumulativeAck` field in `PersistentAcknowledgmentsGroupingTracker` and call the related methods in `doCumulativeAck` and `flushAsync`. It also fixes the problem that two concurrent `flushAsync` calls might send the same ACK command twice. Remove unused field Don't reset in LastCumulativeAck#flush

…6072) ### Motivation There were several issues caused by the thread safe issue of `LastCumulativeAck`, see: - #10586 - #12343 The root cause is that `LastCumulativeAck` could be accessed by different threads, especially in `flushAsync` method. But the fields are accessed directly and no thread safety can be guaranteed. In addition, the current `LastCumulativeAck` class was added in #8996 to hold two object references, but this modification is wrong. Before #8996, there are two CAS operations in `doCumulativeAck` method in case it's called concurretly. Though the composite CAS operation is not atomic. However, after #8996, only CAS operation was performed but it's compared with a `LastCumulativeAck` object, not the two fields (`messageId` and `bitSetRecyclable`). There is another issue that it uses a flag `cumulativeAckFlushRequired` to mark if `lastCumulativeAck` should flush. However, if `flushAsync` was called concurrently, both would send ACK commands to broker. ### Modifications To solve the thread safety issue, this PR move the `LastCumulativeAck` out of the `PersistentAcknowledgmentsGroupingTracker` to disable directly access to the internal fields. Then, the following synchronized methods were added to guarantee the thread safety: - `update`: Guarantee the safe write operations. It also recycles the `BitSetRecyclable` object before assigning new values and indicates itself can be flushed. - `flush`: If it can be flushed, return a thread local `LastCumulativeAck` instance that contains the message ID and the bit set. The bit set is deep copied to avoid the original reference being recycled in another `update` call. In addition, since the `messageId` field is volatile, the `getMessageId` method can always retrieve the latest reference. `LastCumulativeAckTest` is added to verify the sematics above. Based on the new design, we can only maintain a `LastCumulativeAck` field in `PersistentAcknowledgmentsGroupingTracker` and call the related methods in `doCumulativeAck` and `flushAsync`. It also fixes the problem that two concurrent `flushAsync` calls might send the same ACK command twice.

…6072) ### Motivation There were several issues caused by the thread safe issue of `LastCumulativeAck`, see: - #10586 - #12343 The root cause is that `LastCumulativeAck` could be accessed by different threads, especially in `flushAsync` method. But the fields are accessed directly and no thread safety can be guaranteed. In addition, the current `LastCumulativeAck` class was added in #8996 to hold two object references, but this modification is wrong. Before #8996, there are two CAS operations in `doCumulativeAck` method in case it's called concurretly. Though the composite CAS operation is not atomic. However, after #8996, only CAS operation was performed but it's compared with a `LastCumulativeAck` object, not the two fields (`messageId` and `bitSetRecyclable`). There is another issue that it uses a flag `cumulativeAckFlushRequired` to mark if `lastCumulativeAck` should flush. However, if `flushAsync` was called concurrently, both would send ACK commands to broker. ### Modifications To solve the thread safety issue, this PR move the `LastCumulativeAck` out of the `PersistentAcknowledgmentsGroupingTracker` to disable directly access to the internal fields. Then, the following synchronized methods were added to guarantee the thread safety: - `update`: Guarantee the safe write operations. It also recycles the `BitSetRecyclable` object before assigning new values and indicates itself can be flushed. - `flush`: If it can be flushed, return a thread local `LastCumulativeAck` instance that contains the message ID and the bit set. The bit set is deep copied to avoid the original reference being recycled in another `update` call. In addition, since the `messageId` field is volatile, the `getMessageId` method can always retrieve the latest reference. `LastCumulativeAckTest` is added to verify the sematics above. Based on the new design, we can only maintain a `LastCumulativeAck` field in `PersistentAcknowledgmentsGroupingTracker` and call the related methods in `doCumulativeAck` and `flushAsync`. It also fixes the problem that two concurrent `flushAsync` calls might send the same ACK command twice. (cherry picked from commit 936d6fd)

…ache#16072) ### Motivation There were several issues caused by the thread safe issue of `LastCumulativeAck`, see: - apache#10586 - apache#12343 The root cause is that `LastCumulativeAck` could be accessed by different threads, especially in `flushAsync` method. But the fields are accessed directly and no thread safety can be guaranteed. In addition, the current `LastCumulativeAck` class was added in apache#8996 to hold two object references, but this modification is wrong. Before apache#8996, there are two CAS operations in `doCumulativeAck` method in case it's called concurretly. Though the composite CAS operation is not atomic. However, after apache#8996, only CAS operation was performed but it's compared with a `LastCumulativeAck` object, not the two fields (`messageId` and `bitSetRecyclable`). There is another issue that it uses a flag `cumulativeAckFlushRequired` to mark if `lastCumulativeAck` should flush. However, if `flushAsync` was called concurrently, both would send ACK commands to broker. ### Modifications To solve the thread safety issue, this PR move the `LastCumulativeAck` out of the `PersistentAcknowledgmentsGroupingTracker` to disable directly access to the internal fields. Then, the following synchronized methods were added to guarantee the thread safety: - `update`: Guarantee the safe write operations. It also recycles the `BitSetRecyclable` object before assigning new values and indicates itself can be flushed. - `flush`: If it can be flushed, return a thread local `LastCumulativeAck` instance that contains the message ID and the bit set. The bit set is deep copied to avoid the original reference being recycled in another `update` call. In addition, since the `messageId` field is volatile, the `getMessageId` method can always retrieve the latest reference. `LastCumulativeAckTest` is added to verify the sematics above. Based on the new design, we can only maintain a `LastCumulativeAck` field in `PersistentAcknowledgmentsGroupingTracker` and call the related methods in `doCumulativeAck` and `flushAsync`. It also fixes the problem that two concurrent `flushAsync` calls might send the same ACK command twice. (cherry picked from commit 936d6fd) (cherry picked from commit 5eefdf1)

…6072) ### Motivation There were several issues caused by the thread safe issue of `LastCumulativeAck`, see: - #10586 - #12343 The root cause is that `LastCumulativeAck` could be accessed by different threads, especially in `flushAsync` method. But the fields are accessed directly and no thread safety can be guaranteed. In addition, the current `LastCumulativeAck` class was added in #8996 to hold two object references, but this modification is wrong. Before #8996, there are two CAS operations in `doCumulativeAck` method in case it's called concurretly. Though the composite CAS operation is not atomic. However, after #8996, only CAS operation was performed but it's compared with a `LastCumulativeAck` object, not the two fields (`messageId` and `bitSetRecyclable`). There is another issue that it uses a flag `cumulativeAckFlushRequired` to mark if `lastCumulativeAck` should flush. However, if `flushAsync` was called concurrently, both would send ACK commands to broker. ### Modifications To solve the thread safety issue, this PR move the `LastCumulativeAck` out of the `PersistentAcknowledgmentsGroupingTracker` to disable directly access to the internal fields. Then, the following synchronized methods were added to guarantee the thread safety: - `update`: Guarantee the safe write operations. It also recycles the `BitSetRecyclable` object before assigning new values and indicates itself can be flushed. - `flush`: If it can be flushed, return a thread local `LastCumulativeAck` instance that contains the message ID and the bit set. The bit set is deep copied to avoid the original reference being recycled in another `update` call. In addition, since the `messageId` field is volatile, the `getMessageId` method can always retrieve the latest reference. `LastCumulativeAckTest` is added to verify the sematics above. Based on the new design, we can only maintain a `LastCumulativeAck` field in `PersistentAcknowledgmentsGroupingTracker` and call the related methods in `doCumulativeAck` and `flushAsync`. It also fixes the problem that two concurrent `flushAsync` calls might send the same ACK command twice. (cherry picked from commit 936d6fd)

Ack response

6131478

codelipenghui assigned congbobo184 Dec 20, 2020

codelipenghui added this to the 2.8.0 milestone Dec 20, 2020

codelipenghui added the release/2.7.1 label Dec 20, 2020

congbo added 2 commits December 23, 2020 00:36

Modify the ack request by clientCnx pendingRequest

2c59eda

Merge branch 'master' into congbobo184_ack_response_implementation

c38df66

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

congbo added 2 commits December 23, 2020 09:59

Fix some logical

3c2e41b

Merge the ack response into tracker

2174426

eolivelli requested changes Dec 27, 2020

View reviewed changes

codelipenghui mentioned this pull request Dec 28, 2020

Allow ack with response expected without transaction streamnative/pulsar-archived#1880

Closed

modify the lock logical

fe25400

Merge branch 'master' into congbobo184_ack_response_implementation

bde7246

sijie requested review from codelipenghui, jiazhai and sijie January 5, 2021 00:09

congbo added 2 commits January 6, 2021 23:12

Merge branch 'master' into congbobo184_ack_response_implementation

bfe770b

# Conflicts: # pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingTest.java

congbobo184 closed this Jan 11, 2021

congbobo184 reopened this Jan 11, 2021

codelipenghui requested changes Jan 14, 2021

View reviewed changes

Fix some comments

b2ddce8

lhotari added the doc-required Your PR changes impact docs and you will update later. label Jul 8, 2021

Anonymitaet mentioned this pull request Jul 9, 2021

[Doc] Add guides for "consumer" and "txn" to "Development" chapter #11272

Closed

sijie mentioned this pull request Jul 9, 2021

ISSUE-11272: [Doc] Add guides for "consumer" and "txn" to "Development" chapter streamnative/pulsar-archived#2757

Open

lhotari mentioned this pull request Jul 20, 2021

Batch acknowledge can lead to message being never acknowledged (until consumer restart) #7683

Closed

This was referenced Oct 11, 2021

Ack batch index will lose message. #8997

Closed

ISSUE-8997: Ack batch index will lose message. streamnative/pulsar-archived#1906

Closed

BewareMyPower mentioned this pull request Oct 14, 2021

fix lastCumulativeAck.messageId npe when PersistentAcknowledgmentsGroupingTracker.flushSync #12343

Closed

momo-jun mentioned this pull request May 9, 2022

[feature][Doc] Add doc for CommandAckResponse #15497

Merged

5 tasks

Anonymitaet added doc-complete Your PR changes impact docs and the related docs have been already added. and removed doc-required Your PR changes impact docs and you will update later. labels May 10, 2022

BewareMyPower mentioned this pull request Jun 15, 2022

[fix][Java Client] Fix thread safety issue of LastCumulativeAck #16072

Merged

4 tasks

BewareMyPower mentioned this pull request Nov 3, 2022

[feature request] Support ACK response apache/pulsar-client-cpp#102

Closed

	ConsumerBuilder<T> enableAckResponse(boolean ackResponseEnabled);
	ConsumerBuilder<T> isAckReceiptEnabled(boolean ackReceiptEnabled);

		// if don't support multi message ack, it also support ack response, so we should not think about the
		// ack response in this logic

Ack response implementation #8996

Ack response implementation #8996

Conversation

congbobo184 commented Dec 18, 2020 • edited Loading

Motivation

implement

Verifying this change

congbobo184 commented Dec 18, 2020

congbobo184 commented Dec 23, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

congbobo184 commented Dec 30, 2020

congbobo184 commented Dec 30, 2020

congbobo184 commented Dec 31, 2020

congbobo184 commented Jan 5, 2021

sijie commented Jan 6, 2021

congbobo184 commented Jan 11, 2021

congbobo184 commented Jan 12, 2021

codelipenghui 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

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

lhotari commented Jul 7, 2021 • edited Loading

congbobo184 commented Jul 8, 2021

lhotari commented Jul 8, 2021

codelipenghui commented Jul 8, 2021

Anonymitaet commented Jul 9, 2021

lhotari commented Jul 20, 2021

Anonymitaet commented May 9, 2022

Anonymitaet commented May 10, 2022 • edited Loading

congbobo184 commented Dec 18, 2020 •

edited

Loading

lhotari commented Jul 7, 2021 •

edited

Loading

Anonymitaet commented May 10, 2022 •

edited

Loading