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 2017/10/31 21:44:39 UTC
[GitHub] merlimat closed pull request #874: Fix: Non-persistent topic acking sequence for dropped message
merlimat closed pull request #874: Fix: Non-persistent topic acking sequence for dropped message
URL: https://github.com/apache/incubator-pulsar/pull/874
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/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
index 5c48de606..df4be6a6f 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
@@ -588,15 +588,19 @@ protected void handleSend(CommandSend send, ByteBuf headersAndPayload) {
printSendCommandDebug(send, headersAndPayload);
}
- // avoid processing non-persist message if reached max concurrent-message limit
- if (producer.isNonPersistentTopic() && nonPersistentPendingMessages++ > MaxNonPersistentPendingMessages) {
- final long producerId = send.getProducerId();
- final long sequenceId = send.getSequenceId();
- service.getTopicOrderedExecutor().submitOrdered(producer.getTopic(), SafeRun.safeRun(() -> {
- ctx.writeAndFlush(Commands.newSendReceipt(producerId, sequenceId, -1, -1), ctx.voidPromise());
- }));
- producer.recordMessageDrop(send.getNumMessages());
- return;
+ if (producer.isNonPersistentTopic()) {
+ // avoid processing non-persist message if reached max concurrent-message limit
+ if (nonPersistentPendingMessages > MaxNonPersistentPendingMessages) {
+ final long producerId = send.getProducerId();
+ final long sequenceId = send.getSequenceId();
+ service.getTopicOrderedExecutor().submitOrdered(producer.getTopic().getName(), SafeRun.safeRun(() -> {
+ ctx.writeAndFlush(Commands.newSendReceipt(producerId, sequenceId, -1, -1), ctx.voidPromise());
+ }));
+ producer.recordMessageDrop(send.getNumMessages());
+ return;
+ } else {
+ nonPersistentPendingMessages++;
+ }
}
startSendOperation();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonPersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonPersistentTopicTest.java
index 2e3e421f7..4a8d8cf4c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonPersistentTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonPersistentTopicTest.java
@@ -50,16 +50,15 @@
import org.apache.pulsar.broker.service.nonpersistent.NonPersistentTopic;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.impl.ConsumerImpl;
+import org.apache.pulsar.client.impl.ProducerImpl;
import org.apache.pulsar.common.naming.DestinationName;
import org.apache.pulsar.common.naming.NamespaceBundle;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.NonPersistentPublisherStats;
import org.apache.pulsar.common.policies.data.NonPersistentSubscriptionStats;
import org.apache.pulsar.common.policies.data.NonPersistentTopicStats;
-import org.apache.pulsar.common.policies.data.SubscriptionStats;
-import org.apache.pulsar.common.policies.data.PersistentTopicStats;
import org.apache.pulsar.common.policies.data.PropertyAdmin;
-import org.apache.pulsar.common.policies.data.PublisherStats;
+import org.apache.pulsar.common.policies.data.SubscriptionStats;
import org.apache.pulsar.zookeeper.LocalBookkeeperEnsemble;
import org.apache.pulsar.zookeeper.ZookeeperServerTest;
import org.slf4j.Logger;
@@ -743,7 +742,8 @@ public void testMsgDropStat() throws Exception {
consumerConfig2.setReceiverQueueSize(1);
consumerConfig2.setSubscriptionType(SubscriptionType.Shared);
Consumer consumer2 = pulsarClient.subscribe(topicName, "subscriber-2", consumerConfig2);
- Producer producer = pulsarClient.createProducer(topicName, producerConf);
+ ProducerImpl producer = (ProducerImpl) pulsarClient.createProducer(topicName, producerConf);
+ String firstTimeConnected = producer.getConnectedSince();
ExecutorService executor = Executors.newFixedThreadPool(5);
byte[] msgData = "testData".getBytes();
final int totalProduceMessages = 200;
@@ -767,6 +767,8 @@ public void testMsgDropStat() throws Exception {
assertTrue(npStats.msgDropRate > 0);
assertTrue(sub1Stats.msgDropRate > 0);
assertTrue(sub2Stats.msgDropRate > 0);
+ // make sure producer connection not disconnected due to unordered ack
+ assertEquals(firstTimeConnected, producer.getConnectedSince());
producer.close();
consumer.close();
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
index a918776ef..7b3c6294f 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
@@ -183,8 +183,8 @@ protected void handleSendReceipt(CommandSendReceipt sendReceipt) {
}
if (ledgerId == -1 && entryId == -1) {
- log.warn("[{}] Message has been dropped for non-persistent topic producer-id {}", ctx.channel(),
- producerId);
+ log.warn("[{}] Message has been dropped for non-persistent topic producer-id {}-{}", ctx.channel(),
+ producerId, sequenceId);
}
if (log.isDebugEnabled()) {
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services