Avoid enable DLQ on Key_Shared subscription.

[MarvinCai]

Fixes #9156

Motivation

In Pulsar, DLQ only supports non-ordered subscriptions, But now the Key_Shared subscription also allowed enable DLQ.
Should avoid enable DLQ on the Key_Shared subscription tpye.

Modifications

Only send message to DLQ if not Key_Shared subscription type.

Verifying this change

This change added tests and can be verified as follows:

• Added unit test.

Does this pull request potentially affect one of the following parts:

If yes was chosen, please highlight the changes

• Dependencies (does it add or upgrade a dependency): no
• The public API: no
• The schema: no
• The default values of configurations: no
• The wire protocol: no
• The rest endpoints: no
• The admin cli options: no
• Anything that affects deployment: no

Documentation

• Does this pull request introduce a new feature? no
• If yes, how is the feature documented? not documented

[codelipenghui]

@MarvinCai We also need to add verification in the ConsumerBuilder. With the ConsumerBuilder, users should get a PulsarClient exception if they want to enable DLQ on a Key_Shared subscription.

[315157973]

The same logic exists in the void messageReceived method. Is it better for us to prohibit setting this kind of policy in the builder ?

if (deadLetterPolicy != null && possibleSendToDeadLetterTopicMessages != null && redeliveryCount >= deadLetterPolicy.getMaxRedeliverCount()) {
    possibleSendToDeadLetterTopicMessages.put((MessageIdImpl)message.getMessageId(), Collections.singletonList(message));
}

There is retry logic in deadLetterPolicy. If the number of retries is exceeded, how to deal with these messages? But the retryLetterProducer in consumerImpl doesn't seem to do any processing.

[MarvinCai]

@MarvinCai We also need to add verification in the ConsumerBuilder. With the ConsumerBuilder, users should get a PulsarClient exception if they want to enable DLQ on a Key_Shared subscription.

Added check in ConsumerBuilder to throw InvalidConfigurationException if set DeadLetterTopic name for KeyShared subType.

The same logic exists in the void messageReceived method. Is it better for us to prohibit setting this kind of policy in the builder ?

if (deadLetterPolicy != null && possibleSendToDeadLetterTopicMessages != null && redeliveryCount >= deadLetterPolicy.getMaxRedeliverCount()) {
    possibleSendToDeadLetterTopicMessages.put((MessageIdImpl)message.getMessageId(), Collections.singletonList(message));
}

There is retry logic in deadLetterPolicy. If the number of retries is exceeded, how to deal with these messages? But the retryLetterProducer in consumerImpl doesn't seem to do any processing.

For first point, there's a check possibleSendToDeadLetterTopicMessages != null and I've make possibleSendToDeadLetterTopicMessages = null if it's KeyShared subType which is the behavior if DeadLetterPolicy is not provided. (It's safe as we are already doing null check every time we want to use possibleSendToDeadLetterTopicMessages)
For second point, actually we need both check. The check in ConsumerBuilder will prevent user from setting DeadLetterTopic.
However, even if we don't set DeadLetterTopic(e in DeadLetterPolicy, in ConsumerImpl wether to put message in DLQ is currently decided by if reconsumetimes > this.deadLetterPolicy.getMaxRedeliverCount() so additional check will still be needed here as we want retry but not DLQ for Key_Shared subType.
Either by checking if subType == Key_Shared or if StringUtils.isBlank(deadLetterPolicy.getDeadLetterTopic()). I just pick the former cause it make our purpose more clear.

[sijie]

@315157973 @codelipenghui Can you review the PR again?

[MarvinCai]

Now ConsumerImpl constructor will throw exception when client try to directly construct a Consumer with Key_Shared subscription type and DeadLetterTopic. Same to all subclass like ZeroQueueCosumer, RawReader...., this shouldn't affect existing logic using ConsumerBuilder to create consumer as builder will check and make sure there's no DLQ set for Key_Shared sub type.

[MarvinCai]

/pulsarbot run-failure-checks

[MarvinCai]

/pulsarbot run-failure-checks

[MarvinCai]

/pulsarbot run-failure-checks

[MarvinCai]

/pulsarbot run-failure-checks

[MarvinCai]

/pulsarbot run-failure-checks

[MarvinCai]

/pulsarbot run-failure-checks

[MarvinCai]

/pulsarbot run-failure-checks

[MarvinCai]

/pulsarbot run-failure-checks

[MarvinCai]

/pulsarbot run-failure-checks