ISSUE-5840: Subscription mode Key_Shared not working as expected

[sijie]

Original Issue: apache#5840

---

Expected behavior

When running two instances of my app, i expect that pulsar should loadbalance between the two instances. Also, messages with the same key should always go to the same instance.

Actual behavior

However, all messages are handled by instance 1. Instance 2 only receive messages when instance 1 is down.

Steps to reproduce

Create a SpringBoot app with a single Pulsar Consumer Bean:

@Slf4j
@Component
public class CustomerCommandHandler {

  @Autowired
  private PulsarClient pulsarClient;

  @Bean
  public Consumer<Customer> consumer() throws PulsarClientException {
    return pulsarClient.newConsumer(Schema.JSON(Customer.class))
        .topic("eu-west/app1/customers-topic")
        .subscriptionType(SubscriptionType.Key_Shared)
        .subscriptionName("my-subscription")
        .messageListener((MessageListener<Customer>) (consumer, message) -> {
          try {
            log.info("Message received: {}", message.getValue());
            consumer.acknowledge(message);
          } catch (Exception e) {
            consumer.negativeAcknowledge(message);
            e.printStackTrace();
          }
        })
        .subscribe();
  }
}`

Run two instances of this app and puplish messages to the customers-topic.

System configuration

Pulsar version: 2.4.1 and 2.4.2