public static Sensor processAtSourceSensor(final String threadId,
                                               final String taskId,
                                               final String processorNodeId,
                                               final StreamsMetricsImpl streamsMetrics) {
        final Sensor parentSensor = streamsMetrics.taskLevelSensor(threadId, taskId, PROCESS, RecordingLevel.DEBUG);
        addInvocationRateAndCountToSensor(
            parentSensor,
            TASK_LEVEL_GROUP,
            streamsMetrics.taskLevelTagMap(threadId, taskId),
            PROCESS,
            PROCESS_RATE_DESCRIPTION,
            PROCESS_TOTAL_DESCRIPTION
        );
        return throughputSensor(
            threadId,
            taskId,
            processorNodeId,
            PROCESS,
            PROCESS_RATE_DESCRIPTION,
            PROCESS_TOTAL_DESCRIPTION,
            RecordingLevel.DEBUG,
            streamsMetrics,
            parentSensor
        );
    }
@Test
    public void shouldGetProcessAtSourceSensor() {
        final String metricNamePrefix = "process";
        final String descriptionOfCount = "The total number of calls to process";
        final String descriptionOfRate = "The average number of calls to process per second";
        when(streamsMetrics.taskLevelSensor(THREAD_ID, TASK_ID, metricNamePrefix, RecordingLevel.DEBUG))
            .thenReturn(expectedParentSensor);
        when(streamsMetrics.taskLevelTagMap(THREAD_ID, TASK_ID))
            .thenReturn(parentTagMap);
        setUpThroughputSensor(metricNamePrefix, RecordingLevel.DEBUG, expectedParentSensor);

        try (final MockedStatic<StreamsMetricsImpl> streamsMetricsStaticMock = mockStatic(StreamsMetricsImpl.class)) {
            final Sensor sensor = ProcessorNodeMetrics.processAtSourceSensor(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, streamsMetrics);
            streamsMetricsStaticMock.verify(
                () -> StreamsMetricsImpl.addInvocationRateAndCountToSensor(
                    expectedSensor,
                    PROCESSOR_NODE_LEVEL_GROUP,
                    tagMap,
                    metricNamePrefix,
                    descriptionOfRate,
                    descriptionOfCount
                )
            );
            streamsMetricsStaticMock.verify(
                () -> StreamsMetricsImpl.addInvocationRateAndCountToSensor(
                    expectedParentSensor,
                    TASK_LEVEL_GROUP,
                    parentTagMap,
                    metricNamePrefix,
                    descriptionOfRate,
                    descriptionOfCount
                )
            );
            assertThat(sensor, is(expectedSensor));
        }
    }