sequentially close the external stream thread

LGouellec · Oct 21, 2024 · d36208e · d36208e
1 parent 33fad96
commit d36208e
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 12 deletions.
diff --git a/core/Processors/ExternalStreamThread.cs b/core/Processors/ExternalStreamThread.cs
@@ -33,7 +33,6 @@ internal class ExternalStreamThread : IThread
         private IAdminClient adminClient;
         private readonly string logPrefix;
         private readonly Thread thread;
-        private CancellationToken token;
 
         private DateTime lastCommit = DateTime.Now;
         private DateTime lastMetrics = DateTime.Now;
@@ -93,7 +92,7 @@ public void Run()
             {
                 SetState(ThreadState.RUNNING);
 
-                while (!token.IsCancellationRequested)
+                while (State.IsRunning())
                 {
                     if (exception != null)
                     {
@@ -205,13 +204,11 @@ private void CompleteShutdown()
             {
                 if (!IsDisposable)
                 {
-
-                    log.LogInformation($"{logPrefix}Shutting down");
-
-                    SetState(ThreadState.PENDING_SHUTDOWN);
-
                     IsRunning = false;
-
+
+                    if (State != ThreadState.PENDING_SHUTDOWN)
+                        SetState(ThreadState.PENDING_SHUTDOWN);
+
                     CommitOffsets(true);
 
                     var consumer = GetConsumer();
@@ -308,6 +305,9 @@ private void CloseThread()
         {
             try
             {
+                log.LogInformation($"{logPrefix}Shutting down");
+                SetState(ThreadState.PENDING_SHUTDOWN);
+
                 thread.Join();
             }
             catch (Exception e)
@@ -326,8 +326,7 @@ public void Start(CancellationToken token)
                 IsRunning = false;
                 return;
             }
-
-            this.token = token;
+
             IsRunning = true;
 
             if(configuration.Guarantee == ProcessingGuarantee.EXACTLY_ONCE)

diff --git a/core/Processors/Internal/SequentiallyGracefullyShutdownHook.cs b/core/Processors/Internal/SequentiallyGracefullyShutdownHook.cs
@@ -31,9 +31,7 @@ public void Shutdown()
             _tokenSource.Cancel();
 
             foreach (var t in _streamThreads)
-            {
                 t.Dispose();
-            }
 
             _externalStreamThread?.Dispose();
             _globalStreamThread?.Dispose();