From feb4106fd2391bf1a4baaed939113417e4405e3a Mon Sep 17 00:00:00 2001 From: "dependabot[bot]" <49699333+dependabot[bot]@users.noreply.github.com> Date: Tue, 22 Sep 2026 15:31:34 +0000 Subject: [PATCH 01/10] Bump Microsoft.Orleans.Streaming from 10.2.0 to 10.3.1 --- updated-dependencies: - dependency-name: Microsoft.Orleans.Streaming dependency-version: 10.3.1 dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] --- Directory.Packages.props | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Directory.Packages.props b/Directory.Packages.props index ee86b45..bba43b5 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -16,7 +16,7 @@ - + From 99b3fc094112994b2f657243fef4ae2b4a771216 Mon Sep 17 00:00:00 2001 From: Christopher Hair Date: Tue, 22 Sep 2026 13:02:59 -0400 Subject: [PATCH 02/10] Bump Microsoft Orleans packages to version 10.3.1 and update Testcontainers.Kafka to 4.15.0; enhance null handling in deserialization and update method signatures for cancellation support in Kafka integration tests. --- Directory.Packages.props | 8 ++++---- src/Orleans.Streams.Confluent/KafkaBatchContainer.cs | 3 ++- src/Orleans.Streams.Confluent/KafkaQueueAdapter.cs | 2 +- .../KafkaStreamProviderIntegrationTests.cs | 6 +++--- 4 files changed, 10 insertions(+), 9 deletions(-) diff --git a/Directory.Packages.props b/Directory.Packages.props index bba43b5..04d6be1 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -12,13 +12,13 @@ - + - - + + - + \ No newline at end of file diff --git a/src/Orleans.Streams.Confluent/KafkaBatchContainer.cs b/src/Orleans.Streams.Confluent/KafkaBatchContainer.cs index 0fd9fea..38dedce 100644 --- a/src/Orleans.Streams.Confluent/KafkaBatchContainer.cs +++ b/src/Orleans.Streams.Confluent/KafkaBatchContainer.cs @@ -91,7 +91,8 @@ public static KafkaBatchContainer FromPayload(Serializer se ArgumentNullException.ThrowIfNull(serializer); ArgumentNullException.ThrowIfNull(payload); - return serializer.Deserialize(payload); + return serializer.Deserialize(payload) + ?? throw new InvalidOperationException("The Kafka batch payload deserialized to null."); } /// diff --git a/src/Orleans.Streams.Confluent/KafkaQueueAdapter.cs b/src/Orleans.Streams.Confluent/KafkaQueueAdapter.cs index 317e408..276b697 100644 --- a/src/Orleans.Streams.Confluent/KafkaQueueAdapter.cs +++ b/src/Orleans.Streams.Confluent/KafkaQueueAdapter.cs @@ -47,7 +47,7 @@ public IQueueAdapterReceiver CreateReceiver(QueueId queueId) return _receivers[queueId].Value; } - public async Task QueueMessageBatchAsync(StreamId streamId, IEnumerable events, StreamSequenceToken token, Dictionary requestContext) + public async Task QueueMessageBatchAsync(StreamId streamId, IEnumerable events, StreamSequenceToken? token, Dictionary? requestContext) { ArgumentNullException.ThrowIfNull(events); var effectiveRequestContext = requestContext ?? EmptyRequestContext; diff --git a/test/Orleans.Streams.Confluent.IntegrationTests/KafkaStreamProviderIntegrationTests.cs b/test/Orleans.Streams.Confluent.IntegrationTests/KafkaStreamProviderIntegrationTests.cs index 9b9f8c7..2c3c83d 100644 --- a/test/Orleans.Streams.Confluent.IntegrationTests/KafkaStreamProviderIntegrationTests.cs +++ b/test/Orleans.Streams.Confluent.IntegrationTests/KafkaStreamProviderIntegrationTests.cs @@ -156,7 +156,7 @@ await adapter.QueueMessageBatchAsync( batch.Partition.Should().Be((int)queueId.GetNumericId()); batch.GetEvents().Select(tuple => tuple.Item1).Should().ContainSingle().Which.Should().Be("created"); - await receiver.MessagesDeliveredAsync(messages); + await receiver.MessagesDeliveredAsync(messages, CancellationToken.None); await receiver.Shutdown(TimeSpan.FromSeconds(5)); } @@ -200,7 +200,7 @@ public async Task KafkaQueueAdapterReceiver_WhenShutdown_GetQueueMessagesAsync_R await receiver.Initialize(TimeSpan.FromSeconds(5)); await receiver.Shutdown(TimeSpan.FromSeconds(5)); - var messages = await receiver.GetQueueMessagesAsync(10); + var messages = await receiver.GetQueueMessagesAsync(10, CancellationToken.None); messages.Should().BeEmpty(); } @@ -282,7 +282,7 @@ private static async Task> WaitForReceiverMessagesAsync(I var deadline = DateTime.UtcNow.AddSeconds(15); while (DateTime.UtcNow < deadline) { - var messages = await receiver.GetQueueMessagesAsync(10); + var messages = await receiver.GetQueueMessagesAsync(10, CancellationToken.None); if (messages.Count > 0) { return messages; From 127cb9d6685c1c28ae085aefb4d8873100ebf70f Mon Sep 17 00:00:00 2001 From: Christopher Hair Date: Tue, 22 Sep 2026 13:13:17 -0400 Subject: [PATCH 03/10] Enhance cancellation support in GetQueueMessagesAsync and MessagesDeliveredAsync methods --- .../KafkaQueueAdapterReceiver.cs | 20 +++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/src/Orleans.Streams.Confluent/KafkaQueueAdapterReceiver.cs b/src/Orleans.Streams.Confluent/KafkaQueueAdapterReceiver.cs index 70f5c9f..36fa6b3 100644 --- a/src/Orleans.Streams.Confluent/KafkaQueueAdapterReceiver.cs +++ b/src/Orleans.Streams.Confluent/KafkaQueueAdapterReceiver.cs @@ -66,9 +66,14 @@ public Task Initialize(TimeSpan timeout) } public Task> GetQueueMessagesAsync(int maxCount) + => GetQueueMessagesAsync(maxCount, CancellationToken.None); + + public Task> GetQueueMessagesAsync(int maxCount, CancellationToken cancellationToken) { try { + cancellationToken.ThrowIfCancellationRequested(); + lock (_consumerSync) { var consumer = _consumer; @@ -81,6 +86,7 @@ public Task> GetQueueMessagesAsync(int maxCount) var batches = new List(maxCount); while (batches.Count < maxCount) { + cancellationToken.ThrowIfCancellationRequested(); var result = consumer.Consume(TimeSpan.FromMilliseconds(50)); if (result is null) { @@ -106,6 +112,10 @@ public Task> GetQueueMessagesAsync(int maxCount) return Task.FromResult>(batches); } } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + throw; + } catch (Exception ex) { LogErrorReceivingMessagesFailed(queueId, maxCount, ex); @@ -114,7 +124,12 @@ public Task> GetQueueMessagesAsync(int maxCount) } public Task MessagesDeliveredAsync(IList messages) + => MessagesDeliveredAsync(messages, CancellationToken.None); + + public Task MessagesDeliveredAsync(IList messages, CancellationToken cancellationToken) { + cancellationToken.ThrowIfCancellationRequested(); + if (messages.Count == 0) { return Task.CompletedTask; @@ -145,11 +160,16 @@ public Task MessagesDeliveredAsync(IList messages) return Task.CompletedTask; } + cancellationToken.ThrowIfCancellationRequested(); consumer.Commit(commitOffsets); LogDebugMessagesCommitted(queueId, messages.Count); return Task.CompletedTask; } } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + throw; + } catch (Exception ex) { LogErrorCommitFailed(queueId, messages.Count, ex); From 15e48518721c0d6d2ce45164ebada441887093d6 Mon Sep 17 00:00:00 2001 From: Christopher Hair Date: Tue, 22 Sep 2026 13:21:05 -0400 Subject: [PATCH 04/10] Add unit tests for cancellation handling in KafkaQueueAdapterReceiver --- .../KafkaQueueAdapterReceiverTests.cs | 43 +++++++++++++++++++ 1 file changed, 43 insertions(+) create mode 100644 test/Orleans.Streams.Confluent.Tests/KafkaQueueAdapterReceiverTests.cs diff --git a/test/Orleans.Streams.Confluent.Tests/KafkaQueueAdapterReceiverTests.cs b/test/Orleans.Streams.Confluent.Tests/KafkaQueueAdapterReceiverTests.cs new file mode 100644 index 0000000..71284f8 --- /dev/null +++ b/test/Orleans.Streams.Confluent.Tests/KafkaQueueAdapterReceiverTests.cs @@ -0,0 +1,43 @@ +using FluentAssertions; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.VisualStudio.TestTools.UnitTesting; +using Orleans.Streams; + +namespace Orleans.Streams.Confluent.Tests; + +[TestClass] +public sealed class KafkaQueueAdapterReceiverTests +{ + [TestMethod] + public async Task GetQueueMessagesAsync_WhenCancellationRequested_ThrowsOperationCanceledException() + { + using var cancellationTokenSource = new CancellationTokenSource(); + cancellationTokenSource.Cancel(); + var receiver = CreateReceiver(); + + Func act = async () => await receiver.GetQueueMessagesAsync(10, cancellationTokenSource.Token); + + await act.Should().ThrowAsync(); + } + + [TestMethod] + public async Task MessagesDeliveredAsync_WhenCancellationRequested_ThrowsOperationCanceledException() + { + using var cancellationTokenSource = new CancellationTokenSource(); + cancellationTokenSource.Cancel(); + var receiver = CreateReceiver(); + + Func act = async () => await receiver.MessagesDeliveredAsync([], cancellationTokenSource.Token); + + await act.Should().ThrowAsync(); + } + + private static KafkaQueueAdapterReceiver CreateReceiver() + => new( + "provider", + null!, + null!, + NullLogger.Instance, + default, + "consumer-group"); +} From 8396fdca6b9a922050246b5ffa0b2b9c6c1e56a4 Mon Sep 17 00:00:00 2001 From: Christopher Hair Date: Tue, 22 Sep 2026 13:33:07 -0400 Subject: [PATCH 05/10] Enhance cancellation support in GetQueueMessagesAsync and MessagesDeliveredAsync methods; add tests for cancellation handling in integration tests --- .../KafkaQueueAdapterReceiver.cs | 60 +++++++++++++++++-- .../KafkaStreamProviderIntegrationTests.cs | 11 +++- 2 files changed, 66 insertions(+), 5 deletions(-) diff --git a/src/Orleans.Streams.Confluent/KafkaQueueAdapterReceiver.cs b/src/Orleans.Streams.Confluent/KafkaQueueAdapterReceiver.cs index 36fa6b3..52e2861 100644 --- a/src/Orleans.Streams.Confluent/KafkaQueueAdapterReceiver.cs +++ b/src/Orleans.Streams.Confluent/KafkaQueueAdapterReceiver.cs @@ -70,13 +70,16 @@ public Task> GetQueueMessagesAsync(int maxCount) public Task> GetQueueMessagesAsync(int maxCount, CancellationToken cancellationToken) { + IConsumer? consumer = null; + var consumedOffsets = new List(); + try { cancellationToken.ThrowIfCancellationRequested(); lock (_consumerSync) { - var consumer = _consumer; + consumer = _consumer; if (consumer is null) { LogDebugReceiverNotInitialized(queueId); @@ -93,6 +96,15 @@ public Task> GetQueueMessagesAsync(int maxCount, Cancella break; } + if (!result.IsPartitionEOF) + { + consumedOffsets.Add(new TopicPartitionOffset( + new TopicPartition(result.Topic, result.Partition), + result.Offset)); + } + + cancellationToken.ThrowIfCancellationRequested(); + if (result.IsPartitionEOF || result.Message?.Value is null) { continue; @@ -114,6 +126,7 @@ public Task> GetQueueMessagesAsync(int maxCount, Cancella } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { + TrySeekToOffsets(consumer, consumedOffsets); throw; } catch (Exception ex) @@ -128,23 +141,32 @@ public Task MessagesDeliveredAsync(IList messages) public Task MessagesDeliveredAsync(IList messages, CancellationToken cancellationToken) { - cancellationToken.ThrowIfCancellationRequested(); - if (messages.Count == 0) { + cancellationToken.ThrowIfCancellationRequested(); return Task.CompletedTask; } + IConsumer? consumer = null; + var deliveredOffsets = messages + .OfType() + .Select(batch => new TopicPartitionOffset( + new TopicPartition(batch.Topic, new Partition(batch.Partition)), + new Offset(batch.Offset))) + .ToList(); + try { lock (_consumerSync) { - var consumer = _consumer; + consumer = _consumer; if (consumer is null) { return Task.CompletedTask; } + cancellationToken.ThrowIfCancellationRequested(); + var commitOffsets = messages .OfType() .GroupBy(batch => new TopicPartition(batch.Topic, new Partition(batch.Partition))) @@ -168,6 +190,7 @@ public Task MessagesDeliveredAsync(IList messages, Cancellation } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { + TrySeekToOffsets(consumer, deliveredOffsets); throw; } catch (Exception ex) @@ -177,6 +200,35 @@ public Task MessagesDeliveredAsync(IList messages, Cancellation } } + private void TrySeekToOffsets(IConsumer? consumer, IEnumerable offsets) + { + if (consumer is null) + { + return; + } + + lock (_consumerSync) + { + if (!ReferenceEquals(_consumer, consumer)) + { + return; + } + + try + { + foreach (var partitionOffsets in offsets.GroupBy(offset => offset.TopicPartition)) + { + var firstOffset = partitionOffsets.Min(offset => offset.Offset.Value); + consumer.Seek(new TopicPartitionOffset(partitionOffsets.Key, new Offset(firstOffset))); + } + } + catch (Exception exception) + { + _logger.LogError(exception, "Failed to restore Kafka receiver position after cancellation for queue {QueueId}", queueId); + } + } + } + public Task Shutdown(TimeSpan timeout) { IConsumer? consumer; diff --git a/test/Orleans.Streams.Confluent.IntegrationTests/KafkaStreamProviderIntegrationTests.cs b/test/Orleans.Streams.Confluent.IntegrationTests/KafkaStreamProviderIntegrationTests.cs index 2c3c83d..c3a6145 100644 --- a/test/Orleans.Streams.Confluent.IntegrationTests/KafkaStreamProviderIntegrationTests.cs +++ b/test/Orleans.Streams.Confluent.IntegrationTests/KafkaStreamProviderIntegrationTests.cs @@ -156,7 +156,16 @@ await adapter.QueueMessageBatchAsync( batch.Partition.Should().Be((int)queueId.GetNumericId()); batch.GetEvents().Select(tuple => tuple.Item1).Should().ContainSingle().Which.Should().Be("created"); - await receiver.MessagesDeliveredAsync(messages, CancellationToken.None); + using var cancellationTokenSource = new CancellationTokenSource(); + cancellationTokenSource.Cancel(); + + Func deliverCanceled = () => receiver.MessagesDeliveredAsync(messages, cancellationTokenSource.Token); + await deliverCanceled.Should().ThrowAsync(); + + var replayedMessages = await WaitForReceiverMessagesAsync(receiver); + replayedMessages.Should().ContainSingle(); + + await receiver.MessagesDeliveredAsync(replayedMessages, CancellationToken.None); await receiver.Shutdown(TimeSpan.FromSeconds(5)); } From 351ebf0bff4c0f2f08cb5554a9702a1823081927 Mon Sep 17 00:00:00 2001 From: Christopher Hair Date: Tue, 22 Sep 2026 13:46:48 -0400 Subject: [PATCH 06/10] Enhance cancellation support in KafkaQueueAdapterReceiver; add tests for cancellation handling in MessagesDeliveredAsync --- src/Orleans.Streams.Confluent/KafkaQueueAdapterReceiver.cs | 4 ++-- .../KafkaQueueAdapterReceiverTests.cs | 6 +++++- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/src/Orleans.Streams.Confluent/KafkaQueueAdapterReceiver.cs b/src/Orleans.Streams.Confluent/KafkaQueueAdapterReceiver.cs index 52e2861..424ce99 100644 --- a/src/Orleans.Streams.Confluent/KafkaQueueAdapterReceiver.cs +++ b/src/Orleans.Streams.Confluent/KafkaQueueAdapterReceiver.cs @@ -91,6 +91,7 @@ public Task> GetQueueMessagesAsync(int maxCount, Cancella { cancellationToken.ThrowIfCancellationRequested(); var result = consumer.Consume(TimeSpan.FromMilliseconds(50)); + cancellationToken.ThrowIfCancellationRequested(); if (result is null) { break; @@ -160,13 +161,12 @@ public Task MessagesDeliveredAsync(IList messages, Cancellation lock (_consumerSync) { consumer = _consumer; + cancellationToken.ThrowIfCancellationRequested(); if (consumer is null) { return Task.CompletedTask; } - cancellationToken.ThrowIfCancellationRequested(); - var commitOffsets = messages .OfType() .GroupBy(batch => new TopicPartition(batch.Topic, new Partition(batch.Partition))) diff --git a/test/Orleans.Streams.Confluent.Tests/KafkaQueueAdapterReceiverTests.cs b/test/Orleans.Streams.Confluent.Tests/KafkaQueueAdapterReceiverTests.cs index 71284f8..5b4c4a7 100644 --- a/test/Orleans.Streams.Confluent.Tests/KafkaQueueAdapterReceiverTests.cs +++ b/test/Orleans.Streams.Confluent.Tests/KafkaQueueAdapterReceiverTests.cs @@ -26,8 +26,12 @@ public async Task MessagesDeliveredAsync_WhenCancellationRequested_ThrowsOperati using var cancellationTokenSource = new CancellationTokenSource(); cancellationTokenSource.Cancel(); var receiver = CreateReceiver(); + var messages = new IBatchContainer[] + { + new KafkaBatchContainer(StreamId.Create("orders", "order-123"), ["created"], [], "orders-topic", 0, 0) + }; - Func act = async () => await receiver.MessagesDeliveredAsync([], cancellationTokenSource.Token); + Func act = async () => await receiver.MessagesDeliveredAsync(messages, cancellationTokenSource.Token); await act.Should().ThrowAsync(); } From ab1ac38a9a1f84cf75559aa07a4bab1803c96c35 Mon Sep 17 00:00:00 2001 From: Christopher Hair Date: Tue, 22 Sep 2026 14:05:47 -0400 Subject: [PATCH 07/10] Refactor offset restoration logic in KafkaQueueAdapterReceiver; improve error handling during cancellation --- .../KafkaQueueAdapterReceiver.cs | 36 ++++++++++++++++--- 1 file changed, 32 insertions(+), 4 deletions(-) diff --git a/src/Orleans.Streams.Confluent/KafkaQueueAdapterReceiver.cs b/src/Orleans.Streams.Confluent/KafkaQueueAdapterReceiver.cs index 424ce99..af83063 100644 --- a/src/Orleans.Streams.Confluent/KafkaQueueAdapterReceiver.cs +++ b/src/Orleans.Streams.Confluent/KafkaQueueAdapterReceiver.cs @@ -127,7 +127,7 @@ public Task> GetQueueMessagesAsync(int maxCount, Cancella } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { - TrySeekToOffsets(consumer, consumedOffsets); + RestoreOffsetsOrFail(consumer, consumedOffsets); throw; } catch (Exception ex) @@ -190,7 +190,7 @@ public Task MessagesDeliveredAsync(IList messages, Cancellation } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { - TrySeekToOffsets(consumer, deliveredOffsets); + RestoreOffsetsOrFail(consumer, deliveredOffsets); throw; } catch (Exception ex) @@ -200,13 +200,15 @@ public Task MessagesDeliveredAsync(IList messages, Cancellation } } - private void TrySeekToOffsets(IConsumer? consumer, IEnumerable offsets) + private void RestoreOffsetsOrFail(IConsumer? consumer, IEnumerable offsets) { if (consumer is null) { return; } + Exception? restorationFailure = null; + lock (_consumerSync) { if (!ReferenceEquals(_consumer, consumer)) @@ -224,9 +226,35 @@ private void TrySeekToOffsets(IConsumer? consumer, IEnumerable Date: Tue, 22 Sep 2026 14:31:35 -0400 Subject: [PATCH 08/10] Enhance cancellation handling in GetQueueMessagesAsync; add integration test for cancellation scenario --- .../KafkaQueueAdapterReceiver.cs | 140 ++++++++++-------- .../KafkaStreamProviderIntegrationTests.cs | 16 ++ 2 files changed, 92 insertions(+), 64 deletions(-) diff --git a/src/Orleans.Streams.Confluent/KafkaQueueAdapterReceiver.cs b/src/Orleans.Streams.Confluent/KafkaQueueAdapterReceiver.cs index af83063..a3883f1 100644 --- a/src/Orleans.Streams.Confluent/KafkaQueueAdapterReceiver.cs +++ b/src/Orleans.Streams.Confluent/KafkaQueueAdapterReceiver.cs @@ -86,48 +86,55 @@ public Task> GetQueueMessagesAsync(int maxCount, Cancella return Task.FromResult>([]); } - var batches = new List(maxCount); - while (batches.Count < maxCount) + try { - cancellationToken.ThrowIfCancellationRequested(); - var result = consumer.Consume(TimeSpan.FromMilliseconds(50)); - cancellationToken.ThrowIfCancellationRequested(); - if (result is null) - { - break; - } - - if (!result.IsPartitionEOF) + var batches = new List(maxCount); + while (batches.Count < maxCount) { - consumedOffsets.Add(new TopicPartitionOffset( - new TopicPartition(result.Topic, result.Partition), - result.Offset)); + cancellationToken.ThrowIfCancellationRequested(); + var result = consumer.Consume(TimeSpan.FromMilliseconds(50)); + if (result is null) + { + cancellationToken.ThrowIfCancellationRequested(); + break; + } + + if (!result.IsPartitionEOF) + { + consumedOffsets.Add(new TopicPartitionOffset( + new TopicPartition(result.Topic, result.Partition), + result.Offset)); + } + + cancellationToken.ThrowIfCancellationRequested(); + + if (result.IsPartitionEOF || result.Message?.Value is null) + { + continue; + } + + var container = KafkaBatchContainer + .FromPayload(serializer, result.Message.Value) + .WithKafkaMetadata(result.Topic, result.Partition.Value, result.Offset.Value); + batches.Add(container); } - cancellationToken.ThrowIfCancellationRequested(); - - if (result.IsPartitionEOF || result.Message?.Value is null) + if (batches.Count > 0) { - continue; + LogDebugMessagesReceived(queueId, batches.Count); } - var container = KafkaBatchContainer - .FromPayload(serializer, result.Message.Value) - .WithKafkaMetadata(result.Topic, result.Partition.Value, result.Offset.Value); - batches.Add(container); + return Task.FromResult>(batches); } - - if (batches.Count > 0) + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { - LogDebugMessagesReceived(queueId, batches.Count); + RestoreOffsetsOrFail(consumer, consumedOffsets); + throw; } - - return Task.FromResult>(batches); } } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { - RestoreOffsetsOrFail(consumer, consumedOffsets); throw; } catch (Exception ex) @@ -161,36 +168,44 @@ public Task MessagesDeliveredAsync(IList messages, Cancellation lock (_consumerSync) { consumer = _consumer; - cancellationToken.ThrowIfCancellationRequested(); - if (consumer is null) + try { - return Task.CompletedTask; - } + cancellationToken.ThrowIfCancellationRequested(); - var commitOffsets = messages - .OfType() - .GroupBy(batch => new TopicPartition(batch.Topic, new Partition(batch.Partition))) - .Select(group => + if (consumer is null) { - var nextOffset = group.Max(batch => batch.Offset) + 1; - return new TopicPartitionOffset(group.Key, new Offset(nextOffset)); - }) - .ToList(); + return Task.CompletedTask; + } - if (commitOffsets.Count == 0) - { + var commitOffsets = messages + .OfType() + .GroupBy(batch => new TopicPartition(batch.Topic, new Partition(batch.Partition))) + .Select(group => + { + var nextOffset = group.Max(batch => batch.Offset) + 1; + return new TopicPartitionOffset(group.Key, new Offset(nextOffset)); + }) + .ToList(); + + if (commitOffsets.Count == 0) + { + return Task.CompletedTask; + } + + cancellationToken.ThrowIfCancellationRequested(); + consumer.Commit(commitOffsets); + LogDebugMessagesCommitted(queueId, messages.Count); return Task.CompletedTask; } - - cancellationToken.ThrowIfCancellationRequested(); - consumer.Commit(commitOffsets); - LogDebugMessagesCommitted(queueId, messages.Count); - return Task.CompletedTask; + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + RestoreOffsetsOrFail(consumer, deliveredOffsets); + throw; + } } } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { - RestoreOffsetsOrFail(consumer, deliveredOffsets); throw; } catch (Exception ex) @@ -209,27 +224,24 @@ private void RestoreOffsetsOrFail(IConsumer? consumer, IEnumerab Exception? restorationFailure = null; - lock (_consumerSync) + if (!ReferenceEquals(_consumer, consumer)) { - if (!ReferenceEquals(_consumer, consumer)) - { - return; - } + return; + } - try - { - foreach (var partitionOffsets in offsets.GroupBy(offset => offset.TopicPartition)) - { - var firstOffset = partitionOffsets.Min(offset => offset.Offset.Value); - consumer.Seek(new TopicPartitionOffset(partitionOffsets.Key, new Offset(firstOffset))); - } - } - catch (Exception exception) + try + { + foreach (var partitionOffsets in offsets.GroupBy(offset => offset.TopicPartition)) { - _consumer = null; - restorationFailure = exception; + var firstOffset = partitionOffsets.Min(offset => offset.Offset.Value); + consumer.Seek(new TopicPartitionOffset(partitionOffsets.Key, new Offset(firstOffset))); } } + catch (Exception exception) + { + _consumer = null; + restorationFailure = exception; + } if (restorationFailure is null) { diff --git a/test/Orleans.Streams.Confluent.IntegrationTests/KafkaStreamProviderIntegrationTests.cs b/test/Orleans.Streams.Confluent.IntegrationTests/KafkaStreamProviderIntegrationTests.cs index c3a6145..210e2b7 100644 --- a/test/Orleans.Streams.Confluent.IntegrationTests/KafkaStreamProviderIntegrationTests.cs +++ b/test/Orleans.Streams.Confluent.IntegrationTests/KafkaStreamProviderIntegrationTests.cs @@ -148,6 +148,22 @@ await adapter.QueueMessageBatchAsync( new EventSequenceTokenV2(0), new Dictionary()); + using (var assertionConsumer = new ConsumerBuilder(new ConsumerConfig + { + BootstrapServers = bootstrapServers, + GroupId = $"assert-{Guid.NewGuid():N}", + AutoOffsetReset = AutoOffsetReset.Earliest, + AllowAutoCreateTopics = false + }).Build()) + { + assertionConsumer.Assign(new TopicPartition(topicName, new Partition((int)queueId.GetNumericId()))); + await WaitForKafkaMessageAsync(assertionConsumer); + } + + using var readCancellationTokenSource = new CancellationTokenSource(TimeSpan.FromMilliseconds(10)); + Func readCanceled = () => receiver.GetQueueMessagesAsync(2, readCancellationTokenSource.Token); + await readCanceled.Should().ThrowAsync(); + var messages = await WaitForReceiverMessagesAsync(receiver); messages.Should().ContainSingle(); From b2a315edfbce2fb3f7f4ecd846f5c8b1ae7f4537 Mon Sep 17 00:00:00 2001 From: Christopher Hair Date: Tue, 22 Sep 2026 14:48:05 -0400 Subject: [PATCH 09/10] Add cancellation handling instructions for Kafka receiver; update CI workflow to enforce cancellation tests --- .../kafka-stream-cancellation.instructions.md | 40 +++++++++++++++++++ .github/workflows/ci.yml | 5 ++- .../KafkaQueueAdapterReceiver.cs | 1 + 3 files changed, 45 insertions(+), 1 deletion(-) create mode 100644 .github/instructions/kafka-stream-cancellation.instructions.md diff --git a/.github/instructions/kafka-stream-cancellation.instructions.md b/.github/instructions/kafka-stream-cancellation.instructions.md new file mode 100644 index 0000000..03af626 --- /dev/null +++ b/.github/instructions/kafka-stream-cancellation.instructions.md @@ -0,0 +1,40 @@ +--- +description: 'Review and validate cancellation in the Kafka Orleans stream receiver' +applyTo: '**/KafkaQueueAdapterReceiver.cs,**/KafkaQueueAdapterReceiverTests.cs,**/KafkaStreamProviderIntegrationTests.cs' +--- + +# Kafka Receiver Cancellation + +When changing `KafkaQueueAdapterReceiver` or its tests, treat cancellation as an offset-safety concern, not only as an exception-handling concern. + +## Required Invariants + +- Check cancellation before polling, after every Kafka poll, and immediately before returning a batch. +- Record every consumed offset before checking cancellation so a canceled read can rewind the consumer position. +- Restore uncommitted offsets before allowing shutdown or disposal to invalidate the active consumer. +- Treat offset-restoration failure as fatal: invalidate and dispose the consumer, then require reinitialization. +- Check cancellation before returning from delivery, including when the receiver is uninitialized. +- Keep legacy receiver overloads as `CancellationToken.None` compatibility wrappers. + +## Required Tests + +Cover both already-canceled calls and cancellation after Kafka has advanced the local position. Verify that canceled reads and deliveries do not commit or lose messages, and that the next read replays the uncommitted batch. + +Prefer deterministic consumer seams or fakes for timing-sensitive cases. Do not rely on short `CancelAfter` windows as the only proof of post-consumption cancellation behavior. + +## Validation + +Run the focused receiver tests first: + +```powershell +dotnet test test/Orleans.Streams.Confluent.Tests/Orleans.Streams.Confluent.Tests.csproj --no-restore --filter FullyQualifiedName~KafkaQueueAdapterReceiverTests +``` + +Then run the solution build and complete test suite: + +```powershell +dotnet build Orleans.Streams.Confluent.slnx --no-restore +dotnet test Orleans.Streams.Confluent.slnx --no-build --no-restore +``` + +Review the final diff for cancellation/shutdown interleavings before committing. diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f807e2c..34078d3 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -27,7 +27,10 @@ jobs: run: dotnet restore Orleans.Streams.Confluent.slnx --nologo - name: Build - run: dotnet build Orleans.Streams.Confluent.slnx --configuration Release --no-restore --nologo + run: dotnet build Orleans.Streams.Confluent.slnx --configuration Release --no-restore --nologo --warnaserror + + - name: Test (Kafka receiver cancellation) + run: dotnet test test/Orleans.Streams.Confluent.Tests/Orleans.Streams.Confluent.Tests.csproj --configuration Release --no-build --nologo --filter FullyQualifiedName~KafkaQueueAdapterReceiverTests - name: Test (unit) run: dotnet test test/Orleans.Streams.Confluent.Tests/Orleans.Streams.Confluent.Tests.csproj --configuration Release --no-build --nologo diff --git a/src/Orleans.Streams.Confluent/KafkaQueueAdapterReceiver.cs b/src/Orleans.Streams.Confluent/KafkaQueueAdapterReceiver.cs index a3883f1..f708a51 100644 --- a/src/Orleans.Streams.Confluent/KafkaQueueAdapterReceiver.cs +++ b/src/Orleans.Streams.Confluent/KafkaQueueAdapterReceiver.cs @@ -124,6 +124,7 @@ public Task> GetQueueMessagesAsync(int maxCount, Cancella LogDebugMessagesReceived(queueId, batches.Count); } + cancellationToken.ThrowIfCancellationRequested(); return Task.FromResult>(batches); } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) From bfa8e52ac80519bb5b2b226f70d426016a5b9775 Mon Sep 17 00:00:00 2001 From: Christopher Hair Date: Tue, 22 Sep 2026 15:03:40 -0400 Subject: [PATCH 10/10] Add fault handling in KafkaQueueAdapterReceiver; implement _isFaulted state management --- .../KafkaQueueAdapterReceiver.cs | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/src/Orleans.Streams.Confluent/KafkaQueueAdapterReceiver.cs b/src/Orleans.Streams.Confluent/KafkaQueueAdapterReceiver.cs index f708a51..762be10 100644 --- a/src/Orleans.Streams.Confluent/KafkaQueueAdapterReceiver.cs +++ b/src/Orleans.Streams.Confluent/KafkaQueueAdapterReceiver.cs @@ -13,6 +13,7 @@ internal sealed partial class KafkaQueueAdapterReceiver(string providerName, Kaf private readonly ILogger _logger = logger; private readonly object _consumerSync = new(); private IConsumer? _consumer; + private bool _isFaulted; public Task Initialize(TimeSpan timeout) { @@ -31,6 +32,7 @@ public Task Initialize(TimeSpan timeout) var startingOffset = ResolveStartingOffset(consumer, topicPartition, timeout, consumerConfig.AutoOffsetReset); consumer.Assign(new TopicPartitionOffset(topicPartition, startingOffset)); _consumer = consumer; + _isFaulted = false; consumer = null; LogDebugReceiverAssigned(queueId, options.TopicName, queueNumericId); @@ -79,6 +81,7 @@ public Task> GetQueueMessagesAsync(int maxCount, Cancella lock (_consumerSync) { + ThrowIfFaulted(); consumer = _consumer; if (consumer is null) { @@ -168,6 +171,7 @@ public Task MessagesDeliveredAsync(IList messages, Cancellation { lock (_consumerSync) { + ThrowIfFaulted(); consumer = _consumer; try { @@ -241,6 +245,7 @@ private void RestoreOffsetsOrFail(IConsumer? consumer, IEnumerab catch (Exception exception) { _consumer = null; + _isFaulted = true; restorationFailure = exception; } @@ -270,6 +275,14 @@ private void RestoreOffsetsOrFail(IConsumer? consumer, IEnumerab throw new InvalidOperationException("Kafka receiver offset restoration failed after cancellation.", restorationFailure); } + private void ThrowIfFaulted() + { + if (_isFaulted) + { + throw new InvalidOperationException("Kafka receiver is faulted and must be reinitialized."); + } + } + public Task Shutdown(TimeSpan timeout) { IConsumer? consumer;