Fixing BWC tests

Signed-off-by: Martin Gaievski <[email protected]>

Author: martin-gaievski

server/src/main/java/org/opensearch/search/pipeline/Pipeline.java

@@ -289,13 +289,19 @@ <Result extends SearchPhaseResult> void runSearchPhaseResultsTransformer(
         PipelineProcessingContext requestContext
     ) throws SearchPipelineProcessingException {
         long pipelineStart = relativeTimeSupplier.getAsLong();
+        boolean processorsExecuted = false;
+        boolean processorsFailed = false;
+        
+        // Always track phase results metrics to maintain consistent state
         beforeTransformPhaseResults();
+        
         try {
             for (SearchPhaseResultsProcessor searchPhaseResultsProcessor : searchPhaseResultsProcessors) {
                 beforePhaseResultsProcessor(searchPhaseResultsProcessor);
                 long start = relativeTimeSupplier.getAsLong();
                 if (currentPhase.equals(searchPhaseResultsProcessor.getBeforePhase().getName())
                     && nextPhase.equals(searchPhaseResultsProcessor.getAfterPhase().getName())) {
+                    processorsExecuted = true;
                     try {
                         searchPhaseResultsProcessor.process(searchPhaseResult, context, requestContext);
                         long took = TimeUnit.NANOSECONDS.toMillis(relativeTimeSupplier.getAsLong() - start);
@@ -315,6 +321,7 @@ <Result extends SearchPhaseResult> void runSearchPhaseResultsTransformer(
                             );
                         } else {
                             // Only track pipeline-level failure when processor failure is not ignored
+                            processorsFailed = true;
                             onTransformPhaseResultsFailure();
                             throw e;
                         }
@@ -327,9 +334,15 @@ <Result extends SearchPhaseResult> void runSearchPhaseResultsTransformer(
         } finally {
             long took = TimeUnit.NANOSECONDS.toMillis(relativeTimeSupplier.getAsLong() - pipelineStart);
             afterTransformPhaseResults(took);
+            // Notify that processors actually executed for pipeline request tracking
+            if (processorsExecuted) {
+                onPhaseResultsProcessorsExecuted(took, processorsFailed);
+            }
         }
     }
 
+    protected void onPhaseResultsProcessorsExecuted(long timeInNanos, boolean failed) {}
+
     static final Pipeline NO_OP_PIPELINE = new Pipeline(
         SearchPipelineService.NOOP_PIPELINE_ID,
         "Pipeline that does not transform anything",

server/src/main/java/org/opensearch/search/pipeline/PipelineWithMetrics.java

@@ -249,24 +249,18 @@ protected void onResponseProcessorFailed(Processor processor) {
     protected void beforeTransformPhaseResults() {
         super.beforeTransformPhaseResults();
         totalPhaseResultsMetrics.before();
-        // Phase results processing is also tracked as part of individual pipeline request processing
-        pipelineRequestMetrics.before();
     }
 
     @Override
     protected void afterTransformPhaseResults(long timeInNanos) {
         super.afterTransformPhaseResults(timeInNanos);
         totalPhaseResultsMetrics.after(timeInNanos);
-        // Phase results processing is also tracked as part of individual pipeline request processing
-        pipelineRequestMetrics.after(timeInNanos);
     }
 
     @Override
     protected void onTransformPhaseResultsFailure() {
         super.onTransformPhaseResultsFailure();
         totalPhaseResultsMetrics.failed();
-        // Phase results processing failures are also tracked as part of individual pipeline request processing
-        pipelineRequestMetrics.failed();
     }
 
     protected void beforePhaseResultsProcessor(Processor processor) {
@@ -281,6 +275,16 @@ protected void onPhaseResultsProcessorFailed(Processor processor) {
         phaseResultsProcessorMetrics.get(getProcessorKey(processor)).failed();
     }
 
+    @Override
+    protected void onPhaseResultsProcessorsExecuted(long timeInNanos, boolean failed) {
+        // Only track as pipeline request when phase results processors actually execute
+        pipelineRequestMetrics.before();
+        pipelineRequestMetrics.after(timeInNanos);
+        if (failed) {
+            pipelineRequestMetrics.failed();
+        }
+    }
+
     void copyMetrics(PipelineWithMetrics oldPipeline) {
         pipelineRequestMetrics.add(oldPipeline.pipelineRequestMetrics);
         pipelineResponseMetrics.add(oldPipeline.pipelineResponseMetrics);