upgrade to Akka.NET v1.5.46 (#495)

Author: Aaronontheweb

.gitignore

@@ -206,3 +206,4 @@ build/
 .idea
 /.cursor
 BenchmarkDotNet.Artifacts/
+/.claude

Directory.Packages.props

@@ -2,10 +2,9 @@
   <PropertyGroup>
     <ManagePackageVersionsCentrally>true</ManagePackageVersionsCentrally>
     <!-- Nuget package versions -->
-    <AkkaVersion>1.5.42</AkkaVersion>
+    <AkkaVersion>1.5.46</AkkaVersion>
     <!-- Unit test Nuget package versions -->
     <NBenchVersion>1.2.2</NBenchVersion>
-    <TestSdkVersion>17.4.1</TestSdkVersion>
     <AspireVersion>9.3.0</AspireVersion>
   </PropertyGroup>
   <!-- App dependencies -->

examples/EventHub.Consumer/Program.cs

@@ -55,10 +55,10 @@ public static async Task Main(string[] args)
 
         var committerDefaults = CommitterSettings.Create(system);
 
-        // Comment for simple no-commit consumer
+        // Comment for committable consumer
         var control = KafkaConsumer.CommittableSource(consumerSettings, subscription)
             .SelectAsync(1, msg =>
-                Business(msg.Record).ContinueWith(done => (ICommittable)msg.CommitableOffset))
+                Business(msg.Record).ContinueWith(ICommittable (_) => msg.CommitableOffset))
             .ToMaterialized(
                 Committer.Sink(committerDefaults.WithMaxBatch(1)),
                 DrainingControl<NotUsed>.Create)

src/Akka.Streams.Kafka/Stages/Consumers/Abstract/SubSourceLogic.cs

@@ -8,6 +8,7 @@
 using System.Collections.Generic;
 using System.Collections.Immutable;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 using Akka.Actor;
 using Akka.Streams.Dsl;
@@ -140,6 +141,21 @@ private readonly Option<Func<IImmutableSet<TopicPartition>, Task<IImmutableSet<T
     /// </summary>
     private IImmutableSet<TopicPartition> _partitionsToRevoke = ImmutableHashSet<TopicPartition>.Empty;
 
+    /// <summary>
+    /// Cancellation token source for shutting down seek operations when stage is stopping
+    /// </summary>
+    private readonly CancellationTokenSource _shutdownTokenSource = new();
+    
+    /// <summary>
+    /// Track when the stage started to detect race conditions
+    /// </summary>
+    private readonly DateTime _startTime = DateTime.UtcNow;
+    
+    /// <summary>
+    /// Track if seek operation is in progress to detect race conditions
+    /// </summary>
+    private volatile bool _seekInProgress = false;
+
     protected StageActor SourceActor { get; private set; } = null!;
     public IActorRef ConsumerActor { get; private set; } = null!;
 
@@ -295,9 +311,15 @@ public override void PreStart()
 
     public override void PostStop()
     {
+        // Cancel any ongoing seek operations
+        _shutdownTokenSource.Cancel();
+        
         ConsumerActor.Tell(new KafkaConsumerActorMetadata.Internal.StopFromStage(_shape.ToString()), SourceActor.Ref);
 
         Control.OnShutdown();
+        
+        // Clean up cancellation token
+        _shutdownTokenSource.Dispose();
 
         base.PostStop();
     }
