feat(streaming): add AWS Kinesis streaming provider

.github/workflows/ci.yml

@@ -22,6 +22,7 @@ env:
   CONSUL_IMAGE: public.ecr.aws/hashicorp/consul:1.19
   DYNAMODB_IMAGE: amazon/dynamodb-local:latest
   ELASTICMQ_IMAGE: softwaremill/elasticmq-native:latest
+  MINISTACK_IMAGE: ministackorg/ministack:1.4.11@sha256:8b05715b9effc23d998d4a4e0c0a0c4af7f9435f0ff076a30cf27a77326f14b1
   MSSQL_IMAGE: mcr.microsoft.com/mssql/server:latest
   NATS_IMAGE: nats:latest
   POSTGRES_IMAGE: postgres
@@ -916,6 +917,90 @@ jobs:
         path: |
           **/TestResults/*
           **/logs/*
+  test-kinesis:
+    name: AWS Kinesis provider tests
+    runs-on: ubuntu-latest
+    env:
+      BuildExternalAssets: "false"
+# [SuppressMessage("Microsoft.Security", "CS002:SecretInNextLine", Justification="Not a secret")]
+      AWS_ACCESS_KEY_ID: test
+# [SuppressMessage("Microsoft.Security", "CS002:SecretInNextLine", Justification="Not a secret")]
+      AWS_SECRET_ACCESS_KEY: test
+      AWS_REGION: us-east-1
+    strategy:
+      fail-fast: false
+      matrix:
+        provider: ["Kinesis"]
+        framework: ["net8.0", "net10.0"]
+    steps:
+    - uses: actions/checkout@93cb6efe18208431cddfb8368fd83d5badbf9bfd # v5
+    - name: Start Ministack
+      shell: bash
+      run: |
+        set -euo pipefail
+
+        bash .github/scripts/docker-pull-with-retry.sh "$MINISTACK_IMAGE"
+
+        docker run -d --pull=never --name ministack \
+          -p 4566:4566 \
+          -e MINISTACK_REGION="$AWS_REGION" \
+          "$MINISTACK_IMAGE"
+    - name: Start DynamoDB Local
+      shell: bash
+      run: |
+        set -euo pipefail
+
+        bash .github/scripts/docker-pull-with-retry.sh "$DYNAMODB_IMAGE"
+
+        docker run -d --pull=never --name dynamodb \
+          -p 8000:8000 \
+          -e AWS_ACCESS_KEY_ID="$AWS_ACCESS_KEY_ID" \
+          -e AWS_SECRET_ACCESS_KEY="$AWS_SECRET_ACCESS_KEY" \
+          -e AWS_REGION="$AWS_REGION" \
+          "$DYNAMODB_IMAGE"
+    - name: Wait for Ministack and DynamoDB Local
+      run: |
+        echo "Waiting for Ministack and DynamoDB Local to be ready..."
+        timeout 60 bash -c 'until curl --fail --silent http://localhost:4566/_ministack/health > /dev/null && nc -z localhost 8000; do sleep 1; done'
+        echo "Ministack and DynamoDB Local are ready"
+    - name: Setup .NET
+      uses: actions/setup-dotnet@c2fa09f4bde5ebb9d1777cf28262a3eb3db3ced7 # v5
+      with:
+        global-json-file: global.json
+    - name: Test
+      run: dotnet test
+        test/Extensions/Orleans.Streaming.Kinesis.Tests/Orleans.Streaming.Kinesis.Tests.csproj
+        --framework ${{ matrix.framework }}
+        --filter "Category=${{ matrix.provider }}"
+        --blame-hang-timeout 10m
+        --blame-crash-dump-type full
+        --blame-hang-dump-type full
+        --logger "trx;LogFileName=test_results_${{ matrix.provider }}_${{ matrix.framework }}.trx"
+        --
+        -parallel none -noshadow
+      env:
+        ORLEANSKINESISCONNECTIONSTRING: "http://localhost:4566;test;test;us-east-1"
+        ORLEANSDYNAMODBSERVICE: "http://localhost:8000"
+# [SuppressMessage("Microsoft.Security", "CS002:SecretInNextLine", Justification="Not a secret")]
+        ORLEANSDYNAMODBACCESSKEY: "test"
+# [SuppressMessage("Microsoft.Security", "CS002:SecretInNextLine", Justification="Not a secret")]
+        ORLEANSDYNAMODBSECRETKEY: "test"
+    - name: Clean up Ministack
+      if: always()
+      run: docker rm -f ministack || true
+    - name: Clean up DynamoDB Local
+      if: always()
+      run: docker rm -f dynamodb || true
+    - name: Archive Test Results
+      if: always()
+      continue-on-error: true
+      uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
+      with:
+        name: test_output_${{ github.job }}_${{ matrix.framework }}
+        retention-days: 1
+        path: |
+          **/TestResults/*
+          **/logs/*
   test:
     name: Test
     runs-on: ${{ matrix.os }}

Directory.Packages.props

@@ -14,6 +14,7 @@
     <PackageVersion Include="Autofac.Extensions.DependencyInjection" Version="10.0.0" />
     <PackageVersion Include="AwesomeAssertions" Version="9.3.0" />
     <PackageVersion Include="AWSSDK.DynamoDBv2" Version="4.0.14" />
+    <PackageVersion Include="AWSSDK.Kinesis" Version="4.0.2.12" />
     <PackageVersion Include="AWSSDK.SQS" Version="4.0.2.14" />
     <PackageVersion Include="Azure.Core" Version="1.50.0" />
     <PackageVersion Include="Azure.Data.Tables" Version="12.11.0" />

Orleans.slnx

@@ -81,6 +81,7 @@
     <Project Path="src/AWS/Orleans.Clustering.DynamoDB/Orleans.Clustering.DynamoDB.csproj" />
     <Project Path="src/AWS/Orleans.Persistence.DynamoDB/Orleans.Persistence.DynamoDB.csproj" />
     <Project Path="src/AWS/Orleans.Reminders.DynamoDB/Orleans.Reminders.DynamoDB.csproj" />
+    <Project Path="src/AWS/Orleans.Streaming.Kinesis/Orleans.Streaming.Kinesis.csproj" />
     <Project Path="src/AWS/Orleans.Streaming.SQS/Orleans.Streaming.SQS.csproj" />
     <Project Path="src/AWS/Orleans.Transactions.DynamoDB/Orleans.Transactions.DynamoDB.csproj" />
   </Folder>
@@ -157,6 +158,7 @@
   <Folder Name="/test/Extensions/">
     <Project Path="test/Extensions/Orleans.AWS.Tests/Orleans.AWS.Tests.csproj" />
     <Project Path="test/Extensions/Orleans.Clustering.Consul.Tests/Orleans.Clustering.Consul.Tests.csproj" />
+    <Project Path="test/Extensions/Orleans.Streaming.Kinesis.Tests/Orleans.Streaming.Kinesis.Tests.csproj" />
     <Project Path="test/Extensions/Orleans.Streaming.EventHubs.Tests/Orleans.Streaming.EventHubs.Tests.csproj" />
     <Project Path="test/Extensions/Orleans.Clustering.Cassandra.Tests/Orleans.Clustering.Cassandra.Tests.csproj" />
     <Project Path="test/Extensions/Orleans.Cosmos.Tests/Orleans.Cosmos.Tests.csproj" />

docs/site/src/content/docs/streaming/kinesis-streaming.md

@@ -0,0 +1,109 @@
+---
+title: Stream with Amazon Kinesis
+description: Configure Amazon Kinesis Data Streams for Orleans, including durable DynamoDB checkpoints.
+ms.date: 08/07/2026
+ms.topic: how-to
+---
+
+# Stream with Amazon Kinesis
+
+The [`Microsoft.Orleans.Streaming.Kinesis`](https://www.nuget.org/packages/Microsoft.Orleans.Streaming.Kinesis) package connects Orleans persistent streams to [Amazon Kinesis Data Streams](https://docs.aws.amazon.com/streams/latest/dev/introduction.html). Each Kinesis shard is an Orleans queue, so the number of open shards bounds physical read parallelism. Kinesis retention determines how far a consumer can replay.
+
+Create the Kinesis data stream before starting Orleans. The provider discovers its shards but doesn't create, delete, split, or merge the stream.
+
+## Configure the silo
+
+Install the package and register a named provider:
+
+```csharp
+using Orleans.Hosting;
+
+siloBuilder
+    .AddDynamoDBGrainStorage("PubSubStore", options =>
+    {
+        options.Service = "us-east-1";
+        options.ServiceId = "orders";
+        options.TableName = "OrdersPubSub";
+        options.UseProvisionedThroughput = false;
+    })
+    .AddKinesisStreams("Orders", stream =>
+    {
+        stream.ConfigureKinesis(options =>
+        {
+            options.StreamName = "orders";
+            options.Region = "us-east-1";
+        });
+
+        stream.UseDynamoDBCheckpointer(options =>
+        {
+            options.Service = "us-east-1";
+            options.TableName = "OrdersStreamCheckpoints";
+            options.PersistInterval = TimeSpan.FromSeconds(30);
+        });
+    });
+```
+
+`PubSubStore` persists explicit Orleans stream subscriptions. The checkpoint table has a different purpose: it records the last delivered Kinesis sequence number for each shard.
+
+Configure every Orleans client which publishes through the provider with the same provider name, stream name, and region:
+
+```csharp
+clientBuilder.AddKinesisStreams("Orders", options =>
+{
+    options.StreamName = "orders";
+    options.Region = "us-east-1";
+});
+```
+
+When explicit credentials aren't configured, the provider uses the [AWS SDK for .NET credential resolution chain](https://docs.aws.amazon.com/sdk-for-net/v4/developer-guide/creds-assign.html). In production, prefer workload credentials such as an IAM role. Set <xref:Orleans.Streaming.Kinesis.KinesisStreamOptions.Service> when using a custom Kinesis-compatible endpoint.
+
+## Choose checkpoint storage
+
+Kinesis shard iterators are temporary. Orleans therefore stores the last delivered sequence number outside Kinesis and uses it to resume after a restart or queue reassignment.
+
+### DynamoDB table checkpoints
+
+Call <xref:Orleans.Hosting.SiloKinesisStreamConfigurator.UseDynamoDBCheckpointer*> to store checkpoints directly in DynamoDB. The checkpointer:
+
+- Uses one versioned item per Orleans service, provider, and Kinesis shard.
+- Uses consistent reads and conditional writes to prevent a previous queue owner from overwriting a newer checkpoint.
+- Creates an on-demand table by default. Set <xref:Orleans.Configuration.DynamoDBStreamQueueCheckpointerOptions.CreateIfNotExists> to `false` when infrastructure provisioning owns the table.
+- Limits writes using <xref:Orleans.Configuration.DynamoDBStreamQueueCheckpointerOptions.PersistInterval>. A shorter interval reduces replay after failure but increases DynamoDB write traffic.
+
+Set <xref:Orleans.Configuration.DynamoDBStreamQueueCheckpointerOptions.UseProvisionedThroughput>, <xref:Orleans.Configuration.DynamoDBStreamQueueCheckpointerOptions.ReadCapacityUnits>, and <xref:Orleans.Configuration.DynamoDBStreamQueueCheckpointerOptions.WriteCapacityUnits> when the table uses provisioned capacity.
+
+### Grain checkpoints
+
+If no checkpointer is selected, the provider uses Orleans grain-backed checkpoints. Checkpoint grains use `PubSubStore` by default, so that provider must be durable in production:
+
+```csharp
+using Orleans.Streams;
+
+siloBuilder.AddKinesisStreams("Orders", stream =>
+{
+    stream.ConfigureKinesis(options =>
+    {
+        options.StreamName = "orders";
+        options.Region = "us-east-1";
+    });
+
+    stream.UseGrainCheckpointer(options =>
+    {
+        options.StorageProviderName = "PubSubStore";
+        options.CheckpointComparer = StreamCheckpointComparers.Numeric;
+        options.PersistInterval = TimeSpan.FromSeconds(30);
+    });
+});
+```
+
+Both implementations preserve monotonic Kinesis sequence numbers and can replay a small number of already delivered records after an unclean shutdown. Consumers must tolerate duplicate delivery.
+
+## Operations and permissions
+
+Grant the application only the Kinesis data-plane and DynamoDB table permissions required by its configuration. The Kinesis provider lists shards, obtains shard iterators, reads records, and writes records. A provider-managed checkpoint table also requires permissions to describe and create the table and to read and conditionally write checkpoint items.
+
+Monitor Kinesis iterator age, read throttling, provisioned throughput, and retention together with the [Orleans streaming metrics](streaming-operations.md#observe-health). <xref:Orleans.Streaming.Kinesis.KinesisStreamOptions.GetRecordsInterval> defaults to the fastest interval allowed by Kinesis for each shard.
+
+Live resharding isn't supported. If the shard topology changes while the provider is running, receivers stop rather than risk incorrect queue ownership. Restart the Orleans stream provider after splitting or merging shards.
+
+For a complete configuration which uses DynamoDB for clustering, grain state, reminders, and Kinesis checkpoints, see the [AWS Kinesis and DynamoDB sample](https://github.com/dotnet/orleans/tree/main/samples/AWS/KinesisDynamoDB).

docs/site/src/content/docs/streaming/stream-providers.md

@@ -16,6 +16,7 @@ A stream provider connects the Orleans streaming API to a transport and defines
 | Memory | [`Microsoft.Orleans.Streaming`](https://www.nuget.org/packages/Microsoft.Orleans.Streaming) | Stable | No; silo memory only | Yes, within the transient in-memory cache | None |
 | Azure Queue Storage | [`Microsoft.Orleans.Streaming.AzureStorage`](https://www.nuget.org/packages/Microsoft.Orleans.Streaming.AzureStorage) | Stable | Yes, in Azure Storage queues | No | Azure Storage account or Azurite; credentials and a stable Orleans service ID |
 | Azure Event Hubs | [`Microsoft.Orleans.Streaming.EventHubs`](https://www.nuget.org/packages/Microsoft.Orleans.Streaming.EventHubs) | Stable | Yes, within Event Hubs retention | Yes | Event Hubs namespace, hub, consumer group, and checkpoint storage |
+| Amazon Kinesis | [`Microsoft.Orleans.Streaming.Kinesis`](https://www.nuget.org/packages/Microsoft.Orleans.Streaming.Kinesis) | Stable | Yes, within Kinesis retention | Yes | Kinesis data stream, AWS credentials, region, and durable checkpoint storage |
 | Amazon SQS | [`Microsoft.Orleans.Streaming.SQS`](https://www.nuget.org/packages/Microsoft.Orleans.Streaming.SQS) | Stable | Yes, within SQS retention | No | AWS account, queue permissions, region/endpoint configuration |
 | ADO.NET | [`Microsoft.Orleans.Streaming.AdoNet`](https://www.nuget.org/packages/Microsoft.Orleans.Streaming.AdoNet) | **Alpha** | Yes, in relational tables until expiry/dead-letter eviction | No | Supported database, ADO.NET driver, and Orleans streaming SQL schema |
 | NATS JetStream | [`Microsoft.Orleans.Streaming.NATS`](https://www.nuget.org/packages/Microsoft.Orleans.Streaming.NATS) | **Alpha** | Configurable; file storage is the default | No | NATS server with JetStream and sufficient storage; subject/stream administration |
@@ -53,6 +54,10 @@ The examples use durable Azure Table Storage for `PubSubStore`; queue durability
 
 Register [Azure Event Hubs](https://learn.microsoft.com/azure/event-hubs/event-hubs-about) with <xref:Orleans.Hosting.SiloBuilderExtensions.AddEventHubStreams*>. Event Hubs retention and partition positions make this provider rewindable. Configure a consumer group dedicated to the Orleans application and durable checkpoint storage. Partition count bounds physical read parallelism, and retention bounds how far recovery can rewind.
 
+## Amazon Kinesis
+
+Register [Amazon Kinesis Data Streams](https://docs.aws.amazon.com/streams/latest/dev/introduction.html) with <xref:Orleans.Hosting.SiloBuilderExtensions.AddKinesisStreams*>. Kinesis retains events independently of Orleans, and the provider persists each shard's last delivered sequence number so that delivery can resume after shutdown or queue reassignment. See [Stream with Amazon Kinesis](kinesis-streaming.md) for configuration, checkpoint choices, and operational constraints.
+
 ## Amazon SQS
 
 Register [Amazon SQS](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/welcome.html) with <xref:Orleans.Hosting.SiloBuilderExtensions.AddSqsStreams*>. Standard queues use [at-least-once delivery](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/standard-queues-at-least-once-delivery.html), and SQS redelivers after the [visibility timeout](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-visibility-timeout.html) when processing isn't acknowledged. The Orleans provider isn't rewindable. Configure credentials using the deployment environment's AWS credential chain or protected connection configuration, and monitor queue age, redelivery, and dead-letter policy.

docs/site/src/content/docs/toc.yml

@@ -128,6 +128,8 @@ items:
             href: streaming/streams-programming-apis.md
           - name: Stream providers
             href: streaming/stream-providers.md
+          - name: Amazon Kinesis
+            href: streaming/kinesis-streaming.md
           - name: Delivery semantics
             href: streaming/delivery-semantics.md
           - name: Pub-sub storage

samples/AWS/KinesisDynamoDB/KinesisDynamoDB.csproj

@@ -0,0 +1,17 @@
+<Project Sdk="Microsoft.NET.Sdk">
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>net10.0</TargetFramework>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <Nullable>enable</Nullable>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="$(SourceRoot)src\Orleans.Server\Orleans.Server.csproj" />
+    <ProjectReference Include="$(SourceRoot)src\AWS\Orleans.Clustering.DynamoDB\Orleans.Clustering.DynamoDB.csproj" />
+    <ProjectReference Include="$(SourceRoot)src\AWS\Orleans.Persistence.DynamoDB\Orleans.Persistence.DynamoDB.csproj" />
+    <ProjectReference Include="$(SourceRoot)src\AWS\Orleans.Reminders.DynamoDB\Orleans.Reminders.DynamoDB.csproj" />
+    <ProjectReference Include="$(SourceRoot)src\AWS\Orleans.Streaming.Kinesis\Orleans.Streaming.Kinesis.csproj" />
+    <PackageReference Include="Microsoft.Extensions.Hosting" />
+  </ItemGroup>
+</Project>