Repository: apache/kafka
File: connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ConnectMetricsRegistry.java (Java)

Code before:
        connectorTotalTaskCount = createTemplate("connector-total-task-count", WORKER_GROUP_NAME,
            "The number of tasks of the connector on the worker.", workerConnectorTags);
        connectorRunningTaskCount = createTemplate("connector-running-task-count", WORKER_GROUP_NAME,
            "The number of running tasks of the connector on the worker.", workerConnectorTags);
        connectorPausedTaskCount = createTemplate("connector-paused-task-count", WORKER_GROUP_NAME,
            "The number of paused tasks of the connector on the worker.", workerConnectorTags);
        connectorFailedTaskCount = createTemplate("connector-failed-task-count", WORKER_GROUP_NAME,
            "The number of failed tasks of the connector on the worker.", workerConnectorTags);
        connectorUnassignedTaskCount = createTemplate("connector-unassigned-task-count",
            WORKER_GROUP_NAME,
            "The number of unassigned tasks of the connector on the worker.", workerConnectorTags);
        connectorDestroyedTaskCount = createTemplate("connector-destroyed-task-count",
            WORKER_GROUP_NAME,
            "The number of destroyed tasks of the connector on the worker.", workerConnectorTags);
        connectorRestartingTaskCount = createTemplate("connector-restarting-task-count",
            WORKER_GROUP_NAME,
            "The number of restarting tasks of the connector on the worker.", workerConnectorTags);

        connectorStatusMetrics = new HashMap<>();
        connectorStatusMetrics.put(connectorRunningTaskCount, TaskStatus.State.RUNNING);
        connectorStatusMetrics.put(connectorPausedTaskCount, TaskStatus.State.PAUSED);
        connectorStatusMetrics.put(connectorFailedTaskCount, TaskStatus.State.FAILED);
        connectorStatusMetrics.put(connectorUnassignedTaskCount, TaskStatus.State.UNASSIGNED);
        connectorStatusMetrics.put(connectorDestroyedTaskCount, TaskStatus.State.DESTROYED);
        connectorStatusMetrics.put(connectorRestartingTaskCount, TaskStatus.State.RESTARTING);
        connectorStatusMetrics = Map.copyOf(connectorStatusMetrics);

        /* Worker rebalance level */
        Set<String> rebalanceTags = new LinkedHashSet<>(tags);

        connectProtocol = createTemplate("connect-protocol", WORKER_REBALANCE_GROUP_NAME, "The Connect protocol used by this cluster", rebalanceTags);
        leaderName = createTemplate("leader-name", WORKER_REBALANCE_GROUP_NAME, "The name of the group leader.", rebalanceTags);
        epoch = createTemplate("epoch", WORKER_REBALANCE_GROUP_NAME, "The epoch or generation number of this worker.", rebalanceTags);
        rebalanceCompletedTotal = createTemplate("completed-rebalances-total", WORKER_REBALANCE_GROUP_NAME,
                                                "The total number of rebalances completed by this worker.", rebalanceTags);
        rebalanceMode = createTemplate("rebalancing", WORKER_REBALANCE_GROUP_NAME,
                                               "Whether this worker is currently rebalancing.", rebalanceTags);
        rebalanceTimeMax = createTemplate("rebalance-max-time-ms", WORKER_REBALANCE_GROUP_NAME,
                                          "The maximum time in milliseconds spent by this worker to rebalance.", rebalanceTags);
        rebalanceTimeAvg = createTemplate("rebalance-avg-time-ms", WORKER_REBALANCE_GROUP_NAME,
                                          "The average time in milliseconds spent by this worker to rebalance.", rebalanceTags);
        rebalanceTimeSinceLast = createTemplate("time-since-last-rebalance-ms", WORKER_REBALANCE_GROUP_NAME,
                                                "The time in milliseconds since this worker completed the most recent rebalance.", rebalanceTags);

        /* Task Error Handling Metrics */
        Set<String> taskErrorHandlingTags = new LinkedHashSet<>(tags);
        taskErrorHandlingTags.add(CONNECTOR_TAG_NAME);
        taskErrorHandlingTags.add(TASK_TAG_NAME);

        recordProcessingFailures = createTemplate("total-record-failures", TASK_ERROR_HANDLING_GROUP_NAME,
                "The number of record processing failures in this task.", taskErrorHandlingTags);

