fix(worker): properly handle WorkerJobfetcher bean lifecycle

Author: fhussonnois

worker/src/main/java/io/kestra/worker/GrpcStubFactory.java

@@ -22,43 +22,40 @@
  */
 @Factory
 public class GrpcStubFactory {
-
-    @Inject
-    GrpcChannelManager grpcChannelManager;
-
+    
     @Bean
     @Singleton
-    public WorkerControllerServiceBlockingStub blockingWorkerServiceStub() {
-        return WorkerControllerServiceGrpc.newBlockingStub(grpcChannelManager.getDefaultChannel());
+    public WorkerControllerServiceBlockingStub blockingWorkerServiceStub(GrpcChannelManager manager) {
+        return WorkerControllerServiceGrpc.newBlockingStub(manager.getDefaultChannel());
     }
 
     @Bean
     @Singleton
-    public WorkerControllerServiceStub asyncWorkerServiceStub() {
-        return WorkerControllerServiceGrpc.newStub(grpcChannelManager.getDefaultChannel());
+    public WorkerControllerServiceStub asyncWorkerServiceStub(GrpcChannelManager manager) {
+        return WorkerControllerServiceGrpc.newStub(manager.getDefaultChannel());
     }
 
     @Bean
     @Singleton
-    public LivenessControllerServiceBlockingStub workerServiceStub() {
-        return LivenessControllerServiceGrpc.newBlockingStub(grpcChannelManager.getDefaultChannel());
+    public LivenessControllerServiceBlockingStub workerServiceStub(GrpcChannelManager manager) {
+        return LivenessControllerServiceGrpc.newBlockingStub(manager.getDefaultChannel());
     }
 
     @Bean
     @Singleton
-    public ConnectControllerServiceBlockingStub connectControllerServiceBlockingStub() {
-        return ConnectControllerServiceGrpc.newBlockingStub(grpcChannelManager.getDefaultChannel());
+    public ConnectControllerServiceBlockingStub connectControllerServiceBlockingStub(GrpcChannelManager manager) {
+        return ConnectControllerServiceGrpc.newBlockingStub(manager.getDefaultChannel());
     }
 
     @Bean
     @Singleton
-    public WorkerFlowMetaStoreServiceBlockingStub workerFlowMetaStoreServiceBlockingStub() {
-        return WorkerFlowMetaStoreServiceGrpc.newBlockingStub(grpcChannelManager.getDefaultChannel());
+    public WorkerFlowMetaStoreServiceBlockingStub workerFlowMetaStoreServiceBlockingStub(GrpcChannelManager manager) {
+        return WorkerFlowMetaStoreServiceGrpc.newBlockingStub(manager.getDefaultChannel());
     }
 
     @Bean
     @Singleton
-    public KVMetadataServiceBlockingStub kvMetadataServiceBlockingStub() {
-        return KVMetadataServiceGrpc.newBlockingStub(grpcChannelManager.getDefaultChannel());
+    public KVMetadataServiceBlockingStub kvMetadataServiceBlockingStub(GrpcChannelManager manager) {
+        return KVMetadataServiceGrpc.newBlockingStub(manager.getDefaultChannel());
     }
 }

worker/src/main/java/io/kestra/worker/fetchers/WorkerJobFetcher.java

@@ -18,8 +18,8 @@
 import io.kestra.worker.WorkerLoop;
 import io.kestra.worker.queues.WorkerQueue;
 import io.kestra.worker.queues.WorkerQueueRegistry;
-import io.micronaut.context.annotation.Prototype;
 import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
 import lombok.extern.slf4j.Slf4j;
 
 import java.time.Duration;
@@ -47,7 +47,7 @@
  * ACKs are "receipt ACKs" (not completion ACKs) - they signal that the job was received and
  * is queued locally, allowing the controller to clean up its in-memory tracking.
  */
-@Prototype
+@Singleton
 @Slf4j
 public class WorkerJobFetcher extends WorkerLoop {