Repository: apache/kafka
File: streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java (Java)

Code before:
            Stream.concat(
                streamOfNonPausedTasks(),
                pausedTasks().stream()
            );
    }

    private Stream<Task> streamOfNonPausedTasks() {
        return
            Stream.concat(
                tasksAndActions.stream()
                    .filter(taskAndAction -> taskAndAction.action() == Action.ADD)
                    .map(TaskAndAction::task),
                Stream.concat(
                    updatingTasks().stream(),
                    Stream.concat(
                        restoredActiveTasks.stream(),
                        Stream.concat(
                            exceptionsAndFailedTasks.stream().map(ExceptionAndTask::task),
                            removedTasks.stream()))));
    }

    private class StateUpdaterMetrics {
        private static final String STATE_LEVEL_GROUP = "stream-state-updater-metrics";

        private static final String IDLE_RATIO_DESCRIPTION = WINDOWED_RATIO_DESCRIPTION_PREFIX + "of the time this thread spend being idle";
        private static final String RESTORE_RATIO_DESCRIPTION = WINDOWED_RATIO_DESCRIPTION_PREFIX + "of the time this thread spend restoring active tasks";
        private static final String UPDATE_RATIO_DESCRIPTION = WINDOWED_RATIO_DESCRIPTION_PREFIX + "of the time this thread spend updating standby tasks";
        private static final String CHECKPOINT_RATIO_DESCRIPTION = WINDOWED_RATIO_DESCRIPTION_PREFIX + "of the time this thread spend checkpointing tasks restored progress";
        private static final String RESTORE_RECORDS_RATE_DESCRIPTION = RATE_DESCRIPTION + "records restored";
        private static final String RESTORE_RATE_DESCRIPTION = RATE_DESCRIPTION + "restore calls triggered";

        private final Sensor restoreSensor;
        private final Sensor idleRatioSensor;
        private final Sensor activeRestoreRatioSensor;
        private final Sensor standbyRestoreRatioSensor;
        private final Sensor checkpointRatioSensor;

        private final Deque<Sensor> allSensors = new LinkedList<>();
        private final Deque<MetricName> allMetricNames = new LinkedList<>();

        private StateUpdaterMetrics(final StreamsMetricsImpl metrics, final String threadId) {
            final Map<String, String> threadLevelTags = new LinkedHashMap<>();
            threadLevelTags.put(THREAD_ID_TAG, threadId);
            final Metrics metricsRegistry = metrics.metricsRegistry();

            MetricName metricName = metricsRegistry.metricName("active-restoring-tasks",
                STATE_LEVEL_GROUP,
                "The number of active tasks currently undergoing restoration",
                threadLevelTags);
            metricsRegistry.addMetric(metricName, (config, now) -> stateUpdaterThread != null ?
                stateUpdaterThread.numRestoringActiveTasks() : 0);

Review comment (refactor):
```suggestion
        private static final String RESTORE_RATIO_DESCRIPTION = WINDOWED_RATIO_DESCRIPTION_PREFIX + " restoring active tasks";
```

not sure about this one

spent


<img width="338" height="107" alt="Image" src="https://github.com/user-attachments/assets/c474a5e7-0503-4d4e-8f94-3e3ebec7b222" />


spend

<img width="786" height="808" alt="Image" src="https://github.com/user-attachments/assets/3e2b19ac-c198-4e66-a038-347d49f225a1" />


idk if this is any good, all I know this string occurs 22 times making it kind of dry.

"of the time this thread spen"



<img width="272" height="31" alt="Image" src="https://github.com/user-attachments/assets/c92a58fa-45c0-4291-b3c3-7227fd790336" />

Code after:
    private Stream<Task> streamOfTasks() {
        return
            Stream.concat(
                streamOfNonPausedTasks(),
                pausedTasks().stream()
            );
    }

    private Stream<Task> streamOfNonPausedTasks() {
        return
            Stream.concat(
                tasksAndActions.stream()
                    .filter(taskAndAction -> taskAndAction.action() == Action.ADD)
                    .map(TaskAndAction::task),
                Stream.concat(
                    updatingTasks().stream(),
                    Stream.concat(
                        restoredActiveTasks.stream(),
                        Stream.concat(
                            exceptionsAndFailedTasks.stream().map(ExceptionAndTask::task),
                            removedTasks.stream()))));
    }

    private class StateUpdaterMetrics {
        private static final String STATE_LEVEL_GROUP = "stream-state-updater-metrics";

        private static final String IDLE_RATIO_DESCRIPTION = WINDOWED_RATIO_DESCRIPTION_PREFIX + THREAD_TIME_UNIT_DESCRIPTION + "being idle";
        private static final String RESTORE_RATIO_DESCRIPTION = WINDOWED_RATIO_DESCRIPTION_PREFIX + THREAD_TIME_UNIT_DESCRIPTION + "restoring active tasks";
        private static final String UPDATE_RATIO_DESCRIPTION = WINDOWED_RATIO_DESCRIPTION_PREFIX + THREAD_TIME_UNIT_DESCRIPTION + "updating standby tasks";
        private static final String CHECKPOINT_RATIO_DESCRIPTION = WINDOWED_RATIO_DESCRIPTION_PREFIX + THREAD_TIME_UNIT_DESCRIPTION + "checkpointing tasks restored progress";
        private static final String RESTORE_RECORDS_RATE_DESCRIPTION = RATE_DESCRIPTION + "records restored";
        private static final String RESTORE_RATE_DESCRIPTION = RATE_DESCRIPTION + "restore calls triggered";

        private final Sensor restoreSensor;
        private final Sensor idleRatioSensor;
        private final Sensor activeRestoreRatioSensor;
        private final Sensor standbyRestoreRatioSensor;
        private final Sensor checkpointRatioSensor;

        private final Deque<Sensor> allSensors = new LinkedList<>();
        private final Deque<MetricName> allMetricNames = new LinkedList<>();

        private StateUpdaterMetrics(final StreamsMetricsImpl metrics, final String threadId) {
            final Map<String, String> threadLevelTags = new LinkedHashMap<>();
            threadLevelTags.put(THREAD_ID_TAG, threadId);
            final Metrics metricsRegistry = metrics.metricsRegistry();

            MetricName metricName = metricsRegistry.metricName("active-restoring-tasks",
                STATE_LEVEL_GROUP,
                "The number of active tasks currently undergoing restoration",
                threadLevelTags);