Merge pull request #6 from skbkontur/avk/leaderElectionFix

skozulin · web-flow · commit 5abb499d133b · 2021-03-05T13:42:29.000+05:00
Avk/leader election fix
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,4 +1,6 @@
 # Changelog
+## v1.2.1 - 2021.03.05
+- Add cancelation token to the RtqPeriodicJobRunner.
 
 ## v1.1.11 - 2021.02.25
 - Update `@skbkontur/react-ui` package
diff --git a/Cassandra.DistributedTaskQueue.Monitoring/Cassandra.DistributedTaskQueue.Monitoring.csproj b/Cassandra.DistributedTaskQueue.Monitoring/Cassandra.DistributedTaskQueue.Monitoring.csproj
@@ -11,7 +11,7 @@
     <PackageReference Include="Elasticsearch.Net" Version="6.6.0" />
     <PackageReference Include="Microsoft.CSharp" Version="4.7.0" />
     <PackageReference Include="Newtonsoft.Json" Version="12.0.3" />
-    <PackageReference Include="SkbKontur.EventFeeds" Version="1.0.9" />
+    <PackageReference Include="SkbKontur.EventFeeds" Version="1.1.1" />
   </ItemGroup>
 
   <ItemGroup>
diff --git a/Cassandra.DistributedTaskQueue.Monitoring/EventFeed/RtqEventFeedPeriodicJobRunner.cs b/Cassandra.DistributedTaskQueue.Monitoring/EventFeed/RtqEventFeedPeriodicJobRunner.cs
@@ -1,6 +1,7 @@
-﻿using System;
+using System;
 using System.Linq;
 using System.Net;
+using System.Threading;
 
 using JetBrains.Annotations;
 
@@ -24,9 +25,10 @@ public RtqEventFeedPeriodicJobRunner(IRtqPeriodicJobRunner rtqPeriodicJobRunner,
 
         public void RunPeriodicJobWithLeaderElection([NotNull] string jobName,
                                                      TimeSpan delayBetweenIterations,
-                                                     [NotNull] Action jobAction,
+                                                     [NotNull] Action<CancellationToken> jobAction,
                                                      [NotNull] Func<IRunningEventFeed> onTakeTheLead,
-                                                     [NotNull] Func<IRunningEventFeed> onLoseTheLead)
+                                                     [NotNull] Func<IRunningEventFeed> onLoseTheLead,
+                                                     CancellationToken cancellationToken)
         {
             rtqPeriodicJobRunner.RunPeriodicJobWithLeaderElection(jobName,
                                                                   delayBetweenIterations,
@@ -44,7 +46,8 @@ public void RunPeriodicJobWithLeaderElection([NotNull] string jobName,
                                                                           var runningEventFeed = onLoseTheLead();
                                                                           var lagReportingJobId = FormatLagReportingJobId(runningEventFeed.FeedKey);
                                                                           rtqPeriodicJobRunner.StopPeriodicJob(lagReportingJobId);
-                                                                      });
+                                                                      },
+                                                                  cancellationToken);
         }
 
         public void StopPeriodicJobWithLeaderElection([NotNull] string jobName)
diff --git a/Cassandra.DistributedTaskQueue.Monitoring/Indexer/IRtqMonitoringEventFeeder.cs b/Cassandra.DistributedTaskQueue.Monitoring/Indexer/IRtqMonitoringEventFeeder.cs
@@ -1,4 +1,6 @@
-﻿using JetBrains.Annotations;
+using System.Threading;
+
+using JetBrains.Annotations;
 
 using SkbKontur.EventFeeds;
 
@@ -8,6 +10,6 @@ namespace SkbKontur.Cassandra.DistributedTaskQueue.Monitoring.Indexer
     public interface IRtqMonitoringEventFeeder
     {
         [NotNull]
-        IEventFeedsRunner RunEventFeeding();
+        IEventFeedsRunner RunEventFeeding(CancellationToken cancellationToken);
     }
 }
