tree 9861f60220f51c44a5cfe5b3ed2066d47048a737
parent 793ea91e37ead175ccc6d61a62310fd8d3f968d1
author wuYin <wuyinpost@gmail.com> 1604566040 +0800
committer GitHub <noreply@github.com> 1604566040 +0800
gpgsig -----BEGIN PGP SIGNATURE-----
 
 wsBcBAABCAAQBQJfo7wYCRBK7hj4Ov3rIwAAdHIIAD3Yd4lnKeelICLOi0Dh2qHg
 8GQEqbCLVVzqT6iOxSxKqnRkGiU/VSHYtL+pdOC2Y1ChTikfJdZSX1pssQl92x1M
 FXcF5YDGzvEeAc1RQ7p8VHKt60pxWnZFB67TO2C4Bx5m0O8A2HAFKxAmn27f4jv9
 71crRXfFm9SmxPPFpXpgwmt84pOZR7Lhypf/V01ZSVsVAYwDt/sa4EYK9ZJ8P80Y
 NciPTrl57jD9AsYfKGQWjWSsIl703+TY5CloF0vQlFikRk+Xv6U9L1VWYwUQzzMq
 uISxYBM2910j5bUIONIzjW6QEOOL+ogUcMK5s5sIqquWofnPuwRA1FE6H/8DViw=
 =K7Nh
 -----END PGP SIGNATURE-----
 

Fix reconsume broken while using non-FQDN topics (#386)

### Issue
Retry policy not effective with non-FQDN topic.

- reproduction
	```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

	```
	RN[0000] consumer of topic [persistent://public/default/topic-01] not exist unexpectedly  topic="[topic-01 persistent://public/default/my-sub-RETRY]"
	```

### Cause
For MultiTopicConsumer `consumers` map filed:
- key: user provided topic, maybe non-FQDN.
- value: consumer instance.

`ReconsumeLater` using msg's FQDN topic as key to find `consumer` in `consumers`,
 if mismatch with non-FQDN topic, this invoke will be ignored, lead to Retry policy not effective.

### Modifications
- Normalize user provided topics as FQDN topics before initializing consumers.
- Add non-FQDN topic consumption case in Retry policy tests.


### Verifying this change

- [x] Make sure that the change passes the CI checks.