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/28 10:04:52 UTC
[GitHub] [pulsar] youurayy opened a new issue #6157: Failing to cumulative
acknowledge messages as old as 15 seconds
youurayy opened a new issue #6157: Failing to cumulative acknowledge messages as old as 15 seconds
URL: https://github.com/apache/pulsar/issues/6157
Observed behavior: Topic backlog is filling up and never cleared.
Expected behavior: Topic backlog is cleared.
In further tests observed to exhibit over as low as 5 seconds / 5 consecutive messages.
- Pulsar 2.4.2, cluster of 4 * brokers/proxies/bookies/zk
- Single persistent non-partitioned topic
- Failover subscription
- Single Producer
- Single Consumer
Namespace creation:
```sh
bin/pulsar-admin tenants create cb
bin/pulsar-admin namespaces create cb/test
bin/pulsar-admin namespaces set-backlog-quota cb/test --limit 300M --policy consumer_backlog_eviction
bin/pulsar-admin namespaces set-deduplication cb/test --enable
```
Client (shared code):
```java
public class PulsarTestBase {
protected SystemLogger logger;
protected String clientId;
protected PulsarClient pulsar;
protected String pulsarUrl = "pulsar://3.234.220.124:30004/";
protected String topicName = "persistent://cb/test/topic-1";
PulsarTestBase(String clientId) {
this.clientId = clientId;
logger = new SystemLogger.Builder(clientId).withCloudwatch(false).withPrometheus(false).build();
}
protected void createClient() throws PulsarClientException {
pulsar =
PulsarClient.builder()
.serviceUrl(pulsarUrl)
.connectionsPerBroker(1) // default: 1
.connectionTimeout(10, TimeUnit.SECONDS)
.enableTcpNoDelay(true)
.keepAliveInterval(999, TimeUnit.DAYS)
.maxBackoffInterval(5, TimeUnit.SECONDS)
.startingBackoffInterval(1, TimeUnit.SECONDS)
.statsInterval(60, TimeUnit.SECONDS)
// number of threads to be used for handling connections to brokers
.ioThreads(1) // default: 1
// number of threads to be used for message listeners
.listenerThreads(1) // default: 1
// number of concurrent lookup-requests allowed to send on each broker-connection
.maxConcurrentLookupRequests(5000) // default: 5000
.maxLookupRequests(50000) // default: 50000
// how many broker-rejects within 30 secs before connection to broker is recycled
.maxNumberOfRejectedRequestPerConnection(50) // default: 50
// how long to retry a broker op before the op is marked as failed
.operationTimeout(30, TimeUnit.SECONDS) // default: 30 sec
.build();
}
}
```
Producer code:
```java
public class TestProducer extends PulsarTestBase {
public static void main(String[] args) throws Exception {
new TestProducer(args[0]);
}
TestProducer(String clientId) throws Exception {
super(clientId);
createClient();
Producer<Order> producer =
pulsar
.newProducer(Schema.PROTOBUF(Order.class))
.producerName(clientId)
.topic(topicName)
.batchingMaxPublishDelay(1, TimeUnit.MILLISECONDS) // default: 1 ms
.batchingMaxMessages(1000) // default: 1000
// batching disabled, as we produce far more rarely than 1ms range
.enableBatching(false) // default: true
// block vs throw exception if pending-ack queue is full
.blockIfQueueFull(false) // default: false
// how many non-acked msgs to buffer before block or throw-exception
.maxPendingMessages(1000) // default: 1000
.sendTimeout(30, TimeUnit.SECONDS) // default: 30 sec
// .initialSequenceId() // TODO what if non-existent ID passed?
.hashingScheme(HashingScheme.JavaStringHash) // default: java string
.create();
V3.log("msg", "producing messages");
long counter = 0L;
while (true) {
Thread.sleep(1000L);
V3.log("msg", "producing message", "counter", counter);
producer
.newMessage()
.value(Order.newBuilder().setPrice(1d).setSize(counter).build())
.send();
counter++;
}
}
private final BiConsumer<? super MessageId, ? super Throwable> producerResultHandler =
new BiConsumer<MessageId, Throwable>() {
@Override
public void accept(MessageId messageId, Throwable ex) {
if (ex != null) {
V3.log(ex, "msg", "pulsar producer submit error", "*metric", "DroppedData", "value", 1);
}
}
};
}
```
Consumer code:
```java
public class TestConsumer extends PulsarTestBase {
public static void main(String[] args) throws Exception {
new TestConsumer(args[0]);
}
TestConsumer(String clientId) throws Exception {
super(clientId);
createClient();
Consumer<Order> consumer =
pulsar
.newConsumer(Schema.PROTOBUF(Order.class))
.consumerName(clientId)
.subscriptionInitialPosition(SubscriptionInitialPosition.Latest)
.subscriptionType(SubscriptionType.Failover)
.subscriptionName("failover-subscription")
.topic(topicName)
.subscribe();
MessageId firstMsgIdToAck = null;
int msgAckDelayCounter = 0;
V3.log("msg", "waiting for messages");
while (true) {
Message<Order> message = consumer.receive();
Order order = message.getValue();
V3.log(
"msg", "received message",
"sequenceId", message.getSequenceId(),
"price", order.getPrice(),
"size", order.getSize(),
"producer", message.getProducerName(),
"messageId", Pulsar.msgIdToHex(message.getMessageId()));
if (firstMsgIdToAck == null) {
firstMsgIdToAck = message.getMessageId();
V3.log("msg", "storing ack message id", "messageId", Pulsar.msgIdToHex(firstMsgIdToAck));
}
if (++msgAckDelayCounter == 15) {
V3.log("msg", "acknowledging message id", "messageId", Pulsar.msgIdToHex(firstMsgIdToAck));
consumer.acknowledgeCumulativeAsync(firstMsgIdToAck);
firstMsgIdToAck = null;
msgAckDelayCounter = 0;
}
}
}
}
```
Producer log screenshot:
Consumer log screenshot:
Admin console screenshot:
----------------------------------------------------------------
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 #6157: Failing to
cumulative acknowledge messages as old as 15 seconds
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on issue #6157: Failing to cumulative acknowledge messages as old as 15 seconds
URL: https://github.com/apache/pulsar/issues/6157#issuecomment-581725080
@youurayy Sorry, Maybe I haven't get your point. The ack period is control by users, every 10 messages or 20 messages, and consumer enabled group acknowledge by default(default is 100 mills). If want a smaller ack period, decrease the group acknowledge interval and increase cumulative acknowledge frequency.
----------------------------------------------------------------
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 #6157: Failing to
cumulative acknowledge messages as old as 15 seconds
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on issue #6157: Failing to cumulative acknowledge messages as old as 15 seconds
URL: https://github.com/apache/pulsar/issues/6157#issuecomment-582235048
@youurayy
I have run your above example code before, but all stats looks right. The msgBacklog between 15 to 30(I have check the stats and internal-stats many times). This corresponds to your sample code.
From your code:
1. For every 15 messages, the first message is acked.
2. Until next 15 ack happens, need more 15 messages.
So the backlog is between 15 to 30 right? If you ack the last message every 15 messages, the backlog would be 0 to 15(I also test this scenario).
----------------------------------------------------------------
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] youurayy commented on issue #6157: Failing to cumulative
acknowledge messages as old as 15 seconds
Posted by GitBox <gi...@apache.org>.
youurayy commented on issue #6157: Failing to cumulative acknowledge messages as old as 15 seconds
URL: https://github.com/apache/pulsar/issues/6157#issuecomment-581859121
@codelipenghui please try to run the above example code (for a longer time, e.g. 1 minute), and tell me why the messages are *not* getting acknowledged.
If you let the example running, the number of messages that **fail to acknowledge** _increases_.
When a cumulative acknowledgement request is issue (by `consumer.acknowledgeCumulativeAsync(messageId)`), **all** previous messages must be acknowledged by the broker.
This however is not happening.
----------------------------------------------------------------
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] youurayy commented on issue #6157: Failing to cumulative
acknowledge messages as old as 15 seconds
Posted by GitBox <gi...@apache.org>.
youurayy commented on issue #6157: Failing to cumulative acknowledge messages as old as 15 seconds
URL: https://github.com/apache/pulsar/issues/6157#issuecomment-581298132
@codelipenghui yes but that's irrelevant - the fact is that with a smaller ack period the acks go all through. Higher periods will leak acks. This makes Pulsar unusable in all scenarios where consumers need to take more than a few seconds to process and then ack 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] youurayy closed issue #6157: Failing to cumulative
acknowledge messages as old as 15 seconds
Posted by GitBox <gi...@apache.org>.
youurayy closed issue #6157: Failing to cumulative acknowledge messages as old as 15 seconds
URL: https://github.com/apache/pulsar/issues/6157
----------------------------------------------------------------
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] jiazhai commented on issue #6157: Failing to cumulative
acknowledge messages as old as 15 seconds
Posted by GitBox <gi...@apache.org>.
jiazhai commented on issue #6157: Failing to cumulative acknowledge messages as old as 15 seconds
URL: https://github.com/apache/pulsar/issues/6157#issuecomment-581227430
Thanks for the detailed issue reporting @youurayy . we will look into it.
@codelipenghui Is it related with the dispatcher issue that you fixed recently?
----------------------------------------------------------------
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 edited a comment on issue #6157: Failing to
cumulative acknowledge messages as old as 15 seconds
Posted by GitBox <gi...@apache.org>.
codelipenghui edited a comment on issue #6157: Failing to cumulative acknowledge messages as old as 15 seconds
URL: https://github.com/apache/pulsar/issues/6157#issuecomment-581232095
@youurayy I'm not sure the acknowledge request is sent.
```java
if (++msgAckDelayCounter == 15) {
V3.log("msg", "acknowledging message id", "messageId", Pulsar.msgIdToHex(firstMsgIdToAck));
consumer.acknowledgeCumulativeAsync(firstMsgIdToAck);
firstMsgIdToAck = null;
msgAckDelayCounter = 0;
}
```
I notice that sending 15 messages from the log, `msgAckDelayCounter` can't reach 15 right?
----------------------------------------------------------------
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 #6157: Failing to
cumulative acknowledge messages as old as 15 seconds
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on issue #6157: Failing to cumulative acknowledge messages as old as 15 seconds
URL: https://github.com/apache/pulsar/issues/6157#issuecomment-582721045
@youurayy You are welcome. Feel free to start the discussion again if you find any behaviors in pulsar are not match your requirement.
----------------------------------------------------------------
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 edited a comment on issue #6157: Failing to
cumulative acknowledge messages as old as 15 seconds
Posted by GitBox <gi...@apache.org>.
codelipenghui edited a comment on issue #6157: Failing to cumulative acknowledge messages as old as 15 seconds
URL: https://github.com/apache/pulsar/issues/6157#issuecomment-582235048
@youurayy
I have run your above example code before, but all stats looks right. The msgBacklog between 15 to 30(I have check the stats and internal-stats many times). This corresponds to your sample code.
From your code:
1. For every 15 messages, the first message is acked.
2. Until next ack happens, need more 15 messages.
So the backlog is between 15 to 30 right? If you ack the last message every 15 messages, the backlog would be 0 to 15(I also test this scenario).
----------------------------------------------------------------
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] youurayy edited a comment on issue #6157: Failing to
cumulative acknowledge messages as old as 15 seconds
Posted by GitBox <gi...@apache.org>.
youurayy edited a comment on issue #6157: Failing to cumulative acknowledge messages as old as 15 seconds
URL: https://github.com/apache/pulsar/issues/6157#issuecomment-581859121
@codelipenghui please try to run the above example code (for a longer time, e.g. 1 minute), and tell me why the messages are **not** getting acknowledged.
If you let the example running, the number of messages that **fail to acknowledge** _increases_.
When a cumulative acknowledgement request is issued (by `consumer.acknowledgeCumulativeAsync(messageId)`), **all** previous messages must be acknowledged by the broker.
This however is not happening.
----------------------------------------------------------------
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] youurayy commented on issue #6157: Failing to cumulative
acknowledge messages as old as 15 seconds
Posted by GitBox <gi...@apache.org>.
youurayy commented on issue #6157: Failing to cumulative acknowledge messages as old as 15 seconds
URL: https://github.com/apache/pulsar/issues/6157#issuecomment-582673821
@codelipenghui - thanks for looking into this - you are right, and it seems like our app indeed has an issue with a data leak. I'll reopen this later if there's a really problem on the pulsar side. Thanks!
----------------------------------------------------------------
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 #6157: Failing to
cumulative acknowledge messages as old as 15 seconds
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on issue #6157: Failing to cumulative acknowledge messages as old as 15 seconds
URL: https://github.com/apache/pulsar/issues/6157#issuecomment-581232095
@youurayy I'm not sure the acknowledge request is sent.
```
if (++msgAckDelayCounter == 15) {
V3.log("msg", "acknowledging message id", "messageId", Pulsar.msgIdToHex(firstMsgIdToAck));
consumer.acknowledgeCumulativeAsync(firstMsgIdToAck);
firstMsgIdToAck = null;
msgAckDelayCounter = 0;
}
```
I notice that sending 15 messages from the log, `msgAckDelayCounter` can't reach 15 right?
----------------------------------------------------------------
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