merlimat closed pull request #2047: Fixed resetting of producer sequence id
counter after receiving SendError before any message was successfully published
URL: https://github.com/apache/incubator-pulsar/pull/2047
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerBkEnsemblesTests.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerBkEnsemblesTests.java
index 173fb493be..4ecdae497b 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerBkEnsemblesTests.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerBkEnsemblesTests.java
@@ -60,11 +60,11 @@
*/
public class BrokerBkEnsemblesTests {
protected static int BROKER_SERVICE_PORT = 16650;
- PulsarService pulsar;
+ protected PulsarService pulsar;
ServiceConfiguration config;
URL adminUrl;
- PulsarAdmin admin;
+ protected PulsarAdmin admin;
LocalBookkeeperEnsemble bkEnsemble;
@@ -83,7 +83,7 @@ public BrokerBkEnsemblesTests(int numberOfBookies) {
}
@BeforeMethod
- void setup() throws Exception {
+ protected void setup() throws Exception {
try {
// start local bookie and zookeeper
bkEnsemble = new LocalBookkeeperEnsemble(numberOfBookies,
ZOOKEEPER_PORT, 5001);
@@ -118,7 +118,7 @@ void setup() throws Exception {
}
@AfterMethod
- void shutdown() throws Exception {
+ protected void shutdown() throws Exception {
try {
admin.close();
pulsar.close();
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/RackAwareTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/RackAwareTest.java
index 215d3503a8..7499908485 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/RackAwareTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/RackAwareTest.java
@@ -52,7 +52,7 @@ public RackAwareTest() {
@SuppressWarnings("deprecation")
@BeforeClass
- void setup() throws Exception {
+ protected void setup() throws Exception {
super.setup();
// Start bookies with specific racks
@@ -83,7 +83,7 @@ void setup() throws Exception {
}
@AfterClass
- void shutdown() throws Exception {
+ protected void shutdown() throws Exception {
super.shutdown();
for (BookieServer bs : bookies) {
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/SequenceIdWithErrorTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/SequenceIdWithErrorTest.java
new file mode 100644
index 0000000000..21a65a6dd0
--- /dev/null
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/SequenceIdWithErrorTest.java
@@ -0,0 +1,102 @@
+/**
+ * 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 static org.testng.Assert.assertEquals;
+
+import java.util.Collections;
+
+import org.apache.bookkeeper.mledger.ManagedLedger;
+import org.apache.bookkeeper.mledger.ManagedLedgerFactory;
+import org.apache.pulsar.broker.ManagedLedgerClientFactory;
+import org.apache.pulsar.broker.service.BrokerBkEnsemblesTests;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+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.common.naming.TopicName;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+public class SequenceIdWithErrorTest extends BrokerBkEnsemblesTests {
+ @Override
+ @BeforeMethod
+ public void setup() throws Exception {
+ super.setup();
+ }
+
+ @AfterMethod
+ public void cleanup() throws Exception {
+ super.shutdown();
+ }
+
+ /**
+ * Test that sequence id from a producer is correct when there are send
errors
+ */
+ @Test
+ public void testCheckSequenceId() throws Exception {
+ admin.namespaces().createNamespace("prop/my-test",
Collections.singleton("usc"));
+
+ String topicName = "prop/my-test/my-topic";
+ int N = 10;
+
+ PulsarClient client =
PulsarClient.builder().serviceUrl("pulsar://localhost:" +
BROKER_SERVICE_PORT).build();
+
+ // Create consumer
+ Consumer<String> consumer =
client.newConsumer(Schema.STRING).topic(topicName).subscriptionName("sub")
+ .subscribe();
+
+ // Fence the topic by opening the ManagedLedger for the topic outside
the Pulsar broker. This will cause the
+ // broker to fail subsequent send operation and it will trigger a
recover
+ ManagedLedgerClientFactory clientFactory = new
ManagedLedgerClientFactory(pulsar.getConfiguration(),
+ pulsar.getZkClient(), pulsar.getBookKeeperClientFactory());
+ ManagedLedgerFactory mlFactory =
clientFactory.getManagedLedgerFactory();
+ ManagedLedger ml =
mlFactory.open(TopicName.get(topicName).getPersistenceNamingEncoding());
+ ml.close();
+ clientFactory.close();
+
+ // Create a producer
+ Producer<String> producer =
client.newProducer(Schema.STRING).topic(topicName).create();
+
+ for (int i = 0; i < N; i++) {
+ producer.send("Hello-" + i);
+ }
+
+ for (int i = 0; i < N; i++) {
+ Message<String> msg = consumer.receive();
+ assertEquals(msg.getValue(), "Hello-" + i);
+ assertEquals(msg.getSequenceId(), i);
+ consumer.acknowledge(msg);
+ }
+
+ client.close();
+ }
+
+ @Override
+ public void testCrashBrokerWithoutCursorLedgerLeak() throws Exception {
+ // Ignore test
+ }
+
+ @Override
+ public void testSkipCorruptDataLedger() throws Exception {
+ // Ignore test
+ }
+}
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
index b5962ac8e3..5d1ac4142b 100644
---
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
@@ -873,7 +873,10 @@ public void connectionOpened(final ClientCnx cnx) {
this.producerName = producerName;
}
- if (this.lastSequenceIdPublished == -1 &&
conf.getInitialSequenceId() == null) {
+ if (this.msgIdGenerator == 0 &&
conf.getInitialSequenceId() == null) {
+ // Only update sequence id generator if it wasn't
already modified. That means we only want
+ // to update the id generator the first time the
producer gets established, and ignore the
+ // sequence id sent by broker in subsequent
producer reconnects
this.lastSequenceIdPublished = lastSequenceId;
this.msgIdGenerator = lastSequenceId + 1;
}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
