Remove RetryEnable

jiangpengcheng · Apr 9, 2024 · 18c98a4 · 18c98a4
1 parent 9461c8f
commit 18c98a4
Showing 1 changed file with 2 additions and 7 deletions.
diff --git a/pulsar-function-go/pf/instance.go b/pulsar-function-go/pf/instance.go
@@ -318,16 +318,15 @@ func (gi *goInstance) setupConsumer() (chan pulsar.ConsumerMessage, error) {
 	}
 
 	var dlqPolicy *pulsar.DLQPolicy
-	retryEnable := false
-	if retryDetails := gi.context.instanceConf.funcDetails.RetryDetails; retryDetails != nil {
+	retryDetails := gi.context.instanceConf.funcDetails.RetryDetails
+	if retryDetails != nil && retryDetails.DeadLetterTopic != "" {
 		dlqPolicy = &pulsar.DLQPolicy{
 			DeadLetterTopic: retryDetails.DeadLetterTopic,
 		}
 
 		if retryDetails.MaxMessageRetries >= 0 {
 			dlqPolicy.MaxDeliveries = uint32(retryDetails.MaxMessageRetries)
 		}
-		retryEnable = true
 	}
 
 	properties := getProperties(getDefaultSubscriptionName(
@@ -361,7 +360,6 @@ func (gi *goInstance) setupConsumer() (chan pulsar.ConsumerMessage, error) {
 					MessageChannel:              channel,
 					SubscriptionInitialPosition: subscriptionPosition,
 					DLQ:                         dlqPolicy,
-					RetryEnable:                 retryEnable,
 				})
 			} else {
 				consumer, err = gi.client.Subscribe(pulsar.ConsumerOptions{
@@ -373,7 +371,6 @@ func (gi *goInstance) setupConsumer() (chan pulsar.ConsumerMessage, error) {
 					MessageChannel:              channel,
 					SubscriptionInitialPosition: subscriptionPosition,
 					DLQ:                         dlqPolicy,
-					RetryEnable:                 retryEnable,
 				})
 			}
 		} else {
@@ -386,7 +383,6 @@ func (gi *goInstance) setupConsumer() (chan pulsar.ConsumerMessage, error) {
 					MessageChannel:              channel,
 					SubscriptionInitialPosition: subscriptionPosition,
 					DLQ:                         dlqPolicy,
-					RetryEnable:                 retryEnable,
 				})
 			} else {
 				consumer, err = gi.client.Subscribe(pulsar.ConsumerOptions{
@@ -397,7 +393,6 @@ func (gi *goInstance) setupConsumer() (chan pulsar.ConsumerMessage, error) {
 					MessageChannel:              channel,
 					SubscriptionInitialPosition: subscriptionPosition,
 					DLQ:                         dlqPolicy,
-					RetryEnable:                 retryEnable,
 				})
 
 			}