resource/aws_sns_topic_subscription: Use paginated ListSubscriptionsB…

…yTopic and return immediately on errors

Reference: #13409

Output from acceptance testing:

```
--- PASS: TestAccAWSSNSTopicSubscription_basic (13.47s)
--- PASS: TestAccAWSSNSTopicSubscription_rawMessageDelivery (27.13s)
--- PASS: TestAccAWSSNSTopicSubscription_filterPolicy (28.12s)
--- PASS: TestAccAWSSNSTopicSubscription_deliveryPolicy (28.38s)
--- PASS: TestAccAWSSNSTopicSubscription_autoConfirmingEndpoint (48.31s)
--- PASS: TestAccAWSSNSTopicSubscription_autoConfirmingSecuredEndpoint (91.18s)
```

aws/resource_aws_sns_topic_subscription.go

@@ -235,18 +235,16 @@ func subscribeToSNSTopic(d *schema.ResourceData, snsconn *sns.SNS) (output *sns.
 
 			subscription, err := findSubscriptionByNonID(d, snsconn)
 
-			if subscription != nil {
-				output.SubscriptionArn = subscription.SubscriptionArn
-				return nil
+			if err != nil {
+				return resource.NonRetryableError(err)
 			}
 
-			if err != nil {
-				return resource.RetryableError(
-					fmt.Errorf("Error fetching subscriptions for SNS topic %s: %s", topic_arn, err))
+			if subscription == nil {
+				return resource.RetryableError(fmt.Errorf("Endpoint (%s) did not autoconfirm the subscription for topic %s", endpoint, topic_arn))
 			}
 
-			return resource.RetryableError(
-				fmt.Errorf("Endpoint (%s) did not autoconfirm the subscription for topic %s", endpoint, topic_arn))
+			output.SubscriptionArn = subscription.SubscriptionArn
+			return nil
 		})
 
 		if isResourceTimeoutError(err) {
@@ -268,37 +266,52 @@ func subscribeToSNSTopic(d *schema.ResourceData, snsconn *sns.SNS) (output *sns.
 }
 
 // finds a subscription using protocol, endpoint and topic_arn (which is a key in sns subscription)
-func findSubscriptionByNonID(d *schema.ResourceData, snsconn *sns.SNS) (*sns.Subscription, error) {
+func findSubscriptionByNonID(d *schema.ResourceData, conn *sns.SNS) (*sns.Subscription, error) {
 	protocol := d.Get("protocol").(string)
 	endpoint := d.Get("endpoint").(string)
-	topic_arn := d.Get("topic_arn").(string)
+	topicARN := d.Get("topic_arn").(string)
 	obfuscatedEndpoint := obfuscateEndpoint(endpoint)
 
-	req := &sns.ListSubscriptionsByTopicInput{
-		TopicArn: aws.String(topic_arn),
+	input := &sns.ListSubscriptionsByTopicInput{
+		TopicArn: aws.String(topicARN),
 	}
+	var result *sns.Subscription
 
-	for {
-		res, err := snsconn.ListSubscriptionsByTopic(req)
-
-		if err != nil {
-			return nil, fmt.Errorf("Error fetching subscriptions for topic %s : %s", topic_arn, err)
+	err := conn.ListSubscriptionsByTopicPages(input, func(page *sns.ListSubscriptionsByTopicOutput, lastPage bool) bool {
+		if page == nil {
+			return !lastPage
 		}
 
-		for _, subscription := range res.Subscriptions {
-			log.Printf("[DEBUG] check subscription with Subscription EndPoint %s (local: %s), Protocol %s, topicARN %s and SubscriptionARN %s", *subscription.Endpoint, obfuscatedEndpoint, *subscription.Protocol, *subscription.TopicArn, *subscription.SubscriptionArn)
-			if *subscription.Endpoint == obfuscatedEndpoint && *subscription.Protocol == protocol && *subscription.TopicArn == topic_arn && !subscriptionHasPendingConfirmation(subscription.SubscriptionArn) {
-				return subscription, nil
+		for _, subscription := range page.Subscriptions {
+			if subscription == nil {
+				continue
+			}
+
+			if aws.StringValue(subscription.Endpoint) != obfuscatedEndpoint {
+				continue
 			}
-		}
 
-		// if there are more than 100 subscriptions then go to the next 100 otherwise return an error
-		if res.NextToken != nil {
-			req.NextToken = res.NextToken
-		} else {
-			return nil, fmt.Errorf("Error finding subscription for topic %s with endpoint %s and protocol %s", topic_arn, endpoint, protocol)
+			if aws.StringValue(subscription.Protocol) != protocol {
+				continue
+			}
+
+			if aws.StringValue(subscription.TopicArn) != topicARN {
+				continue
+			}
+
+			if subscriptionHasPendingConfirmation(subscription.SubscriptionArn) {
+				continue
+			}
+
+			result = subscription
+
+			return false
 		}
-	}
+
+		return !lastPage
+	})
+
+	return result, err
 }
 
 // returns true if arn is nil or has both pending and confirmation words in the arn