Message dispatching based on consumer priority-level

[rdhabalia]

Motivation

Addressing #134 : Message dispatching based on consumer priority-level

Result

Shared mode consumer priority, the messages will only goes to the consumers with higher priority if there's permit, otherwise the message can also be pushed to the consumers with lower priority.

[yahoocla]

CLA is valid!

[merlimat]

The change looks good, though I'd like to understand more the logic to see that it doesn't add any penalty when priorities are not used

[merlimat]

I think it would be more intuitive to use descending priorities. (eg: 0=max-priority, 1, 2, ).

Priority 0 should also be the effective default applied to consumers that either:

• Are running the old client version
• Haven't set the configuration value

[rdhabalia]

I think it would be more intuitive to use descending priorities. (eg: 0=max-priority, 1, 2, ).

Actually, right now that's how the implementation is: 0=max-priority, 1=second, 2=..
sorting: consumerList.sort((c1, c2) -> c1.getPriorityLevel() - c2.getPriorityLevel());

I will update the documentation part to make it more clear.

[merlimat]

The logic looks fine here, although it would be helpful to have a comment to explain the algorithm in its entirety.

Also, it would be good to verify here that there no performance regression in the case when priorities are not used. Eg. running a quick stress test and verify the CPU usage in broker doesn't shoot up compared to before.

[rdhabalia]

The logic looks fine here, although it would be helpful to have a comment to explain the algorithm in its entirety.

Added algorithm explaination

running a quick stress test and verify the CPU usage in broker doesn't shoot up compared to before.

I have run stress rest and I don't see impact in CPU usage. we don't perform extra logic when we have consumers on same priority-level.

without-changes of : Priority-level

with-changes of : Priority-level

Test
Shared-Subscription with 30 consumers

 @Test
    public void testSharedSubscptionConsumerLoad() throws Exception {
        final String topicName = "persistent://my-property/use/my-ns/my-topic1";
        final String subscriberName = "my-subscriber-name";
        final int totalConsumers = 30;

        ConsumerConfiguration conf = new ConsumerConfiguration();
        conf.setSubscriptionType(SubscriptionType.Shared);
        conf.setMessageListener((consumer, msg) -> {
            try {
                consumer.acknowledge(msg);
            } catch (PulsarClientException e) {
                e.printStackTrace();
            }
        });
        List<Consumer> consumers = Lists.newArrayList();
        for (int i = 0; i < totalConsumers; i++) {
            consumers.add(pulsarClient.subscribe(topicName, subscriberName, conf));
        }
        Producer producer = pulsarClient.createProducer("persistent://my-property/use/my-ns/my-topic1");
        byte[] msg = "my-message-".getBytes();
        int totalMsgs = Integer.MAX_VALUE;
        for (int i = 0; i < totalMsgs; i++) {
            producer.sendAsync(msg);
        }

    }

[merlimat]

👍