Fix backward compatibility issues with batch index acknowledgment. (a…

…pache#7655) ### Motivation Fix backward compatibility issues with batch index acknowledgment. ### Modifications Disable batch index acknowledgment by default at the consumer side.
instructure · Jul 28, 2020 · fffd9f1 · fffd9f1
1 parent 42d17c8
commit fffd9f1
Show file tree

Hide file tree

Showing 12 changed files with 163 additions and 3 deletions.
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BatchMessageIndexAckTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BatchMessageIndexAckTest.java
@@ -67,6 +67,7 @@ public void testBatchMessageIndexAckForSharedSubscription() throws PulsarClientE
             .subscriptionName("sub")
             .receiverQueueSize(100)
             .subscriptionType(SubscriptionType.Shared)
+            .enableBatchIndexAcknowledgment(true)
             .negativeAckRedeliveryDelay(2, TimeUnit.SECONDS)
             .subscribe();
 
@@ -138,6 +139,7 @@ public void testBatchMessageIndexAckForExclusiveSubscription() throws PulsarClie
             .topic(topic)
             .subscriptionName("sub")
             .receiverQueueSize(100)
+            .enableBatchIndexAcknowledgment(true)
             .subscribe();
 
         @Cleanup
@@ -207,6 +209,7 @@ public void testDoNotRecycleAckSetMultipleTimes() throws Exception  {
         Consumer<byte[]> consumer = pulsarClient.newConsumer()
                 .acknowledgmentGroupTime(1, TimeUnit.MILLISECONDS)
                 .topic(topic)
+                .enableBatchIndexAcknowledgment(true)
                 .subscriptionName("test")
                 .subscribe();
 

diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java
@@ -609,6 +609,12 @@ public interface ConsumerBuilder<T> extends Cloneable {
      */
     ConsumerBuilder<T> enableRetry(boolean retryEnable);
 
+    /**
+     * Enable or disable the batch index acknowledgment. To enable this feature must ensure batch index acknowledgment
+     * feature is enabled at the broker side.
+     */
+    ConsumerBuilder<T> enableBatchIndexAcknowledgment(boolean batchIndexAcknowledgmentEnabled);
+
     /**
      * Consumer buffers chunk messages into memory until it receives all the chunks of the original message. While
      * consuming chunk-messages, chunks from same message might not be contiguous in the stream and they might be mixed

diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java
@@ -415,6 +415,12 @@ public ConsumerBuilder<T> enableRetry(boolean retryEnable) {
         return this;
     }
 
+    @Override
+    public ConsumerBuilder<T> enableBatchIndexAcknowledgment(boolean batchIndexAcknowledgmentEnabled) {
+        conf.setBatchIndexAckEnabled(batchIndexAcknowledgmentEnabled);
+        return this;
+    }
+
     @Override
     public ConsumerBuilder<T> expireTimeOfIncompleteChunkedMessage(long duration, TimeUnit unit) {
         conf.setExpireTimeOfIncompleteChunkedMessageMillis(unit.toMillis(duration));

diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
@@ -544,9 +544,11 @@ protected CompletableFuture<Void> doAcknowledge(MessageId messageId, AckType ack
                             ackType);
                 }
             } else {
-                BatchMessageIdImpl batchMessageId = (BatchMessageIdImpl) messageId;
-                acknowledgmentsGroupingTracker.addBatchIndexAcknowledgment(batchMessageId, batchMessageId.getBatchIndex(),
-                    batchMessageId.getBatchSize(), ackType, properties);
+                if (conf.isBatchIndexAckEnabled()) {
+                    BatchMessageIdImpl batchMessageId = (BatchMessageIdImpl) messageId;
+                    acknowledgmentsGroupingTracker.addBatchIndexAcknowledgment(batchMessageId, batchMessageId.getBatchIndex(),
+                            batchMessageId.getBatchSize(), ackType, properties);
+                }
                 // other messages in batch are still pending ack.
                 return CompletableFuture.completedFuture(null);
             }

diff --git a/...ar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationData.java b/...ar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationData.java
@@ -125,6 +125,8 @@ public class ConsumerConfigurationData<T> implements Serializable, Cloneable {
 
     private KeySharedPolicy keySharedPolicy;
 
+    private boolean batchIndexAckEnabled = false;
+
     @JsonIgnore
     public String getSingleTopic() {
         checkArgument(topicNames.size() == 1);

diff --git a/.../apache/pulsar/tests/integration/backwardscompatibility/PulsarStandaloneTestSuite2_5.java b/.../apache/pulsar/tests/integration/backwardscompatibility/PulsarStandaloneTestSuite2_5.java
@@ -0,0 +1,42 @@
+/**
+ * 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.tests.integration.backwardscompatibility;
+
+import org.apache.pulsar.tests.integration.containers.PulsarContainer;
+import org.apache.pulsar.tests.integration.topologies.PulsarStandaloneTestBase;
+import org.testng.ITest;
+import org.testng.annotations.AfterSuite;
+import org.testng.annotations.BeforeSuite;
+
+public class PulsarStandaloneTestSuite2_5 extends PulsarStandaloneTestBase implements ITest {
+
+    @BeforeSuite
+    public void setUpCluster() throws Exception {
+        super.startCluster(PulsarContainer.PULSAR_2_5_IMAGE_NAME);
+    }
+
+    @AfterSuite
+    public void tearDownCluster() throws Exception {
+        super.stopCluster();
+    }
+    @Override
+    public String getTestName() {
+        return "pulsar-standalone-suite";
+    }
+}
diff --git a/...rc/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/SmokeTest2_2.java b/...rc/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/SmokeTest2_2.java
@@ -33,4 +33,8 @@ public void testBatchMessagePublishAndConsume(String serviceUrl, boolean isPersi
         super.testBatchMessagePublishAndConsume(serviceUrl, isPersistent);
     }
 
+    @Test(dataProvider = "StandaloneServiceUrlAndTopics")
+    public void testBatchIndexAckDisabled(String serviceUrl, boolean isPersistent) throws Exception {
+        super.testBatchIndexAckDisabled(serviceUrl);
+    }
 }
diff --git a/...rc/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/SmokeTest2_3.java b/...rc/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/SmokeTest2_3.java
@@ -33,4 +33,8 @@ public void testBatchMessagePublishAndConsume(String serviceUrl, boolean isPersi
         super.testBatchMessagePublishAndConsume(serviceUrl, isPersistent);
     }
 
+    @Test(dataProvider = "StandaloneServiceUrlAndTopics")
+    public void testBatchIndexAckDisabled(String serviceUrl, boolean isPersistent) throws Exception {
+        super.testBatchIndexAckDisabled(serviceUrl);
+    }
 }
diff --git a/...rc/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/SmokeTest2_4.java b/...rc/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/SmokeTest2_4.java
@@ -33,4 +33,8 @@ public void testBatchMessagePublishAndConsume(String serviceUrl, boolean isPersi
         super.testBatchMessagePublishAndConsume(serviceUrl, isPersistent);
     }
 
+    @Test(dataProvider = "StandaloneServiceUrlAndTopics")
+    public void testBatchIndexAckDisabled(String serviceUrl, boolean isPersistent) throws Exception {
+        super.testBatchIndexAckDisabled(serviceUrl);
+    }
 }
diff --git a/...rc/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/SmokeTest2_5.java b/...rc/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/SmokeTest2_5.java
@@ -0,0 +1,40 @@
+/**
+ * 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.tests.integration.backwardscompatibility;
+
+import org.testng.annotations.Test;
+
+public class SmokeTest2_5 extends PulsarStandaloneTestSuite2_5 {
+
+    @Test(dataProvider = "StandaloneServiceUrlAndTopics")
+    public void testPublishAndConsume(String serviceUrl, boolean isPersistent) throws Exception {
+        super.testPublishAndConsume(serviceUrl, isPersistent);
+    }
+
+    @Test(dataProvider = "StandaloneServiceUrlAndTopics")
+    public void testBatchMessagePublishAndConsume(String serviceUrl, boolean isPersistent) throws Exception {
+        super.testBatchMessagePublishAndConsume(serviceUrl, isPersistent);
+    }
+
+    @Test(dataProvider = "StandaloneServiceUrlAndTopics")
+    public void testBatchIndexAckDisabled(String serviceUrl, boolean isPersistent) throws Exception {
+        super.testBatchIndexAckDisabled(serviceUrl);
+    }
+}
diff --git a/...gration/src/test/java/org/apache/pulsar/tests/integration/containers/PulsarContainer.java b/...gration/src/test/java/org/apache/pulsar/tests/integration/containers/PulsarContainer.java
@@ -41,6 +41,7 @@ public abstract class PulsarContainer<SelfT extends PulsarContainer<SelfT>> exte
     public static final int BROKER_HTTP_PORT = 8080;
 
     public static final String DEFAULT_IMAGE_NAME = "apachepulsar/pulsar-test-latest-version:latest";
+    public static final String PULSAR_2_5_IMAGE_NAME = "apachepulsar/pulsar:2.5.0";
     public static final String PULSAR_2_4_IMAGE_NAME = "apachepulsar/pulsar:2.4.0";
     public static final String PULSAR_2_3_IMAGE_NAME = "apachepulsar/pulsar:2.3.0";
     public static final String PULSAR_2_2_IMAGE_NAME = "apachepulsar/pulsar:2.2.0";

diff --git a/...egration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarTestBase.java b/...egration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarTestBase.java
@@ -24,13 +24,17 @@
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+
 import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.Message;
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.Producer;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionType;
 import org.apache.pulsar.common.util.FutureUtil;
+import org.junit.Assert;
 
 public class PulsarTestBase {
 
@@ -130,4 +134,46 @@ public void testBatchMessagePublishAndConsume(String serviceUrl, boolean isPersi
         }
     }
 
+    public void testBatchIndexAckDisabled(String serviceUrl) throws Exception {
+        String topicName = generateTopicName("test-batch-index-ack-disabled", true);
+        final int numMessages = 100;
+        try (PulsarClient client = PulsarClient.builder()
+                .serviceUrl(serviceUrl)
+                .build()) {
+
+            try (Consumer<Integer> consumer = client.newConsumer(Schema.INT32)
+                    .topic(topicName)
+                    .subscriptionName("sub")
+                    .receiverQueueSize(100)
+                    .subscriptionType(SubscriptionType.Shared)
+                    .enableBatchIndexAcknowledgment(false)
+                    .ackTimeout(1, TimeUnit.SECONDS)
+                    .subscribe();) {
+
+                try (Producer<Integer> producer = client.newProducer(Schema.INT32)
+                        .topic(topicName)
+                        .batchingMaxPublishDelay(50, TimeUnit.MILLISECONDS)
+                        .create()) {
+
+                    List<CompletableFuture<MessageId>> futures = new ArrayList<>();
+                    for (int i = 0; i < numMessages; i++) {
+                        futures.add(producer.sendAsync(i));
+                    }
+                    // Wait for all messages are publish succeed.
+                    FutureUtil.waitForAll(futures).get();
+                }
+
+                for (int i = 0; i < numMessages; i++) {
+                    Message<Integer> m = consumer.receive();
+                    if (i % 2 == 0) {
+                        consumer.acknowledge(m);
+                    }
+                }
+
+                Message<Integer> redelivery = consumer.receive(3, TimeUnit.SECONDS);
+                Assert.assertNotNull(redelivery);
+            }
+        }
+    }
+
 }