fix: Ack/ModAck failures in non exactly once delivery flow should not be retried.

Author: Rishabh-V

apis/Google.Cloud.PubSub.V1/Google.Cloud.PubSub.V1.Tests/SubscriberClientTest.cs

@@ -423,15 +423,12 @@ public override void HandleAckResponses(IReadOnlyList<AckNackResponse> responses
                 // For exactly once delivery, only messages that succeed or fail permanently appear here, i.e., only messages whose status is finalized.
                 // The messages with temporary failures whose status may change in future appear here only when they succeed or fail permanently.
                 // In non exactly once delivery, we show status of every message as Success because every message acknowledgement is treated as fire and forget.
-                // If message failed with recoverable error, it will be re-queued and hence may appear again here with the same success status, so adding distinct items only.
                 Responses.Locked(() => Responses.AddRange(responses.Where(item => !Responses.Any(j => j.MessageId == item.MessageId))));
 
-
             public override void HandleNackResponses(IReadOnlyList<AckNackResponse> responses) =>
                 // For exactly once delivery, only messages that succeed or fail permanently appear here, i.e., only messages whose status is finalized.
                 // The messages with temporary failures whose status may change in future appear here only when they succeed or fail permanently.
                 // In non exactly once delivery, we show status of every message as Success because every message acknowledgement is treated as fire and forget.
-                // If message failed with recoverable error, it will be re-queued and hence may appear again here with the same success status, so adding distinct items only.
                 Responses.Locked(() => Responses.AddRange(responses.Where(item => !Responses.Any(j => j.MessageId == item.MessageId))));
         }
 

apis/Google.Cloud.PubSub.V1/Google.Cloud.PubSub.V1/SubscriberClient.cs

@@ -1727,39 +1727,12 @@ private void HandleAckResponse(Task writeTask, List<string> ackIds, List<string>
                 _pushInFlight -= (ackIds?.Count ?? 0) + (nackIds?.Count ?? 0) + (extendIds?.Count ?? 0);
                 if (writeTask.IsFaulted)
                 {
-                    if (writeTask.Exception.As<RpcException>()?.IsRecoverable() ?? false)
+                    // Check if it's an RpcException. If it is, then ignore it and continue. We may want to log it later. 
+                    // Other non-gRPC unrecoverable errors will continue to be thrown.
+                    if (writeTask.Exception.As<RpcException>() is null)
                     {
-                        // Recoverable write error, requeue data and continue.
-                        // ackIds and nackIds are never both set in the same call, so no need to share a lock.
-                        if (hasAckIds)
-                        {
-                            lock (_lock)
-                            {
-                                _ackQueue.Requeue(ackIds);
-                            }
-                        }
-                        if (hasNackIds)
-                        {
-                            lock (_lock)
-                            {
-                                _nackQueue.Requeue(nackIds);
-                            }
-                        }
-                        if (extendIds != null && extendIds.Count > 0)
-                        {
-                            _extendQueue.Requeue(extendIds);
-                        }
-                        // TODO: Backoff
-                    }
-                    else
-                    {
-                        // Check if it's an RpcException. If it is, then ignore it and continue. We may want to log it later. 
-                        // Other non-gRPC unrecoverable errors will continue to be thrown.
-                        if (writeTask.Exception.As<RpcException>() is null)
-                        {
-                            // It is a non-gRPC unrecoverable error; throw exception.
-                            throw writeTask.Exception.FlattenIfPossible();
-                        }
+                        // It is a non-gRPC unrecoverable error; throw exception.
+                        throw writeTask.Exception.FlattenIfPossible();
                     }
                 }
                 // Immediately send more data if there is any to send.