@Override
    public int compute(QB batch, boolean flush, boolean shutdown) {
        return workerObserver.observeExecutionComputation(batch, () -> execution.compute(batch, flush, shutdown));
    }
@Test
    public void compute() throws IOException {
        final ManualAdvanceClock manualAdvanceClock = new ManualAdvanceClock(Instant.now());
        final MetricExt rootMetric = MetricExtFactory.newMetricExtFromTestClock(manualAdvanceClock);
        final MockCompiledExecution mockQueueBatchExecution = new MockCompiledExecution(manualAdvanceClock);

        final AbstractNamespacedMetricExt processEventsNamespace = namespaceMetric(rootMetric, "events");
        final AbstractNamespacedMetricExt pipelineAEventsNamespace = namespaceMetric(rootMetric, "pipelines", "a", "events");
        final AbstractNamespacedMetricExt pipelineBEventsNamespace = namespaceMetric(rootMetric, "pipelines", "b", "events");

        // we create two worker observers, one for each pipeline, connected to the relevant metric namespaces
        final WorkerObserver pipelineAWorkerObserver = new WorkerObserver(processEventsNamespace, pipelineAEventsNamespace);
        final WorkerObserver pipelineBWorkerObserver = new WorkerObserver(processEventsNamespace, pipelineBEventsNamespace);

        // we create three observed executions to test, one for pipeline A, and two for pipeline B
        final ObservedExecution<MockQueueBatch> executionPipelineAWorker1 = pipelineAWorkerObserver.ofExecution(mockQueueBatchExecution);
        final ObservedExecution<MockQueueBatch> executionPipelineBWorker1 = pipelineBWorkerObserver.ofExecution(mockQueueBatchExecution);
        final ObservedExecution<MockQueueBatch> executionPipelineBWorker2 = pipelineBWorkerObserver.ofExecution(mockQueueBatchExecution);

        // in pipeline A, we take 110.9ms to filter 100 events and output 10 events
        final MockQueueBatch mockQueueBatchA = new MockQueueBatch(100, 10, 110_900_000L);
        final int eventsOutputA = executionPipelineAWorker1.compute(mockQueueBatchA, false, false);
        assertThat(eventsOutputA, is(equalTo(10)));

        // in pipeline B on worker 1, we take 1010.9ms to filter 1000 events and output 100 events
        final MockQueueBatch mockQueueBatchB = new MockQueueBatch(1000, 100, 1_010_900_000L);
        final int eventsOutputB = executionPipelineBWorker1.compute(mockQueueBatchB, false, false);
        assertThat(eventsOutputB, is(equalTo(100)));

        // again in pipeline B on worker 1, we take 10010.9ms to filter 1000 events and output 1000 events
        final MockQueueBatch mockQueueBatchB2 = new MockQueueBatch(1000, 1000, 10_010_900_000L);
        final int eventsOutputB2 = executionPipelineBWorker1.compute(mockQueueBatchB2, false, false);
        assertThat(eventsOutputB2, is(equalTo(1000)));

        // and in pipeline B on worker 2, we take 100010.9ms to filter 1000 events and output 10000 events
        final MockQueueBatch mockQueueBatchB3 = new MockQueueBatch(1000, 10000, 100_010_900_000L);
        final int eventsOutputB3 = executionPipelineBWorker2.compute(mockQueueBatchB3, false, false);
        assertThat(eventsOutputB3, is(equalTo(10000)));

        // validate that the inbound filter counts made it to our independent pipeline metrics and to the combined process
        final LongCounter pipelineAEventsFilteredCounter = LongCounter.fromRubyBase(pipelineAEventsNamespace, MetricKeys.FILTERED_KEY);
        final LongCounter pipelineBEventsFilteredCounter = LongCounter.fromRubyBase(pipelineBEventsNamespace, MetricKeys.FILTERED_KEY);
        final LongCounter processEventsFilteredCounter = LongCounter.fromRubyBase(processEventsNamespace, MetricKeys.FILTERED_KEY);
        assertThat(pipelineAEventsFilteredCounter.getValue(), is(equalTo(100L)));
        assertThat(pipelineBEventsFilteredCounter.getValue(), is(equalTo(3000L)));
        assertThat(processEventsFilteredCounter.getValue(), is(equalTo(3100L)));

        // validate that the outbound execution counts made it to our independent pipeline metrics and to the combined process
        final LongCounter pipelineAEventsOutCounter = LongCounter.fromRubyBase(pipelineAEventsNamespace, MetricKeys.OUT_KEY);
        final LongCounter pipelineBEventsOutCounter = LongCounter.fromRubyBase(pipelineBEventsNamespace, MetricKeys.OUT_KEY);
        final LongCounter processEventsOutCounter = LongCounter.fromRubyBase(processEventsNamespace, MetricKeys.OUT_KEY);
        assertThat(pipelineAEventsOutCounter.getValue(), is(equalTo(10L)));
        assertThat(pipelineBEventsOutCounter.getValue(), is(equalTo(11100L)));
        assertThat(processEventsOutCounter.getValue(), is(equalTo(11110L)));

        // validate that the timings were reported to our independent pipeline metrics and to the combined process
        final TimerMetric pipelineADurationTimer = TimerMetric.fromRubyBase(pipelineAEventsNamespace, MetricKeys.DURATION_IN_MILLIS_KEY);
        final TimerMetric pipelineBDurationTimer = TimerMetric.fromRubyBase(pipelineBEventsNamespace, MetricKeys.DURATION_IN_MILLIS_KEY);
        final TimerMetric processDurationTimer = TimerMetric.fromRubyBase(processEventsNamespace, MetricKeys.DURATION_IN_MILLIS_KEY);
        assertThat(pipelineADurationTimer.getValue(), is(equalTo(110L)));    // 110.9 -> 110
        assertThat(pipelineBDurationTimer.getValue(), is(equalTo(111032L))); // 1010.9 + 10010.9 + 100010.9 = 111032.7 -> 111032
        assertThat(processDurationTimer.getValue(), is(equalTo(111143L)));   // 110.9 + 101.9 + 1001.9 + 10001.9 = 111143.6 -> 111143
    }