diff --git a/Cassandra.DistributedTaskQueue.Monitoring/Indexer/RtqMonitoringEventFeeder.cs b/Cassandra.DistributedTaskQueue.Monitoring/Indexer/RtqMonitoringEventFeeder.cs
@@ -1,4 +1,5 @@
-﻿using System;
+using System;
+using System.Threading;
 
 using JetBrains.Annotations;
 
@@ -41,7 +42,7 @@ public RtqMonitoringEventFeeder(ILog logger,
         public IGlobalTime GlobalTime { get; }
 
         [NotNull]
-        public IEventFeedsRunner RunEventFeeding()
+        public IEventFeedsRunner RunEventFeeding(CancellationToken cancellationToken)
         {
             return eventFeedFactory
                    .WithOffsetType<string>()
@@ -50,7 +51,7 @@ public IEventFeedsRunner RunEventFeeding()
                                                .WithBlade($"{indexerSettings.EventFeedKey}_Blade1", delay : TimeSpan.FromMinutes(15)))
                    .WithOffsetInterpreter(offsetInterpreter)
                    .WithOffsetStorageFactory(bladeId => new RtqElasticsearchOffsetStorage(elasticsearchClient, offsetInterpreter, bladeId.BladeKey))
-                   .RunFeeds(delayBetweenIterations : TimeSpan.FromMinutes(1));
+                   .RunFeeds(delayBetweenIterations : TimeSpan.FromMinutes(1), cancellationToken);
         }
 
         private readonly ILog logger;
diff --git a/Cassandra.DistributedTaskQueue.Monitoring/TaskCounter/IRtqTaskCounterEventFeeder.cs b/Cassandra.DistributedTaskQueue.Monitoring/TaskCounter/IRtqTaskCounterEventFeeder.cs
@@ -1,4 +1,6 @@
-﻿using JetBrains.Annotations;
+using System.Threading;
+
+using JetBrains.Annotations;
 
 using SkbKontur.EventFeeds;
 
@@ -7,6 +9,6 @@ namespace SkbKontur.Cassandra.DistributedTaskQueue.Monitoring.TaskCounter
     [PublicAPI]
     public interface IRtqTaskCounterEventFeeder
     {
-        ( /*[NotNull]*/ IEventFeedsRunner, /*[NotNull]*/ RtqTaskCounterStateManager) RunEventFeeding();
+        ( /*[NotNull]*/ IEventFeedsRunner, /*[NotNull]*/ RtqTaskCounterStateManager) RunEventFeeding(CancellationToken cancellationToken);
     }
 }
diff --git a/Cassandra.DistributedTaskQueue.Monitoring/TaskCounter/RtqTaskCounterEventFeeder.cs b/Cassandra.DistributedTaskQueue.Monitoring/TaskCounter/RtqTaskCounterEventFeeder.cs
@@ -1,4 +1,6 @@
-﻿using GroBuf;
+using System.Threading;
+
+using GroBuf;
 
 using JetBrains.Annotations;
 
