Fixes #544

Motivation

After the debugging, when a FETCH request contains many partitions, for each partition, KafkaTopicConsumerManager#remove will be executed in the same pulsar-io-<suffix> thread, which will eventually call MessageIdUtils#getPositionForOffset to get PositionImpl of the offset.

However, getPositionForOffset is a synchronous method that waits until a CompletableFuture is done. The future is returned by ManagedLedgerImpl#asyncFindPosition that calls ManagedLedgerImpl#asyncReadEntry internally. If the producer is publishing messages at the same time, the asyncAddEntry method could be called for the same ManagedLedgerImpl object. It will somehow cause the deadlock.

Modifications

• Just use ManagedLedgerImpl#asyncFindPosition instead of the synchronous call in MessageIdUtils#getPositionForOffset.
• Caches the future of Pair<ManagedCursor, Long> in KafkaTopicConsumerManager instead of the Pair<ManagedCursor, Long>. Then in MessageFetchContext, use the ManagedCursor in future's callback .
• Add a unit test to ensure after this change, the non-durable cursor will be still created once for each consumer. This condition is easily to break in a refactor like this PR and it may cause performance problem because cursor might be created each time a FETCH request arrived. So here I added a test to avoid the case.