/*
 * Copyright Debezium Authors.
 *
 * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
 */
package io.debezium.connector.oracle.logminer;

import java.math.BigInteger;
import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.time.OffsetDateTime;
import java.time.ZoneOffset;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.debezium.annotation.ThreadSafe;
import io.debezium.connector.base.ChangeEventQueueMetrics;
import io.debezium.connector.common.CdcSourceTaskContext;
import io.debezium.connector.oracle.AbstractOracleStreamingChangeEventSourceMetrics;
import io.debezium.connector.oracle.OracleConnectorConfig;
import io.debezium.connector.oracle.Scn;
import io.debezium.pipeline.source.spi.EventMetadataProvider;
import io.debezium.util.LRUCacheMap;
import io.debezium.util.Strings;

/**
 * Oracle Streaming Metrics implementation for the Oracle LogMiner streaming adapter.
 *
 * @author Chris Cranford
 */
@ThreadSafe
public class LogMinerStreamingChangeEventSourceMetrics
        extends AbstractOracleStreamingChangeEventSourceMetrics
        implements LogMinerStreamingChangeEventSourceMetricsMXBean {

    private static final Logger LOGGER = LoggerFactory.getLogger(LogMinerStreamingChangeEventSourceMetrics.class);

    private static final long MILLIS_PER_SECOND = 1000L;
    private static final int TRANSACTION_ID_SET_SIZE = 10;

    private final OracleConnectorConfig connectorConfig;

    private final Instant startTime;
    private final Clock clock;

    private final AtomicReference<Scn> currentScn = new AtomicReference<>(Scn.NULL);
    private final AtomicReference<Scn> offsetScn = new AtomicReference<>(Scn.NULL);
    private final AtomicReference<Scn> commitScn = new AtomicReference<>(Scn.NULL);
    private final AtomicReference<Scn> oldestScn = new AtomicReference<>(Scn.NULL);
    private final AtomicReference<Instant> oldestScnTime = new AtomicReference<>();
    private final AtomicReference<String[]> currentLogFileNames = new AtomicReference<>(new String[0]);
    private final AtomicReference<String[]> redoLogStatuses = new AtomicReference<>(new String[0]);
    private final AtomicReference<ZoneOffset> databaseZoneOffset = new AtomicReference<>(ZoneOffset.UTC);

    private final AtomicInteger batchSize = new AtomicInteger();
    private final AtomicInteger logSwitchCount = new AtomicInteger();
    private final AtomicInteger logMinerQueryCount = new AtomicInteger();

    private final AtomicLong sleepTime = new AtomicLong();
    private final AtomicLong minimumLogsMined = new AtomicLong();
    private final AtomicLong maximumLogsMined = new AtomicLong();
    private final AtomicLong maxBatchProcessingThroughput = new AtomicLong();
    private final AtomicLong timeDifference = new AtomicLong();
    private final AtomicLong processedRowsCount = new AtomicLong();
    private final AtomicLong activeTransactionCount = new AtomicLong();
    private final AtomicLong rolledBackTransactionCount = new AtomicLong();
    private final AtomicLong oversizedTransactionCount = new AtomicLong();
    private final AtomicLong changesCount = new AtomicLong();
    private final AtomicLong scnFreezeCount = new AtomicLong();

    private final DurationHistogramMetric batchProcessingDuration = new DurationHistogramMetric();
    private final DurationHistogramMetric fetchQueryDuration = new DurationHistogramMetric();
    private final DurationHistogramMetric commitDuration = new DurationHistogramMetric();
    private final DurationHistogramMetric lagFromSourceDuration = new DurationHistogramMetric();
    private final DurationHistogramMetric miningSessionStartupDuration = new DurationHistogramMetric();
    private final DurationHistogramMetric parseTimeDuration = new DurationHistogramMetric();
    private final DurationHistogramMetric resultSetNextDuration = new DurationHistogramMetric();

    private final MaxLongValueMetric userGlobalAreaMemory = new MaxLongValueMetric();
    private final MaxLongValueMetric processGlobalAreaMemory = new MaxLongValueMetric();

    private final LRUSet<String> abandonedTransactionIds = new LRUSet<>(TRANSACTION_ID_SET_SIZE);
    private final LRUSet<String> rolledBackTransactionIds = new LRUSet<>(TRANSACTION_ID_SET_SIZE);

    public LogMinerStreamingChangeEventSourceMetrics(CdcSourceTaskContext taskContext,
                                                     ChangeEventQueueMetrics changeEventQueueMetrics,
                                                     EventMetadataProvider metadataProvider,
                                                     OracleConnectorConfig connectorConfig) {
        this(taskContext, changeEventQueueMetrics, metadataProvider, connectorConfig, Clock.systemUTC());
    }

    public LogMinerStreamingChangeEventSourceMetrics(CdcSourceTaskContext taskContext,
                                                     ChangeEventQueueMetrics changeEventQueueMetrics,
                                                     EventMetadataProvider metadataProvider,
                                                     OracleConnectorConfig connectorConfig,
                                                     Clock clock) {
        super(taskContext, changeEventQueueMetrics, metadataProvider);
        this.connectorConfig = connectorConfig;
        this.batchSize.set(connectorConfig.getLogMiningBatchSizeDefault());
        this.sleepTime.set(connectorConfig.getLogMiningSleepTimeDefault().toMillis());
        this.clock = clock;
        this.startTime = clock.instant();
        reset();
    }

    @Override
    public void reset() {
        super.reset();

        changesCount.set(0);
        processedRowsCount.set(0);
        logMinerQueryCount.set(0);
        activeTransactionCount.set(0);
        rolledBackTransactionCount.set(0);
        oversizedTransactionCount.set(0);
        scnFreezeCount.set(0);

        fetchQueryDuration.reset();
        batchProcessingDuration.reset();
        parseTimeDuration.reset();
        miningSessionStartupDuration.reset();
        userGlobalAreaMemory.reset();
        processGlobalAreaMemory.reset();
        lagFromSourceDuration.reset();
        commitDuration.reset();

        abandonedTransactionIds.reset();
        rolledBackTransactionIds.reset();

        oldestScnTime.set(null);
    }

    @Override
    public long getMillisecondsToKeepTransactionsInBuffer() {
        return connectorConfig.getLogMiningTransactionRetention().toMillis();
    }

    @Override
    public long getSleepTimeInMilliseconds() {
        return sleepTime.get();
    }

    @Override
    public BigInteger getCurrentScn() {
        return currentScn.get().asBigInteger();
    }

    @Override
    public BigInteger getOffsetScn() {
        return offsetScn.get().asBigInteger();
    }

    @Override
    public BigInteger getCommittedScn() {
        return commitScn.get().asBigInteger();
    }

    @Override
    public BigInteger getOldestScn() {
        return oldestScn.get().asBigInteger();
    }

    @Override
    public long getOldestScnAgeInMilliseconds() {
        if (Objects.isNull(oldestScnTime.get())) {
            return 0L;
        }
        return Duration.between(Instant.now(), oldestScnTime.get()).toMillis();
    }

    @Override
    public String[] getCurrentLogFileNames() {
        return currentLogFileNames.get();
    }

    @Override
    public int getBatchSize() {
        return batchSize.get();
    }

    @Override
    public long getMinimumMinedLogCount() {
        return minimumLogsMined.get();
    }

    @Override
    public long getMaximumMinedLogCount() {
        return maximumLogsMined.get();
    }

    @Override
    public String[] getRedoLogStatuses() {
        return redoLogStatuses.get();
    }

    @Override
    public int getLogSwitchCount() {
        return logSwitchCount.get();
    }

    @Override
    public long getTotalProcessedRows() {
        return processedRowsCount.get();
    }

    @Override
    public long getNumberOfActiveTransactions() {
        return activeTransactionCount.get();
    }

    @Override
    public long getNumberOfRolledBackTransactions() {
        return rolledBackTransactionCount.get();
    }

    @Override
    public long getNumberOfOversizedTransactions() {
        return oversizedTransactionCount.get();
    }

    @Override
    public long getTotalChangesCount() {
        return changesCount.get();
    }

    @Override
    public long getFetchQueryCount() {
        return logMinerQueryCount.get();
    }

    @Override
    public long getScnFreezeCount() {
        return scnFreezeCount.get();
    }

    @Override
    public long getLastDurationOfFetchQueryInMilliseconds() {
        return fetchQueryDuration.getLast().toMillis();
    }

    @Override
    public long getMaxDurationOfFetchQueryInMilliseconds() {
        return fetchQueryDuration.getMaximum().toMillis();
    }

    @Override
    public long getLastBatchProcessingTimeInMilliseconds() {
        return batchProcessingDuration.getLast().toMillis();
    }

    @Override
    public long getMinBatchProcessingTimeInMilliseconds() {
        return batchProcessingDuration.getMinimum().toMillis();
    }

    @Override
    public long getMaxBatchProcessingTimeInMilliseconds() {
        return batchProcessingDuration.getMaximum().toMillis();
    }

    @Override
    public long getTotalBatchProcessingTimeInMilliseconds() {
        return batchProcessingDuration.getTotal().toMillis();
    }

    @Override
    public long getCommitThroughput() {
        final long timeSpent = Duration.between(startTime, clock.instant()).toMillis();
        return getNumberOfCommittedTransactions() * MILLIS_PER_SECOND / (timeSpent != 0 ? timeSpent : 1);
    }

    @Override
    public long getLastBatchProcessingThroughput() {
        final Duration lastBatchProcessingDuration = batchProcessingDuration.getLast();
        if (lastBatchProcessingDuration.isZero()) {
            return 0L;
        }
        return Math.round(((float) getLastCapturedDmlCount() / lastBatchProcessingDuration.toMillis()) * 1000);
    }

    @Override
    public long getMaxBatchProcessingThroughput() {
        return this.maxBatchProcessingThroughput.get();
    }

    @Override
    public long getAverageBatchProcessingThroughput() {
        final Duration totalBatchProcessingDuration = batchProcessingDuration.getTotal();
        if (totalBatchProcessingDuration.isZero()) {
            return 0L;
        }
        return Math.round(((float) getTotalCapturedDmlCount() / totalBatchProcessingDuration.toMillis()) * 1000);
    }

    @Override
    public long getLastCommitDurationInMilliseconds() {
        return commitDuration.getLast().toMillis();
    }

    @Override
    public long getMaxCommitDurationInMilliseconds() {
        return commitDuration.getMaximum().toMillis();
    }

    @Override
    public long getLastMiningSessionStartTimeInMilliseconds() {
        return miningSessionStartupDuration.getLast().toMillis();
    }

    @Override
    public long getMaxMiningSessionStartTimeInMilliseconds() {
        return miningSessionStartupDuration.getMaximum().toMillis();
    }

    @Override
    public long getTotalMiningSessionStartTimeInMilliseconds() {
        return miningSessionStartupDuration.getTotal().toMillis();
    }

    @Override
    public long getTotalParseTimeInMilliseconds() {
        return parseTimeDuration.getTotal().toMillis();
    }

    @Override
    public long getTotalResultSetNextTimeInMilliseconds() {
        return resultSetNextDuration.getTotal().toMillis();
    }

    @Override
    public long getLagFromSourceInMilliseconds() {
        return lagFromSourceDuration.getLast().toMillis();
    }

    @Override
    public long getMinLagFromSourceInMilliseconds() {
        return lagFromSourceDuration.getMinimum().toMillis();
    }

    @Override
    public long getMaxLagFromSourceInMilliseconds() {
        return lagFromSourceDuration.getMaximum().toMillis();
    }

    @Override
    public long getMiningSessionUserGlobalAreaMemoryInBytes() {
        return userGlobalAreaMemory.getValue();
    }

    @Override
    public long getMiningSessionUserGlobalAreaMaxMemoryInBytes() {
        return userGlobalAreaMemory.getMax();
    }

    @Override
    public long getMiningSessionProcessGlobalAreaMemoryInBytes() {
        return processGlobalAreaMemory.getValue();
    }

    @Override
    public long getMiningSessionProcessGlobalAreaMaxMemoryInBytes() {
        return processGlobalAreaMemory.getMax();
    }

    @Override
    public Set<String> getAbandonedTransactionIds() {
        return abandonedTransactionIds.getAll();
    }

    @Override
    public Set<String> getRolledBackTransactionIds() {
        return rolledBackTransactionIds.getAll();
    }

    /**
     * @return database current zone offset
     */
    public ZoneOffset getDatabaseOffset() {
        return databaseZoneOffset.get();
    }

    /**
     * Set the currently used batch size for querying LogMiner.
     *
     * @param batchSize batch size used for querying LogMiner
     */
    public void setBatchSize(int batchSize) {
        this.batchSize.set(batchSize);
    }

    /**
     * Set the connector's currently used sleep/pause time between LogMiner queries.
     *
     * @param sleepTime sleep time between LogMiner queries
     */
    public void setSleepTime(long sleepTime) {
        this.sleepTime.set(sleepTime);
    }

    /**
     * Set the current system change number from the database.
     *
     * @param currentScn database current system change number
     */
    public void setCurrentScn(Scn currentScn) {
        this.currentScn.set(currentScn);
    }

    /**
     * Set the offset's low-watermark system change number.
     *
     * @param offsetScn offset's restart system change number, i.e. {@code scn} attribute
     */
    public void setOffsetScn(Scn offsetScn) {
        this.offsetScn.set(offsetScn);
    }

    /**
     * Sets the offset's high-watermark system change number.
     *
     * @param commitScn offset's commit system change number, i.e. {@code commit_scn} attribute
     */
    // todo: getter is Committed, should this match?
    public void setCommitScn(Scn commitScn) {
        this.commitScn.set(commitScn);
    }

    /**
     * Sets the details for the oldest system change number in the transaction buffer.
     *
     * @param oldestScn oldest system change number
     * @param changeTime time when the oldest system change number was created
     */
    public void setOldestScnDetails(Scn oldestScn, Instant changeTime) {
        this.oldestScn.set(oldestScn);
        this.oldestScnTime.set(changeTime);
    }

    /**
     * Set the current iteration's logs that are being mined.
     *
     * @param logFileNames current set of logs that are part of the mining session.
     */
    public void setCurrentLogFileNames(Set<String> logFileNames) {
        this.currentLogFileNames.set(logFileNames.toArray(String[]::new));
        if (logFileNames.size() < minimumLogsMined.get()) {
            minimumLogsMined.set(logFileNames.size());
        }
        else if (minimumLogsMined.get() == 0) {
            minimumLogsMined.set(logFileNames.size());
        }
        if (logFileNames.size() > maximumLogsMined.get()) {
            maximumLogsMined.set(logFileNames.size());
        }
    }

    /**
     * Set the current logs and their respective statuses.
     *
     * @param statuses map of file names as key and the file's status as the value.
     */
    public void setRedoLogStatuses(Map<String, String> statuses) {
        redoLogStatuses.set(statuses.entrySet().stream()
                .map(entry -> entry.getKey() + " | " + entry.getValue())
                .toArray(String[]::new));
    }

    /**
     * Set the number of log switches in the past day.
     *
     * @param logSwitchCount number of log switches
     */
    public void setSwitchCount(int logSwitchCount) {
        this.logSwitchCount.set(logSwitchCount);
    }

    /**
     * Sets the last number of rows processed by the LogMiner mining iteration.
     *
     * @param processedRowsCount number of rows processed in the last mining iteration
     */
    public void setLastProcessedRowsCount(long processedRowsCount) {
        this.processedRowsCount.getAndAdd(processedRowsCount);
    }

    /**
     * Sets the number of current, active transactions in the transaction buffer.
     *
     * @param activeTransactionCount number of active transactions
     */
    public void setActiveTransactionCount(long activeTransactionCount) {
        this.activeTransactionCount.set(activeTransactionCount);
    }

    /**
     * Increments the number of rolled back transactions.
     */
    public void incrementRolledBackTransactionCount() {
        rolledBackTransactionCount.incrementAndGet();
    }

    /**
     * Increments the number of over-sized transactions.
     */
    public void incrementOversizedTransactionCount() {
        oversizedTransactionCount.incrementAndGet();
    }

    /**
     * Increments the total changes seen.
     */
    public void incrementTotalChangesCount() {
        changesCount.incrementAndGet();
    }

    /**
     * Increments the number of LogMiner queries executed.
     */
    public void incrementLogMinerQueryCount() {
        logMinerQueryCount.incrementAndGet();
    }

    /**
     * Increments the number of times the system change number is considered frozen and has not
     * changed over several consecutive LogMiner query batches.
     */
    public void incrementScnFreezeCount() {
        scnFreezeCount.incrementAndGet();
    }

    /**
     * Sets the number of times the system change number is considered frozen and has not changed over several consecutive
     * LogMiner query batches.
     *
     * @param scnFreezeCount number of times the system change number is considered frozen
     */
    public void setScnFreezeCount(long scnFreezeCount) {
        this.scnFreezeCount.set(scnFreezeCount);
    }

    /**
     * Sets the duration of the last LogMiner query execution.
     *
     * @param duration duration of the last LogMiner query
     */
    public void setLastDurationOfFetchQuery(Duration duration) {
        fetchQueryDuration.set(duration);
        logMinerQueryCount.incrementAndGet();
    }

    /**
     * Sets the duration of the total processing of the last LogMiner query result-set.
     *
     * @param duration duration of the total processing of the last LogMiner query result-set
     */
    public void setLastBatchProcessingDuration(Duration duration) {
        batchProcessingDuration.set(duration);
        if (getLastBatchProcessingThroughput() > getMaxBatchProcessingThroughput()) {
            maxBatchProcessingThroughput.set(getLastBatchProcessingThroughput());
        }
    }

    /**
     * Sets the duration of the last transaction commit processing.
     *
     * @param duration duration of the last transaction commit processing
     */
    public void setLastCommitDuration(Duration duration) {
        commitDuration.set(duration);
    }

    /**
     * Sets the duration of the last LogMiner mining session start-up and data dictionary load.
     *
     * @param duration duration of the last LogMiner session start-up
     */
    public void setLastMiningSessionStartDuration(Duration duration) {
        miningSessionStartupDuration.set(duration);
    }

    /**
     * Sets the duration for parsing the last SQL statement.
     *
     * @param duration duration for parsing the last SQL statement
     */
    public void setLastParseTimeDuration(Duration duration) {
        parseTimeDuration.set(duration);
    }

    /**
     * Sets the duration for the last {@code ResultSet#next} function call.
     *
     * @param duration duration for the last result set next call
     */
    public void setLastResultSetNextDuration(Duration duration) {
        resultSetNextDuration.set(duration);
    }

    /**
     * Set the database's current user global area (UGA) memory statistics.
     *
     * @param memory current user global area memory
     * @param maxMemory maximum user global area memory
     */
    public void setUserGlobalAreaMemory(long memory, long maxMemory) {
        userGlobalAreaMemory.setValue(memory);
        userGlobalAreaMemory.setMax(maxMemory);
    }

    /**
     * Set the database's current process global area (PGA) memory statistics.
     *
     * @param memory current process global area memory
     * @param maxMemory maximum process global area memory
     */
    public void setProcessGlobalAreaMemory(long memory, long maxMemory) {
        processGlobalAreaMemory.setValue(memory);
        processGlobalAreaMemory.setMax(maxMemory);
    }

    /**
     * Add a transaction to the recently tracked abandoned transactions metric.
     *
     * @param transactionId transaction identifier
     */
    public void addAbandonedTransactionId(String transactionId) {
        if (!Strings.isNullOrBlank(transactionId)) {
            abandonedTransactionIds.add(transactionId);
        }
    }

    /**
     * Add a transaction to the recently rolled back transactions metric.
     *
     * @param transactionId transaction identifier
     */
    public void addRolledBackTransactionId(String transactionId) {
        if (!Strings.isNullOrBlank(transactionId)) {
            rolledBackTransactionIds.add(transactionId);
        }
    }

    /**
     * Sets the database time zone and calculates the difference in time between the database server
     * and the connector. These values are necessary to calculate lag metrics.
     *
     * @param databaseSystemTime the database {@code SYSTIMESTAMP} value
     */
    public void setDatabaseTimeDifference(OffsetDateTime databaseSystemTime) {
        this.databaseZoneOffset.set(databaseSystemTime.getOffset());
        LOGGER.trace("Timezone offset of database time is {} seconds.", databaseZoneOffset.get().getTotalSeconds());

        final Instant now = clock.instant();
        final long timeDifferenceInMilliseconds = Duration.between(databaseSystemTime.toInstant(), now).toMillis();
        this.timeDifference.set(timeDifferenceInMilliseconds);
        LOGGER.trace("Current time {} ms, database difference {} ms", now.toEpochMilli(), timeDifferenceInMilliseconds);
    }

    /**
     * Calculates the lag metrics based on the provided event's database change time.
     *
     * @param changeTime the change time when the database recorded the event
     */
    public void calculateLagFromSource(Instant changeTime) {
        if (changeTime != null) {
            final Instant adjustedTime = changeTime.plusMillis(timeDifference.longValue())
                    .minusSeconds(databaseZoneOffset.get().getTotalSeconds());
            lagFromSourceDuration.set(Duration.between(adjustedTime, clock.instant()).abs());
        }
    }

    @Override
    public String toString() {
        return "LogMinerStreamingChangeEventSourceMetrics{" +
                "connectorConfig=" + connectorConfig +
                ", startTime=" + startTime +
                ", clock=" + clock +
                ", currentScn=" + currentScn +
                ", offsetScn=" + offsetScn +
                ", commitScn=" + commitScn +
                ", oldestScn=" + oldestScn +
                ", oldestScnTime=" + oldestScnTime +
                ", currentLogFileNames=" + currentLogFileNames +
                ", redoLogStatuses=" + redoLogStatuses +
                ", databaseZoneOffset=" + databaseZoneOffset +
                ", batchSize=" + batchSize +
                ", logSwitchCount=" + logSwitchCount +
                ", logMinerQueryCount=" + logMinerQueryCount +
                ", sleepTime=" + sleepTime +
                ", minimumLogsMined=" + minimumLogsMined +
                ", maximumLogsMined=" + maximumLogsMined +
                ", maxBatchProcessingThroughput=" + maxBatchProcessingThroughput +
                ", timeDifference=" + timeDifference +
                ", processedRowsCount=" + processedRowsCount +
                ", activeTransactionCount=" + activeTransactionCount +
                ", rolledBackTransactionCount=" + rolledBackTransactionCount +
                ", oversizedTransactionCount=" + oversizedTransactionCount +
                ", changesCount=" + changesCount +
                ", scnFreezeCount=" + scnFreezeCount +
                ", batchProcessingDuration=" + batchProcessingDuration +
                ", fetchQueryDuration=" + fetchQueryDuration +
                ", commitDuration=" + commitDuration +
                ", lagFromSourceDuration=" + lagFromSourceDuration +
                ", miningSessionStartupDuration=" + miningSessionStartupDuration +
                ", parseTimeDuration=" + parseTimeDuration +
                ", resultSetNextDuration=" + resultSetNextDuration +
                ", userGlobalAreaMemory=" + userGlobalAreaMemory +
                ", processGlobalAreaMemory=" + processGlobalAreaMemory +
                ", abandonedTransactionIds=" + abandonedTransactionIds +
                ", rolledBackTransactionIds=" + rolledBackTransactionIds +
                "} " + super.toString();
    }

    /**
     * Utility class for tracking histogram-based values for a duration-based metric.
     */
    @ThreadSafe
    static class DurationHistogramMetric {

        private final AtomicReference<Duration> min = new AtomicReference<>(Duration.ZERO);
        private final AtomicReference<Duration> max = new AtomicReference<>(Duration.ZERO);
        private final AtomicReference<Duration> last = new AtomicReference<>(Duration.ZERO);
        private final AtomicReference<Duration> total = new AtomicReference<>(Duration.ZERO);

        /**
         * Resets the duration metric
         */
        void reset() {
            min.set(Duration.ZERO);
            max.set(Duration.ZERO);
            last.set(Duration.ZERO);
            total.set(Duration.ZERO);
        }

        /**
         * Sets the last duration-based value for the histogram.
         *
         * @param lastDuration last duration
         */
        void set(Duration lastDuration) {
            last.set(lastDuration);
            total.accumulateAndGet(lastDuration, Duration::plus);
            if (max.get().toMillis() < lastDuration.toMillis()) {
                max.set(lastDuration);
            }
            final long minimumValue = min.get().toMillis();
            if (minimumValue > lastDuration.toMillis()) {
                min.set(lastDuration);
            }
            else if (minimumValue == 0L) {
                min.set(lastDuration);
            }
        }

        Duration getMinimum() {
            return min.get();
        }

        Duration getMaximum() {
            return max.get();
        }

        Duration getLast() {
            return last.get();
        }

        Duration getTotal() {
            return total.get();
        }
    }

    /**
     * Utility class for tracking the current and maximum long value.
     */
    @ThreadSafe
    static class MaxLongValueMetric {

        private final AtomicLong value = new AtomicLong();
        private final AtomicLong max = new AtomicLong();

        public void reset() {
            value.set(0L);
            max.set(0L);
        }

        public void setValueAndCalculateMax(long value) {
            this.value.set(value);
            if (max.get() < value) {
                max.set(value);
            }
        }

        public void setValue(long value) {
            this.value.set(value);
        }

        public void setMax(long max) {
            if (this.max.get() < max) {
                this.max.set(max);
            }
        }

        public long getValue() {
            return value.get();
        }

        public long getMax() {
            return max.get();
        }
    }

    /**
     * Utility class for maintaining a least-recently-used list of values.
     *
     * @param <T> the argument type to be stored
     */
    @ThreadSafe
    static class LRUSet<T> {

        private final AtomicReference<LRUCacheMap<T, T>> cache = new AtomicReference<>();
        private final int capacity;

        LRUSet(int capacity) {
            this.cache.set(new LRUCacheMap<>(capacity));
            this.capacity = capacity;
        }

        public void reset() {
            this.cache.set(new LRUCacheMap<>(capacity));
        }

        public void add(T value) {
            this.cache.get().put(value, value);
        }

        public Set<T> getAll() {
            return this.cache.get().keySet();
        }
    }

}