@@ -379,17 +401,58 @@ private void SeekAndEmitSubSources(IImmutableSet<TopicPartition> formerlyUnknown
 
         async Task AskToSeekOffsets()
         {
+            _seekInProgress = true;
+            
+            if (Log.IsDebugEnabled)
+                Log.Debug("#{0} Starting seek operation for partitions: {1}", _actorNumber, offsets.JoinToString(", "));
+                
             try
             {
                 await ConsumerActor.Ask(new KafkaConsumerActorMetadata.Internal.Seek(offsets),
-                    TimeSpan.FromSeconds(10));
+                    TimeSpan.FromSeconds(10), _shutdownTokenSource.Token);
+                    
+                if (Log.IsDebugEnabled)
+                    Log.Debug("#{0} Seek operation completed successfully", _actorNumber);
+                    
                 _updatePendingPartitionsAndEmitSubSourcesCallback(formerlyUnknown);
             }
-            catch (Exception)
+            catch (OperationCanceledException) when (_shutdownTokenSource.IsCancellationRequested)
             {
-                // only exceptions that can be thrown here are related to TCS cancellation / timeout
-                _stageFailCallback(new ConsumerFailed(
-                    $"{_actorNumber} Consumer failed during seek, Ask timed out. Partitions: {offsets.JoinToString(", ")}"));
+                // Stage is shutting down, this is expected - don't treat as fatal error
+                if (Log.IsDebugEnabled)
+                    Log.Debug("#{0} Seek operation cancelled due to stage shutdown (expected). Partitions: {1}", 
+                        _actorNumber, offsets.JoinToString(", "));
+            }
+            catch (Exception ex)
+            {
+                var timeSinceStart = DateTime.UtcNow.Subtract(_startTime);
+                
+                // Other exceptions (timeout, etc.) - check if we're shutting down before failing
+                if (_shutdownTokenSource.IsCancellationRequested)
+                {
+                    // Stage is shutting down, timeout is expected
+                    if (Log.IsDebugEnabled)
+                        Log.Debug("#{0} Seek operation timed out during stage shutdown (expected). Partitions: {1}", 
+                            _actorNumber, offsets.JoinToString(", "));
+                }
+                else
+                {
+                    // Check for race condition pattern - very fast failure
+                    if (timeSinceStart.TotalMilliseconds < 100)
+                    {
+                        Log.Warning("#{0} Seek operation failed very quickly ({1}ms after start) - possible race condition. " +
+                                   "Error: {2}, Partitions: {3}", 
+                            _actorNumber, timeSinceStart.TotalMilliseconds, ex.Message, offsets.JoinToString(", "));
+                    }
+                    
+                    // Unexpected timeout while stage is still active
+                    _stageFailCallback(new ConsumerFailed(
+                        $"{_actorNumber} Consumer failed during seek, Ask timed out. Partitions: {offsets.JoinToString(", ")}", ex));
+                }
+            }
+            finally
+            {
+                _seekInProgress = false;
             }
         }
     }
@@ -497,9 +560,28 @@ private void PerformStop()
 
     private void PerformShutdown()
     {
+        var timeSinceStart = DateTime.UtcNow.Subtract(_startTime);
+        
+        // Only log if this looks like the race condition (very fast shutdown with seek in progress)
+        if (timeSinceStart.TotalMilliseconds < 100 && _seekInProgress)
+        {
+            Log.Warning("#{0} Potential race condition detected: PerformShutdown called {1}ms after start " +
+                       "with seek operation still in progress. Pending: {2}, InStartup: {3}, SubSources: {4}",
+                _actorNumber, timeSinceStart.TotalMilliseconds, _pendingPartitions.Count, 
+                _partitionsInStartup.Count, _subSources.Count);
+        }
+        else if (Log.IsDebugEnabled)
+        {
+            Log.Debug("#{0} PerformShutdown called normally after {1}ms. SeekInProgress: {2}", 
+                _actorNumber, timeSinceStart.TotalMilliseconds, _seekInProgress);
+        }
+        
         Log.Info("Completing. Partitions [{0}], StageActor {1}", string.Join(", ", _subSources.Keys), SourceActor.Ref);
         SetKeepGoing(true);
 
+        // Cancel any ongoing seek operations immediately to avoid timeouts
+        _shutdownTokenSource.Cancel();
+
         // TODO from alpakka: we should wait for subsources to be shutdown and next shutdown main stage
         _subSources.Values.Select(c => c.ControlAndStageActor.Control).ForEach(control => control.Shutdown());