@@ -45,7 +47,7 @@ public RtqTaskCounterEventFeeder(ILog logger,
         [NotNull]
         public IGlobalTime GlobalTime { get; }
 
-        public ( /*[NotNull]*/ IEventFeedsRunner, /*[NotNull]*/ RtqTaskCounterStateManager) RunEventFeeding()
+        public ( /*[NotNull]*/ IEventFeedsRunner, /*[NotNull]*/ RtqTaskCounterStateManager) RunEventFeeding(CancellationToken cancellationToken)
         {
             var stateManager = new RtqTaskCounterStateManager(logger, serializer, taskDataRegistry, stateStorage, settings, offsetInterpreter, perfGraphiteReporter);
             var eventConsumer = new RtqTaskCounterEventConsumer(stateManager, handleTasksMetaStorage, perfGraphiteReporter);
@@ -58,7 +60,7 @@ public RtqTaskCounterEventFeeder(ILog logger,
                                    .WithOffsetInterpreter(offsetInterpreter)
                                    .WithOffsetStorageFactory(bladeId => stateManager.CreateOffsetStorage(bladeId))
                                    .WithSingleLeaderElectionKey(stateManager.CompositeFeedKey)
-                                   .RunFeeds(settings.DelayBetweenEventFeedingIterations);
+                                   .RunFeeds(settings.DelayBetweenEventFeedingIterations, cancellationToken);
             return (eventFeedsRunner, stateManager);
         }
 
diff --git a/Cassandra.DistributedTaskQueue/Cassandra.DistributedTaskQueue.csproj b/Cassandra.DistributedTaskQueue/Cassandra.DistributedTaskQueue.csproj
@@ -9,7 +9,7 @@
 
   <ItemGroup>
     <PackageReference Include="GroBuf" Version="1.5.5" />
-    <PackageReference Include="SkbKontur.Cassandra.DistributedLock" Version="2.2.8" />
+    <PackageReference Include="SkbKontur.Cassandra.DistributedLock" Version="2.2.12" />
     <PackageReference Include="SkbKontur.Cassandra.GlobalTimestamp" Version="1.0.8" />
     <PackageReference Include="SkbKontur.Cassandra.ThriftClient" Version="2.4.23" />
     <PackageReference Include="SkbKontur.Graphite.Client" Version="1.3.16" />
diff --git a/Cassandra.DistributedTaskQueue/Handling/IRtqPeriodicJobRunner.cs b/Cassandra.DistributedTaskQueue/Handling/IRtqPeriodicJobRunner.cs
@@ -1,4 +1,5 @@
-﻿using System;
+using System;
+using System.Threading;
 
 using JetBrains.Annotations;
 
@@ -15,9 +16,10 @@ void RunPeriodicJob([NotNull] string jobId,
 
         void RunPeriodicJobWithLeaderElection([NotNull] string jobId,
                                               TimeSpan delayBetweenIterations,
-                                              [NotNull] Action jobAction,
+                                              [NotNull] Action<CancellationToken> jobAction,
                                               [NotNull] Action onTakeTheLead,
-                                              [NotNull] Action onLoseTheLead);
+                                              [NotNull] Action onLoseTheLead,
+                                              CancellationToken cancellationToken);
 
         void StopPeriodicJobWithLeaderElection([NotNull] string jobId);
     }
diff --git a/global.json b/global.json
@@ -1,5 +1,5 @@
 {
   "sdk": {
-    "version": "5.0.100"
+    "version": "5.0.102"
   }
 }
diff --git a/version.json b/version.json
@@ -1,5 +1,5 @@
 {
-  "version": "1.1",
+  "version": "1.2",
   "assemblyVersion": {
     "precision": "build"
   },

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,6 @@`
`1`		`-using JetBrains.Annotations;`
	`1`	`+using System.Threading;`
	`2`	`+`
	`3`	`+using JetBrains.Annotations;`
`2`	`4`
`3`	`5`	`using SkbKontur.EventFeeds;`
`4`	`6`
`@@ -8,6 +10,6 @@ namespace SkbKontur.Cassandra.DistributedTaskQueue.Monitoring.Indexer`
`8`	`10`	`public interface IRtqMonitoringEventFeeder`
`9`	`11`	`{`
`10`	`12`	`[NotNull]`
`11`		`- IEventFeedsRunner RunEventFeeding();`
	`13`	`+ IEventFeedsRunner RunEventFeeding(CancellationToken cancellationToken);`
`12`	`14`	`}`
`13`	`15`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1,5 +1,5 @@`
`1`	`1`	`{`
`2`	`2`	`"sdk": {`
`3`		`- "version": "5.0.100"`
	`3`	`+ "version": "5.0.102"`
`4`	`4`	`}`
`5`	`5`	`}`