Add best consume practice

[liangyepianzhou]

✅ Contribution Checklist

• I read the contribution guide
• I updated the versioned docs

[asafm]

Even if the topic has partitions across brokers? How?

[liangyepianzhou]

Yes, it is. For example, there is topic 1 with five partitions; five consumers can subscribe to topic 1 in the failover mode, and only one consumer can subscribe to topic 1 in the exclusive mode. The add consumer logic just like this:

                        case Exclusive:
                            if (dispatcher == null || dispatcher.getType() != SubType.Exclusive) {
                                previousDispatcher = dispatcher;
                                dispatcher = new PersistentDispatcherSingleActiveConsumer(
                                        cursor, SubType.Exclusive, 0, topic, this);
                            }
                            break;
.....
                        case Failover:
                            int partitionIndex = TopicName.getPartitionIndex(topicName);
                            if (partitionIndex < 0) {
                                // For non partition topics, use a negative index so
                                // dispatcher won't sort consumers before picking
                                // an active consumer for the topic.
                                partitionIndex = -1;
                            }

                            if (dispatcher == null || dispatcher.getType() != SubType.Failover) {
                                previousDispatcher = dispatcher;
                                dispatcher = new PersistentDispatcherSingleActiveConsumer(cursor, SubType.Failover,
                                                partitionIndex, topic, this);
                            }
                            break;

[asafm]

Back--> Stand-by. The other consumers which are not active, are also connected to the partition in stand-by.

[liangyepianzhou]

Great! Thanks for the correction.

[asafm]

Too confusing.
Key-shared allows to have messages of the same key (e.g. customer-id) sent to the same consumer, where in Shared the message are sent in a round-robin manner to all consumers.

[liangyepianzhou]

Key-shared allows to have messages of the same key (e.g. customer-id) sent to the same consumer

Key-shared sent messages with the same key to the same set of consumers instead of the single same consumer.

[liangyepianzhou]

And there are many selector implementations of the selector for the key-shared mode.
For the ConsistentHashingStickyKeyConsumerSelector, it could be like the following:

    public Consumer select(int hash) {
        rwLock.readLock().lock();
        try {
            if (hashRing.isEmpty()) {
                return null;
            }

            List<Consumer> consumerList;
            Map.Entry<Integer, List<Consumer>> ceilingEntry = hashRing.ceilingEntry(hash);
            if (ceilingEntry != null) {
                consumerList =  ceilingEntry.getValue();
            } else {
                consumerList = hashRing.firstEntry().getValue();
            }

            return consumerList.get(hash % consumerList.size());
        } finally {
            rwLock.readLock().unlock();
        }
    }

[asafm]

"subscription - in effect: all messages up to the selected message will be marked as processed (acknowledged)"

[asafm]

"For multiple partitions..." --> When consuming you might get messages from more then one partition. You need to make sure you call cumulative-acknowlege per the partition. @BewareMyPower I need help here

[liangyepianzhou]

The cumulative ack is recommended to be used in the exclusive mode and Failover mode.
For the Exclusive mode, the messages from all the partitions are sent to the same consumer. But the cumulative ack only marks the messages as processed for the single partition. For the Exclusive mode, if they consume 100 messages in a topic with 5 partitions and cumulative ack the last message that is a message in partition4. That does not mean the 100 messages before the last message are processed, it only means the messages in partition 4 before the last message are marked as processed. So I add a notice here.
You can see more information in the following tests.

    @Test
    public void test() throws Exception {
        String subName = "test";
        String topicName = TopicName.get(NAMESPACE1 + "/" + "testCreateTransactionSystemTopic").toString();
        admin.topics().createPartitionedTopic(topicName, 3);
        Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).create();
        Consumer<byte[]> consumer1 = pulsarClient.newConsumer().receiverQueueSize(100).topic(topicName)
                .subscriptionName(subName)
                .subscriptionType(SubscriptionType.Exclusive).subscribe();

        for (int i = 0; i < 100; i++) {
            producer.newMessage().send();
        }
        Message lastMessage = null;
        for (int i = 0; i < 100; i++) {
            lastMessage = consumer1.receive();
        }
        log.info("ack message {}", lastMessage.getMessageId());
        consumer1.acknowledgeCumulative(lastMessage);

        consumer1.redeliverUnacknowledgedMessages();
        for (int i = 0; i < 30; i++) {
            lastMessage = consumer1.receive();
            log.info("receive message {}", lastMessage.getMessageId());
        }
    }

[asafm]

I don't understand why. @codelipenghui ?

[liangyepianzhou]

When users use transactions to acknowledge messages, Pulsar must ensure that a message can not be acknowledged twice. The second acknowledgment for one message with a transaction will result in a TransactionConflictException and abort the transaction.

[asafm]

Why?

[liangyepianzhou]

The comment you left could be one reason. The other is that one transaction could include many messages, and sending them all to a retry topic or DLQ may not be appropriate when the transaction is aborted.

[asafm]

We have to stop here IMO. Something here in the structure makes it way too long IMO.

[asafm]

Didn't understand anything here.

[liangyepianzhou]

Maybe the expression here is not very precise. Pulsar has not been able to implement the Exactly-once semantics in Exclusive and Failover modes, and there have always been repeated acks.

[asafm]

Any chance to use mermaid for diagrams?

[visortelle]

+1

Mermaid is suitable for most cases, but for some complex diagrams (if any), it also could be draw.io (diagrams.net) that could be stored in the repository as .xml files.

[dao-jun]

should be Shared not Shaved