tree 8b348fed37c1597da44425ee9908c0fcd37ddd5f
parent 9b20cc0841602f3767d32b19edb1db55fd0bb209
author wuYin <wuyinpost@gmail.com> 1609758622 +0800
committer GitHub <noreply@github.com> 1609758622 +0800
gpgsig -----BEGIN PGP SIGNATURE-----
 
 wsBcBAABCAAQBQJf8veeCRBK7hj4Ov3rIwAAdHIIAB1snN4SsMWzXIQc9+IQGS8v
 P2oKE9zSMF6Gi4WP5n13AgjK//FSQbSNBwP2l4/gzxY7DNCMQ4PgSRLcGgvXBbhE
 2FUmtWyFPmP+MZJ0XSR3y3Zz4H0MIsBm4ylhUIfVFEXJAmld5sVI2Nc5420iXiZJ
 BZbLHGWTtiostI0senMjukMPAQ+QNdcX6HBNrzUnOOH84W9sA6oDvKX/TRgpM7BF
 aLRH+ytcAsiq62FKtPJE4KdJ8uBWnLPoensLnmmCoJnGSFntAGovlhpmBU7st0sD
 pgqq0s4K0i6Mu8c71hnlMa3z+DMblJGdrj0zwJWay34GmJB/o1Cc7eoENLXteaE=
 =CvYI
 -----END PGP SIGNATURE-----
 

fix retry policy not effective with partitioned topic (#425)

### Issue
Retry policy not effective with partitioned topic.

- reproduction
 topic-01 have 2 partitions:
	```go
	client, _ := pulsar.NewClient(pulsar.ClientOptions{URL: "pulsar://localhost:6650"})
	consumer, _ := client.Subscribe(pulsar.ConsumerOptions{
		Topic:            "topic-01",
		SubscriptionName: "my-sub",
		RetryEnable:      true,
		DLQ:              &pulsar.DLQPolicy{MaxDeliveries: 2},
	})
	msg, _ := consumer.Receive(context.Background())
	consumer.ReconsumeLater(msg, 5*time.Second)
	```
- logs

    ```
    WARN[0000] consumer of topic [persistent://public/default/topic-01-partition-0] not exist unexpectedly  topic=" [persistent://public/default/topic-01 persistent://platform/psr/my-sub-RETRY]"
    ```

### Cause
For MultiTopicConsumer `consumers` map filed:
- key: FQDN topics, without partition number suffix.
- value: consumer instance.

`ReconsumeLater` using msg's partition number suffix topic as key, to find `consumer` in `consumers`, but allways not found, lead to Retry policy not effective.

### Modifications
- Trim partition number in partitioned topic before using it

### Verifying this change

- [x] Make sure that the change passes the CI checks.