Delayed acks impl

[merlimat]

Motivation

This PR is based on top of #1450. I'll rebase once first one is merged.

The processing of acknowledgments constitutes is broker a big chunk of CPU work. Even then, we throttle the persistence of acks to BK (default is once per sec).

In addition to that, acknowledgments generates a large number of very small IP packets and these are most expensive to process by OS/NIC (compared to fewer bigger packets).

This change introduces client side batching for acknowledgments. By default client, will group the acks and send out a single protobuf command every 100ms. Consumer can configure the delay, or set it to 0 to fall back to original behavior.

[merlimat]

Rebased to master

[merlimat]

retest this please

[rdhabalia]

@merlimat this is not a backward compatible change with old broker (version < 2.0) because, old brokers don't understand group-ack command so, combination of new-client (>=2.0) and old-broker will not work and it can create ack-holes for the client.

also CommandAck schema is also changed

Before: required MessageIdData message_id = 3;
After: repeated MessageIdData message_id = 3;

I think we were making sure that we don't break backward compatibility so, If client uses new version and if it requires to rollback broker version then it can not create an issue here. Any thoughts to fix this issue?

[rdhabalia]

can we add CommandAckDeprecated in proto-schema which can be used by client if broker-verison is < V12 ?

[rdhabalia]

@merlimat any update on it?

[merlimat]

@rdhabalia The code should already be sending 1 single ack if the broker is < 2.0 :

https://github.com/apache/incubator-pulsar/blob/233c77fca1ac1bfbc439c2246a45673ff366bfc3/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java#L171 with the check validating for v12 : https://github.com/apache/incubator-pulsar/blob/1893323bc2f22001b63ea0d310ec47e88889e3e0/pulsar-common/src/main/java/org/apache/pulsar/common/api/Commands.java#L1073

Let me verify that the repeated field, when 1 single entity is present, is equivalent to the "required" type.

[merlimat]

@rdhabalia I did the test by using :

• Broker 1.22.1
• Producer 1.22 (perf-producer)
• Consumer from master (2.2-SNAPSHOT) (perf-consumer)

I could no reproduce the issue. There are no unacked messages. Do you have a better way to reproduce?

[rdhabalia]

The code should already be sending 1 single ack if the broker is < 2.0 :

I see, I missed it to check.

I think the issue is that I have cherry-picked ActiveConsumerListener-change into broker-1.22 to support function on broker-1.22 and this PR upgraded pulsar-broker version=V12 and because of that client was sending group-ack to broker which broker couldn't understand.
Thanks for pointing it out, I will fix internal broker's patch.