fixup! Refactor: pass scanId as parameter instead of instance field and fix metric timing

fix(tests): update tests to reflect messageId removal and scanId parameter changes

Author: benzekrimaha

extensions/lifecycle/bucketProcessor/LifecycleBucketProcessor.js

@@ -98,8 +98,10 @@ class LifecycleBucketProcessor {
         this._pausedLocations = new Set();
         this._locationStatusStream = null;
 
-        // scan tracking for metrics
-        this._currentScanId = null;
+        // Conductor scan ID for metrics; kept across internal queue drains so
+        // Kafka batches from the same scan do not retrigger onBucketProcessorScanStart.
+        this._metricsScanId = null;
+        this._processorScanMetricsActive = false;
 
         this.clientManager = new ClientManager({
             id: 'lifecycle',
@@ -158,13 +160,13 @@ class LifecycleBucketProcessor {
                     error: err,
                 });
             }
-            if (this._currentScanId) {
-                this._log.info('bucket processor scan complete', {
+            if (this._processorScanMetricsActive) {
+                this._log.info('bucket processor internal task queue drained', {
                     method: 'LifecycleBucketProcessor.drain',
-                    conductorScanId: this._currentScanId,
+                    conductorScanId: this._metricsScanId,
                 });
                 LifecycleMetrics.onBucketProcessorScanEnd(this._log);
-                this._currentScanId = null;
+                this._processorScanMetricsActive = false;
             }
         });
 
@@ -293,11 +295,12 @@ class LifecycleBucketProcessor {
         }
         const { bucket, owner, accountId, taskVersion } = result.target;
         const conductorScanId = result.contextInfo && result.contextInfo.conductorScanId;
-        if (conductorScanId && conductorScanId !== this._currentScanId) {
-            this._currentScanId = conductorScanId;
+        if (conductorScanId && conductorScanId !== this._metricsScanId) {
+            this._metricsScanId = conductorScanId;
             LifecycleMetrics.onBucketProcessorScanStart(
                 this._log, Date.now()
             );
+            this._processorScanMetricsActive = true;
             this._log.info('new conductor scan detected', {
                 method:
                     'LifecycleBucketProcessor._processBucketEntry',