Fix.

CHANGELOG.md

@@ -15,7 +15,7 @@ to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 ## 1.44.0 - 2024/10/16
 
 - When registering cron tasks, log error if job already exists, but task is in error state.
-  - If silent mode is turned on, then this log will not appear.
+- If silent mode is turned on, then this log will not appear.
 
 ## 1.43.0 - 2024/08/09
 

tw-tasks-core/src/main/java/com/transferwise/tasks/processing/GlobalProcessingState.java

@@ -49,7 +49,7 @@ public Bucket(int minPriority, int maxPriority) {
       }
     }
 
-    // Optimization to avoid waiting behind a lock.
+    // Optimization to avoid waiting behind a lock, but also reducing the amount of signals and context switches.
     // One ongoing version update is enough to wake up all necessary components.
     private AtomicBoolean versionUpdateInProgress = new AtomicBoolean();
 

tw-tasks-core/src/main/java/com/transferwise/tasks/processing/TasksProcessingService.java

@@ -299,7 +299,6 @@ protected void markAsError(IBaseTask task, String bucketId) {
 
   protected ProcessTaskResponse grabTaskForProcessing(String bucketId, BaseTask task) {
     GlobalProcessingState.Bucket bucket = globalProcessingState.getBuckets().get(bucketId);
-    BucketProperties bucketProperties = bucketsManager.getBucketProperties(bucketId);
 
     ITaskHandler taskHandler = taskHandlerRegistry.getTaskHandler(task);
     if (taskHandler == null) {

tw-tasks-core/src/main/java/com/transferwise/tasks/triggering/KafkaTasksExecutionTriggerer.java

@@ -474,6 +474,8 @@ private void commitOffsetsWithLowFrequency(ConsumerBucket consumerBucket) {
             .map(e -> e.getKey() + ":" + e.getValue().offset()).collect(Collectors.joining(", ")));
       }
 
+      releaseCompletedOffsets(consumerBucket);
+
       consumerBucket.getKafkaConsumer().commitAsync(consumerBucket.getOffsetsToBeCommitted(), (map, e) -> {
         if (e != null) {
           coreMetricsTemplate.registerKafkaTasksExecutionTriggererCommit(bucketId, false, false);