Repository: apache/kafka
File: metadata/src/main/java/org/apache/kafka/controller/metrics/QuorumControllerMetrics.java (Java)

Code before:
    private static final MetricName EVENT_QUEUE_OPERATIONS_TIMED_OUT_COUNT = getMetricName(
        "KafkaController", "EventQueueOperationsTimedOutCount");
    private static final MetricName NEW_ACTIVE_CONTROLLERS_COUNT = getMetricName(
        "KafkaController", "NewActiveControllersCount");

    private static final String TIME_SINCE_LAST_HEARTBEAT_RECEIVED_METRIC_NAME = "TimeSinceLastHeartbeatReceivedMs";
    private static final String BROKER_ID_TAG = "broker";
    private static final MetricConfig METRIC_CONFIG = new MetricConfig();

    private final Optional<MetricsRegistry> registry;
    private final Time time;
    private volatile boolean active;
    private final AtomicLong lastAppliedRecordOffset = new AtomicLong(0);
    private final AtomicLong lastCommittedRecordOffset = new AtomicLong(0);
    private final AtomicLong lastAppliedRecordTimestamp = new AtomicLong(0);
    private final Consumer<Long> eventQueueTimeUpdater;
    private final Consumer<Long> eventQueueProcessingTimeUpdater;
    public final TimeRatio avgIdleTimeRatio;

    private final AtomicLong timedOutHeartbeats = new AtomicLong(0);
    private final AtomicLong operationsStarted = new AtomicLong(0);
    private final AtomicLong operationsTimedOut = new AtomicLong(0);
    private final AtomicLong newActiveControllers = new AtomicLong(0);
    private final Map<Integer, Long> brokerContactTimesMs = new ConcurrentHashMap<>();
    private final int sessionTimeoutMs;
    private volatile OptionalLong idleStartTime = OptionalLong.empty();

    private Consumer<Long> newHistogram(MetricName name, boolean biased) {
        if (registry.isPresent()) {
            Histogram histogram = registry.get().newHistogram(name, biased);
            return histogram::update;
        } else {
            return __ -> { };
        }
    }

    public QuorumControllerMetrics(
        Optional<MetricsRegistry> registry,
        Time time,
        int sessionTimeoutMs
    ) {
        this.registry = registry;
        this.time = time;
        this.active = false;
        registry.ifPresent(r -> r.newGauge(ACTIVE_CONTROLLER_COUNT, new Gauge<Integer>() {
            @Override
            public Integer value() {
                return active ? 1 : 0;
            }
        }));
        this.eventQueueTimeUpdater = newHistogram(EVENT_QUEUE_TIME_MS, true);

Review comment (suggestion):
I think we should use an `AtomicLong` here and `compareAndSet` in our update methods, since we could be calling `updateIdleEndTime()` from the `deferredEventQueue` and `updateIdleStartTime()` from the controller event thread concurrently.

Code after:
    private static final MetricName NEW_ACTIVE_CONTROLLERS_COUNT = getMetricName(
        "KafkaController", "NewActiveControllersCount");

    private static final String TIME_SINCE_LAST_HEARTBEAT_RECEIVED_METRIC_NAME = "TimeSinceLastHeartbeatReceivedMs";
    private static final String BROKER_ID_TAG = "broker";

    private final Optional<MetricsRegistry> registry;
    private final Time time;
    private volatile boolean active;
    private final AtomicLong lastAppliedRecordOffset = new AtomicLong(0);
    private final AtomicLong lastCommittedRecordOffset = new AtomicLong(0);
    private final AtomicLong lastAppliedRecordTimestamp = new AtomicLong(0);
    private final Consumer<Long> eventQueueTimeUpdater;
    private final Consumer<Long> eventQueueProcessingTimeUpdater;
    private final TimeRatio avgIdleTimeRatio;

    private final AtomicLong timedOutHeartbeats = new AtomicLong(0);
    private final AtomicLong operationsStarted = new AtomicLong(0);
    private final AtomicLong operationsTimedOut = new AtomicLong(0);
    private final AtomicLong newActiveControllers = new AtomicLong(0);
    private final Map<Integer, Long> brokerContactTimesMs = new ConcurrentHashMap<>();
    private final int sessionTimeoutMs;

    private Consumer<Long> newHistogram(MetricName name, boolean biased) {
        if (registry.isPresent()) {
            Histogram histogram = registry.get().newHistogram(name, biased);
            return histogram::update;
        } else {
            return __ -> { };
        }
    }

    public QuorumControllerMetrics(
        Optional<MetricsRegistry> registry,
        Time time,
        int sessionTimeoutMs
    ) {
        this.registry = registry;
        this.time = time;
        this.active = false;
        registry.ifPresent(r -> r.newGauge(ACTIVE_CONTROLLER_COUNT, new Gauge<Integer>() {
            @Override
            public Integer value() {
                return active ? 1 : 0;
            }
        }));
        this.eventQueueTimeUpdater = newHistogram(EVENT_QUEUE_TIME_MS, true);
        this.eventQueueProcessingTimeUpdater = newHistogram(EVENT_QUEUE_PROCESSING_TIME_MS, true);
        this.sessionTimeoutMs = sessionTimeoutMs;
        this.avgIdleTimeRatio = new TimeRatio(1);
        registry.ifPresent(r -> r.newGauge(LAST_APPLIED_RECORD_OFFSET, new Gauge<Long>() {