You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2020/01/04 04:52:02 UTC
[GitHub] [pulsar] zzzming opened a new pull request #5990: [Issue 5969]
prevent redelivery of acked batch message at the client api
zzzming opened a new pull request #5990: [Issue 5969] prevent redelivery of acked batch message at the client api
URL: https://github.com/apache/pulsar/pull/5990
### Motivation
To address redelivered of acked batch message described by https://github.com/apache/pulsar/issues/5969
This is only a client side change. It covers cases of not-to-deliver-acked batch message on the client side only. Since it tracks acked messages of partial acked batch in memory, restart of client will lose the state tracking.
### Modifications
A new BatchAckedTracker is added to track all acked message in any partially acked batches. Redelivery of unacked batch will be evaluated against this tracker so that it sends only unacked messages to the client in ConsumerImpl. When all messages are acked by the client in redelivery, ConsumerImpl sends the batch ack to the broker.
This only addresses the issue in Java client. C++, and standalone Go clients updates will be tracked in separated PR.
### Verifying this change
Three new unit tests are introduced.
### Does this pull request potentially affect one of the following parts:
*If `yes` was chosen, please highlight the changes*
- Dependencies (does it add or upgrade a dependency): (no)
- The public API: (no)
- The schema: ( no)
- The default values of configurations: (no)
- The wire protocol: (no)
- The rest endpoints: (no)
- The admin cli options: (no)
- Anything that affects deployment: (no)
### Documentation
- Does this pull request introduce a new feature? (no)
- If yes, how is the feature documented? (not applicable)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [pulsar] zzzming commented on a change in pull request #5990:
[Issue 5969] prevent redelivery of acked batch message at the client api
Posted by GitBox <gi...@apache.org>.
zzzming commented on a change in pull request #5990: [Issue 5969] prevent redelivery of acked batch message at the client api
URL: https://github.com/apache/pulsar/pull/5990#discussion_r363448644
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchAckedTracker.java
##########
@@ -0,0 +1,87 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import org.apache.pulsar.client.api.MessageId;
+
+/**
+ * Tracks any partial acked batches and its acked messages
+ * This will prevent acked message redelivery to the client only at the client API level.
+ * This class does not track batch with all acked message. We trust broker won't deliver again.
+ */
+class BatchAckedTracker {
+
+ // a map of partial acked batch and its messages already acked
+ @VisibleForTesting
+ Map<String, Set<BatchMessageIdImpl>> ackedBatches = new HashMap<String, Set<BatchMessageIdImpl>>();
Review comment:
Currently it uses Set of integer to track the acked batch. So the memory footprint is small. I think it can even use bitset like the BatchAcker. I try to keep this isolated and testable from AcknowledgmentGroupingTracker. What do you think?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [pulsar] gmethvin commented on a change in pull request #5990:
[Issue 5969] prevent redelivery of acked batch message at the client api
Posted by GitBox <gi...@apache.org>.
gmethvin commented on a change in pull request #5990: [Issue 5969] prevent redelivery of acked batch message at the client api
URL: https://github.com/apache/pulsar/pull/5990#discussion_r363080545
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchAckedTracker.java
##########
@@ -0,0 +1,87 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import org.apache.pulsar.client.api.MessageId;
+
+/**
+ * Tracks any partial acked batches and its acked messages
+ * This will prevent acked message redelivery to the client only at the client API level.
+ * This class does not track batch with all acked message. We trust broker won't deliver again.
+ */
+class BatchAckedTracker {
+
+ // a map of partial acked batch and its messages already acked
+ @VisibleForTesting
+ Map<String, Set<BatchMessageIdImpl>> ackedBatches = new HashMap<String, Set<BatchMessageIdImpl>>();
+
+ public BatchAckedTracker() {
+ }
+
+ // we start to track
+ public void negativeAcked(BatchMessageIdImpl messageId) {
+
+ }
+
+ // If this message should be delivered and tracks this message if it is a batch message
+ public boolean deliver(MessageId messageId) {
+ if (messageId instanceof BatchMessageIdImpl) {
+ BatchMessageIdImpl id = (BatchMessageIdImpl) messageId;
+ String batchId = getBatchId(id);
+ if (ackedBatches.containsKey(batchId)){
+ Set<BatchMessageIdImpl> batch = ackedBatches.get(batchId);
+ if (batch.contains(id)) {
+ return false;
+ }
+ }
+ }
+ // deliver non batch message and any other cases
+ return true;
+ }
+
+ /**
+ *
+ * @param messageId batchMessageIdImpl
+ * @return boolean isAllMsgAcked for the batch
+ */
+ public boolean ack (BatchMessageIdImpl messageId) {
Review comment:
Can this be called from multiple threads? If so I think there are some potential concurrency issues with this code.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [pulsar] sijie commented on a change in pull request #5990: [Issue
5969] prevent redelivery of acked batch message at the client api
Posted by GitBox <gi...@apache.org>.
sijie commented on a change in pull request #5990: [Issue 5969] prevent redelivery of acked batch message at the client api
URL: https://github.com/apache/pulsar/pull/5990#discussion_r363154114
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchAckedTracker.java
##########
@@ -0,0 +1,87 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import org.apache.pulsar.client.api.MessageId;
+
+/**
+ * Tracks any partial acked batches and its acked messages
+ * This will prevent acked message redelivery to the client only at the client API level.
+ * This class does not track batch with all acked message. We trust broker won't deliver again.
+ */
+class BatchAckedTracker {
+
+ // a map of partial acked batch and its messages already acked
+ @VisibleForTesting
+ Map<String, Set<BatchMessageIdImpl>> ackedBatches = new HashMap<String, Set<BatchMessageIdImpl>>();
Review comment:
If there is a missing ack, it can result in the client using a lot of memory for storing acked batches. Why can't we leverage existing AcknowledgmentGroupingTracker for doing that job? It already has information about acked messages.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [pulsar] sijie commented on a change in pull request #5990: [Issue
5969] prevent redelivery of acked batch message at the client api
Posted by GitBox <gi...@apache.org>.
sijie commented on a change in pull request #5990: [Issue 5969] prevent redelivery of acked batch message at the client api
URL: https://github.com/apache/pulsar/pull/5990#discussion_r363672688
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchAckedTracker.java
##########
@@ -0,0 +1,87 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import org.apache.pulsar.client.api.MessageId;
+
+/**
+ * Tracks any partial acked batches and its acked messages
+ * This will prevent acked message redelivery to the client only at the client API level.
+ * This class does not track batch with all acked message. We trust broker won't deliver again.
+ */
+class BatchAckedTracker {
+
+ // a map of partial acked batch and its messages already acked
+ @VisibleForTesting
+ Map<String, Set<BatchMessageIdImpl>> ackedBatches = new HashMap<String, Set<BatchMessageIdImpl>>();
Review comment:
1) I think we should go with bitset. Otherwise, the memory usage can become out of control.
2) Since the logic here seems to be duplicating the functionality of AcknowledgmentGroupingTracker. I would prefer not adding duplicated classes. It is confusing and hard to maintain.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [pulsar] zzzming commented on issue #5990: [Issue 5969] prevent
redelivery of acked batch message at the client api
Posted by GitBox <gi...@apache.org>.
zzzming commented on issue #5990: [Issue 5969] prevent redelivery of acked batch message at the client api
URL: https://github.com/apache/pulsar/pull/5990#issuecomment-573772493
@codelipenghui Thanks for the update. I'll look forward to the broker side implementation. Do we have an ETA on this?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [pulsar] zzzming commented on issue #5990: [Issue 5969] prevent
redelivery of acked batch message at the client api
Posted by GitBox <gi...@apache.org>.
zzzming commented on issue #5990: [Issue 5969] prevent redelivery of acked batch message at the client api
URL: https://github.com/apache/pulsar/pull/5990#issuecomment-571010464
> Since this is on the client side I assume you will want to do it in the clients for other languages as well, right?
Yes. Java client is used to validate the logic in this PR. Will be expand to C++ client in another PR.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [pulsar] codelipenghui commented on issue #5990: [Issue 5969]
prevent redelivery of acked batch message at the client api
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on issue #5990: [Issue 5969] prevent redelivery of acked batch message at the client api
URL: https://github.com/apache/pulsar/pull/5990#issuecomment-572418780
@zzzming In shared subscription mode, this will cause more confusion for users, because messages can be redelivery to other consumers. The status of each consumer is not synchronized. If this approach only work with single active consumer subscription mode, i think we can solve it better on the server side, client filter messages based on information provided by the broker. So that, we can keep the client side simple, This is better for clients in different languages.
And if we solve this problem from the server side in the future(I prefer this solution https://github.com/apache/pulsar/issues/5969#issuecomment-570101434 ), the current PR mechanism will also be discarded, so we should consider whether it is necessary.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [pulsar] zzzming commented on a change in pull request #5990:
[Issue 5969] prevent redelivery of acked batch message at the client api
Posted by GitBox <gi...@apache.org>.
zzzming commented on a change in pull request #5990: [Issue 5969] prevent redelivery of acked batch message at the client api
URL: https://github.com/apache/pulsar/pull/5990#discussion_r364458073
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchAckedTracker.java
##########
@@ -0,0 +1,87 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import org.apache.pulsar.client.api.MessageId;
+
+/**
+ * Tracks any partial acked batches and its acked messages
+ * This will prevent acked message redelivery to the client only at the client API level.
+ * This class does not track batch with all acked message. We trust broker won't deliver again.
+ */
+class BatchAckedTracker {
+
+ // a map of partial acked batch and its messages already acked
+ @VisibleForTesting
+ Map<String, Set<BatchMessageIdImpl>> ackedBatches = new HashMap<String, Set<BatchMessageIdImpl>>();
Review comment:
@sijie Updated the code with BitSet.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [pulsar] zzzming commented on a change in pull request #5990:
[Issue 5969] prevent redelivery of acked batch message at the client api
Posted by GitBox <gi...@apache.org>.
zzzming commented on a change in pull request #5990: [Issue 5969] prevent redelivery of acked batch message at the client api
URL: https://github.com/apache/pulsar/pull/5990#discussion_r363163077
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchAckedTracker.java
##########
@@ -0,0 +1,87 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import org.apache.pulsar.client.api.MessageId;
+
+/**
+ * Tracks any partial acked batches and its acked messages
+ * This will prevent acked message redelivery to the client only at the client API level.
+ * This class does not track batch with all acked message. We trust broker won't deliver again.
+ */
+class BatchAckedTracker {
+
+ // a map of partial acked batch and its messages already acked
+ @VisibleForTesting
+ Map<String, Set<BatchMessageIdImpl>> ackedBatches = new HashMap<String, Set<BatchMessageIdImpl>>();
+
+ public BatchAckedTracker() {
+ }
+
+ // we start to track
+ public void negativeAcked(BatchMessageIdImpl messageId) {
+
+ }
+
+ // If this message should be delivered and tracks this message if it is a batch message
+ public boolean deliver(MessageId messageId) {
+ if (messageId instanceof BatchMessageIdImpl) {
+ BatchMessageIdImpl id = (BatchMessageIdImpl) messageId;
+ String batchId = getBatchId(id);
+ if (ackedBatches.containsKey(batchId)){
+ Set<BatchMessageIdImpl> batch = ackedBatches.get(batchId);
+ if (batch.contains(id)) {
+ return false;
+ }
+ }
+ }
+ // deliver non batch message and any other cases
+ return true;
+ }
+
+ /**
+ *
+ * @param messageId batchMessageIdImpl
+ * @return boolean isAllMsgAcked for the batch
+ */
+ public boolean ack (BatchMessageIdImpl messageId) {
Review comment:
Concurrent access is managed by the underline batchedAcked concurrent map and synchronized set. I try to manage the locks at the lowest data structure level as possible.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services