Review comment (suggestion):
`connectorStatusMetrics` could be initialized with `Map.of`. Also, it could be a "final" variable

Code after:
            "The number of tasks of the connector on the worker.", workerConnectorTags);
        connectorRunningTaskCount = createTemplate("connector-running-task-count", WORKER_GROUP_NAME,
            "The number of running tasks of the connector on the worker.", workerConnectorTags);
        connectorPausedTaskCount = createTemplate("connector-paused-task-count", WORKER_GROUP_NAME,
            "The number of paused tasks of the connector on the worker.", workerConnectorTags);
        connectorFailedTaskCount = createTemplate("connector-failed-task-count", WORKER_GROUP_NAME,
            "The number of failed tasks of the connector on the worker.", workerConnectorTags);
        connectorUnassignedTaskCount = createTemplate("connector-unassigned-task-count",
            WORKER_GROUP_NAME,
            "The number of unassigned tasks of the connector on the worker.", workerConnectorTags);
        connectorDestroyedTaskCount = createTemplate("connector-destroyed-task-count",
            WORKER_GROUP_NAME,
            "The number of destroyed tasks of the connector on the worker.", workerConnectorTags);
        connectorRestartingTaskCount = createTemplate("connector-restarting-task-count",
            WORKER_GROUP_NAME,
            "The number of restarting tasks of the connector on the worker.", workerConnectorTags);

        connectorStatusMetrics = Map.of(
            connectorRunningTaskCount, TaskStatus.State.RUNNING,
            connectorPausedTaskCount, TaskStatus.State.PAUSED,
            connectorFailedTaskCount, TaskStatus.State.FAILED,
            connectorUnassignedTaskCount, TaskStatus.State.UNASSIGNED,
            connectorDestroyedTaskCount, TaskStatus.State.DESTROYED,
            connectorRestartingTaskCount, TaskStatus.State.RESTARTING
        );

        /* Worker rebalance level */
        Set<String> rebalanceTags = new LinkedHashSet<>(tags);

        connectProtocol = createTemplate("connect-protocol", WORKER_REBALANCE_GROUP_NAME, "The Connect protocol used by this cluster", rebalanceTags);
        leaderName = createTemplate("leader-name", WORKER_REBALANCE_GROUP_NAME, "The name of the group leader.", rebalanceTags);
        epoch = createTemplate("epoch", WORKER_REBALANCE_GROUP_NAME, "The epoch or generation number of this worker.", rebalanceTags);
        rebalanceCompletedTotal = createTemplate("completed-rebalances-total", WORKER_REBALANCE_GROUP_NAME,
                                                "The total number of rebalances completed by this worker.", rebalanceTags);
        rebalanceMode = createTemplate("rebalancing", WORKER_REBALANCE_GROUP_NAME,
                                               "Whether this worker is currently rebalancing.", rebalanceTags);
        rebalanceTimeMax = createTemplate("rebalance-max-time-ms", WORKER_REBALANCE_GROUP_NAME,
                                          "The maximum time in milliseconds spent by this worker to rebalance.", rebalanceTags);
        rebalanceTimeAvg = createTemplate("rebalance-avg-time-ms", WORKER_REBALANCE_GROUP_NAME,
                                          "The average time in milliseconds spent by this worker to rebalance.", rebalanceTags);
        rebalanceTimeSinceLast = createTemplate("time-since-last-rebalance-ms", WORKER_REBALANCE_GROUP_NAME,
                                                "The time in milliseconds since this worker completed the most recent rebalance.", rebalanceTags);

        /* Task Error Handling Metrics */
        Set<String> taskErrorHandlingTags = new LinkedHashSet<>(tags);
        taskErrorHandlingTags.add(CONNECTOR_TAG_NAME);
        taskErrorHandlingTags.add(TASK_TAG_NAME);

        recordProcessingFailures = createTemplate("total-record-failures", TASK_ERROR_HANDLING_GROUP_NAME,
                "The number of record processing failures in this task.", taskErrorHandlingTags);
        recordProcessingErrors = createTemplate("total-record-errors", TASK_ERROR_HANDLING_GROUP_NAME,