Updates ExpireJobCacheWorker to use replica as long as possible

app/workers/all_queues.yml

@@ -1459,7 +1459,7 @@
   :urgency: :high
   :resource_boundary: :unknown
   :weight: 3
-  :idempotent: true
+  :idempotent:
   :tags: []
 - :name: pipeline_cache:expire_pipeline_cache
   :worker_name: ExpirePipelineCacheWorker

app/workers/expire_job_cache_worker.rb

 # frozen_string_literal: true
 
-class ExpireJobCacheWorker
+class ExpireJobCacheWorker # rubocop:disable Scalability/IdempotentWorker
   include ApplicationWorker
 
-  data_consistency :always
+  data_consistency :delayed, feature_flag: :load_balancing_for_expire_job_cache_worker
 
   sidekiq_options retry: 3
   include PipelineQueue
 
   queue_namespace :pipeline_cache
   urgency :high
-  idempotent!
+  # This worker should be idempotent, but we're switching to data_consistency
+  # :sticky and there is an ongoing incompatibility, so it needs to be disabled for
+  # now. The following line can be uncommented and this comment removed once
+  # https://gitlab.com/gitlab-org/gitlab/-/issues/325291 is resolved.
+  # idempotent!
 
   # rubocop: disable CodeReuse/ActiveRecord
   def perform(job_id)

config/feature_flags/development/load_balancing_for_expire_job_cache_worker.yml

+---
+name: load_balancing_for_expire_job_cache_worker
+introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/68791
+rollout_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/339137
+milestone: '14.3'
+type: development
+group: group::pipeline authoring
+default_enabled: false

spec/workers/expire_job_cache_worker_spec.rb

@@ -13,27 +13,6 @@ RSpec.describe ExpireJobCacheWorker do
 
       let(:job_args) { job.id }
 
-      include_examples 'an idempotent worker' do
-        it 'invalidates Etag caching for the job path' do
-          job_path = "/#{project.full_path}/builds/#{job.id}.json"
-
-          spy_store = Gitlab::EtagCaching::Store.new
-
-          allow(Gitlab::EtagCaching::Store).to receive(:new) { spy_store }
-
-          expect(spy_store).to receive(:touch)
-            .exactly(worker_exec_times).times
-            .with(job_path)
-            .and_call_original
-
-          expect(ExpirePipelineCacheWorker).to receive(:perform_async)
-            .with(pipeline.id)
-            .exactly(worker_exec_times).times
-
-          subject
-        end
-      end
-
       it 'does not perform extra queries', :aggregate_failures do
         worker = described_class.new
         recorder = ActiveRecord::QueryRecorder.new { worker.perform(job.id) }
@@ -51,6 +30,11 @@ RSpec.describe ExpireJobCacheWorker do
         expect(namespace_queries.size).to eq(0)
         expect(route_queries.size).to eq(0)
       end
+
+      it_behaves_like 'worker with data consistency',
+        described_class,
+        feature_flag: :load_balancing_for_expire_job_cache_worker,
+        data_consistency: :delayed
     end
 
     context 'when there is no job in the pipeline' do