Make ShardIteratorType and PollingIntervalMs configurable, fix shard re-discovery

- Add ShardIteratorType config (default: LATEST) propagated from Aspire options
- Add PollingIntervalMs config (default: 1000) for tunable poll frequency
- Fix shard re-discovery to only replace exhausted iterators, preserving active ones
- Add key-value parsing for new config options

Author: normj

Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Processes/DynamoDBStreamsEventSource/DynamoDBStreamsEventSourceBackgroundService.cs

@@ -169,19 +169,26 @@ private async Task PollStream(string streamArn, CancellationToken stoppingToken)
             if (shardIterators.Any(s => s == null))
             {
                 var newShards = await GetShardIterators(streamArn, stoppingToken);
-                // Merge: keep active iterators, add any new ones
-                var activeIterators = shardIterators.Where(s => s != null).ToList();
-                foreach (var newIter in newShards)
+                // Replace only null entries with new iterators, keep active ones
+                for (int i = 0; i < shardIterators.Count; i++)
                 {
-                    if (newIter != null)
-                        activeIterators.Add(newIter);
+                    if (shardIterators[i] == null && i < newShards.Count)
+                    {
+                        shardIterators[i] = newShards[i];
+                    }
                 }
-                shardIterators = activeIterators;
+                // Append any additional new shards discovered
+                for (int i = shardIterators.Count; i < newShards.Count; i++)
+                {
+                    shardIterators.Add(newShards[i]);
+                }
+                // Remove remaining null entries (fully exhausted shards with no replacement)
+                shardIterators = shardIterators.Where(s => s != null).ToList();
             }
 
             if (!hasRecords)
             {
-                await Task.Delay(1000, stoppingToken);
+                await Task.Delay(_config.PollingIntervalMs, stoppingToken);
             }
         }
     }
@@ -223,7 +230,7 @@ private async Task PollStream(string streamArn, CancellationToken stoppingToken)
             {
                 StreamArn = streamArn,
                 ShardId = shard.ShardId,
-                ShardIteratorType = ShardIteratorType.LATEST
+                ShardIteratorType = new ShardIteratorType(_config.ShardIteratorType)
             }, stoppingToken);
 
             iterators.Add(iteratorResponse.ShardIterator);

Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Processes/DynamoDBStreamsEventSource/DynamoDBStreamsEventSourceBackgroundServiceConfig.cs

@@ -27,4 +27,14 @@ public class DynamoDBStreamsEventSourceBackgroundServiceConfig
     /// The DynamoDB table name to read streams from.
     /// </summary>
     public required string TableName { get; init; }
+
+    /// <summary>
+    /// The shard iterator type to use when reading from the stream.
+    /// </summary>
+    public required string ShardIteratorType { get; init; } = "LATEST";
+
+    /// <summary>
+    /// The polling interval in milliseconds between stream reads when no records are found.
+    /// </summary>
+    public required int PollingIntervalMs { get; init; } = DynamoDBStreamsEventSourceProcess.DefaultPollingIntervalMs;
 }

Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Processes/DynamoDBStreamsEventSource/DynamoDBStreamsEventSourceConfig.cs

@@ -39,4 +39,16 @@ internal class DynamoDBStreamsEventSourceConfig
     /// The DynamoDB table name to read streams from.
     /// </summary>
     public string? TableName { get; set; }
+
+    /// <summary>
+    /// The shard iterator type to use when reading from the stream.
+    /// Valid values: LATEST, TRIM_HORIZON. Default is LATEST.
+    /// </summary>
+    public string? ShardIteratorType { get; set; }
+
+    /// <summary>
+    /// The polling interval in milliseconds between stream reads when no records are found.
+    /// Default is 1000.
+    /// </summary>
+    public int? PollingIntervalMs { get; set; }
 }

Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Processes/DynamoDBStreamsEventSource/DynamoDBStreamsEventSourceProcess.cs

@@ -14,6 +14,7 @@ namespace Amazon.Lambda.TestTool.Processes.DynamoDBStreamsEventSource;
 public class DynamoDBStreamsEventSourceProcess
 {
     internal const int DefaultBatchSize = 100;
+    internal const int DefaultPollingIntervalMs = 1000;
 
     /// <summary>
     /// The Parent task for all the tasks started for each DynamoDB Streams event source.
@@ -76,7 +77,9 @@ public static DynamoDBStreamsEventSourceProcess Startup(RunCommandSettings setti
                 BatchSize = config.BatchSize ?? DefaultBatchSize,
                 FunctionName = config.FunctionName ?? LambdaRuntimeApi.DefaultFunctionName,
                 LambdaRuntimeApi = lambdaRuntimeApi,
-                TableName = tableName
+                TableName = tableName,
+                ShardIteratorType = config.ShardIteratorType ?? "LATEST",
+                PollingIntervalMs = config.PollingIntervalMs ?? DefaultPollingIntervalMs
             };
 
             builder.Services.AddSingleton(backgroundServiceConfig);
@@ -184,6 +187,16 @@ internal static List<DynamoDBStreamsEventSourceConfig> LoadDynamoDBStreamsEventS
                     case "tablename":
                         config.TableName = keyValuePair[1].Trim();
                         break;
+                    case "sharditeratortype":
+                        config.ShardIteratorType = keyValuePair[1].Trim();
+                        break;
+                    case "pollingintervalms":
+                        if (!int.TryParse(keyValuePair[1].Trim(), out var pollingInterval))
+                        {
+                            throw new InvalidOperationException("Value for polling interval is not a formatted integer");
+                        }
+                        config.PollingIntervalMs = pollingInterval;
+                        break;
                 }
             }