Cosmos DB: Add retry policy support for Cosmos DB Trigger

src/WebJobs.Extensions.CosmosDB/Trigger/CosmosDBTriggerBinding.cs

@@ -6,6 +6,7 @@
 using System.Reflection;
 using System.Threading.Tasks;
 using Microsoft.Azure.Cosmos;
+using Microsoft.Azure.WebJobs.Host;
 using Microsoft.Azure.WebJobs.Host.Bindings;
 using Microsoft.Azure.WebJobs.Host.Listeners;
 using Microsoft.Azure.WebJobs.Host.Protocols;
@@ -14,6 +15,7 @@
 
 namespace Microsoft.Azure.WebJobs.Extensions.CosmosDB
 {
+    [SupportsRetry]
     internal class CosmosDBTriggerBinding<T> : ITriggerBinding
     {
         private static readonly IReadOnlyDictionary<string, Type> _emptyBindingContract = new Dictionary<string, Type>();

test/WebJobs.Extensions.CosmosDB.Tests/CosmosDBEndToEndTests.cs

@@ -72,6 +72,38 @@ await TestHelpers.Await(() =>
             }
         }
 
+        [Fact]
+        public async Task CosmosDBEndToEnd_WithRetry()
+        {
+            using (var host = await StartHostAsync(typeof(EndToEndTestClass_Retry)))
+            {
+                var client = await InitializeDocumentClientAsync(host.Services.GetRequiredService<IConfiguration>());
+
+                // Call the outputs function directly, which will write out 3 documents 
+                // using with the 'input' property set to the value we provide.
+                var input = Guid.NewGuid().ToString();
+                var parameter = new Dictionary<string, object>();
+                parameter["input"] = input;
+
+                await host.GetJobHost().CallAsync(nameof(EndToEndTestClass_Retry.Outputs), parameter);
+
+                await TestHelpers.Await(() =>
+                {
+                    var logMessages = _loggerProvider.GetAllLogMessages();
+                    foreach (LogMessage logMsg in logMessages)
+                    {
+                        if (logMsg.Exception != null)
+                        {
+                            Console.WriteLine(logMsg.Exception.InnerException.Message);
+                        }
+                    }
+
+                    return logMessages.Count(p => p.FormattedMessage != null && p.FormattedMessage.Contains("Trigger called!")) == 6
+                        && logMessages.Count(p => p.Exception != null && p.Exception.InnerException.Message.Contains("Test exception") && !p.Category.StartsWith("Host.Results")) == 1;
+                });
+            }
+        }
+
         private async Task<CosmosClient> InitializeDocumentClientAsync(IConfiguration configuration)
         {
             var client = new CosmosClient(configuration.GetConnectionStringOrSetting(Constants.DefaultConnectionStringName).Value);
@@ -172,5 +204,39 @@ public static void TriggerWithString(
                 }
             }
         }
+
+        private static class EndToEndTestClass_Retry
+        {
+            private static bool shouldThrow = true;
+
+            [NoAutomaticTrigger]
+            public static async Task Outputs(
+                string input,
+                [CosmosDB(DatabaseName, CollectionName, CreateIfNotExists = true)] IAsyncCollector<object> collector,
+                ILogger log)
+            {
+                for (int i = 0; i < 3; i++)
+                {
+                    await collector.AddAsync(new { input = input, id = Guid.NewGuid().ToString() });
+                }
+            }
+
+            [FixedDelayRetry(5, "00:00:01")]
+            public static void Trigger(
+                [CosmosDBTrigger(DatabaseName, CollectionName, CreateLeaseContainerIfNotExists = true)] IReadOnlyList<Item> documents,
+                ILogger log)
+            {
+                foreach (var document in documents)
+                {
+                    log.LogInformation($"Trigger called!");
+                }
+
+                if (shouldThrow)
+                {
+                    shouldThrow = false;
+                    throw new Exception("Test exception");
+                }
+            }
+        }
     }
 }