<commit_before>package com.booking.replication;

import com.booking.replication.applier.Applier;
import com.booking.replication.applier.Partitioner;
import com.booking.replication.applier.Seeker;
import com.booking.replication.augmenter.Augmenter;
import com.booking.replication.augmenter.AugmenterFilter;
import com.booking.replication.augmenter.model.event.AugmentedEvent;
import com.booking.replication.checkpoint.CheckpointApplier;
import com.booking.replication.commons.checkpoint.Binlog;
import com.booking.replication.commons.checkpoint.Checkpoint;
import com.booking.replication.commons.checkpoint.ForceRewindException;
import com.booking.replication.commons.map.MapFlatter;
import com.booking.replication.commons.metrics.Metrics;
import com.booking.replication.controller.WebServer;
import com.booking.replication.coordinator.Coordinator;
import com.booking.replication.streams.Streams;
import com.booking.replication.supplier.Supplier;
import com.booking.replication.supplier.model.RawEvent;
import com.booking.utils.BootstrapReplicator;

import com.codahale.metrics.Gauge;
import com.codahale.metrics.MetricRegistry;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;

import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.DefaultParser;
import org.apache.commons.cli.HelpFormatter;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;

public class Replicator {

    public interface Configuration {
        String CHECKPOINT_PATH      = "checkpoint.path";
        String CHECKPOINT_DEFAULT   = "checkpoint.default";
        String REPLICATOR_THREADS   = "replicator.threads";
        String REPLICATOR_TASKS     = "replicator.tasks";

        String OVERRIDE_CHECKPOINT_START_POSITION   = "override.checkpoint.start.position";
        String OVERRIDE_CHECKPOINT_BINLOG_FILENAME  = "override.checkpoint.binLog.filename";
        String OVERRIDE_CHECKPOINT_BINLOG_POSITION  = "override.checkpoint.binLog.position";
        String OVERRIDE_CHECKPOINT_GTID_SET         = "override.checkpoint.gtidSet";
    }

    private static final Logger LOG = LogManager.getLogger(Replicator.class);
    private static final String COMMAND_LINE_SYNTAX = "java -jar mysql-replicator-<version>.jar";

    private final String checkpointPath;
    private final String checkpointDefault;
    private final Coordinator coordinator;
    private final Supplier supplier;
    private final Augmenter augmenter;
    private final AugmenterFilter augmenterFilter;
    private final Seeker seeker;
    private final Partitioner partitioner;
    private final Applier applier;
    private final Metrics<?> metrics;
    private final String errorCounter;
    private final CheckpointApplier checkpointApplier;
    private final WebServer webServer;
    private final AtomicLong checkPointDelay;

    private final Streams<Collection<AugmentedEvent>, Collection<AugmentedEvent>> destinationStream;
    private final Streams<RawEvent, Collection<AugmentedEvent>> sourceStream;

    private final String METRIC_COORDINATOR_DELAY               = MetricRegistry.name("coordinator", "delay");
    private final String METRIC_STREAM_DESTINATION_QUEUE_SIZE   = MetricRegistry.name("streams", "destination", "queue", "size");
    private final String METRIC_STREAM_SOURCE_QUEUE_SIZE        = MetricRegistry.name("streams", "source", "queue", "size");

    public Replicator(final Map<String, Object> configuration) {

        Object checkpointPath = configuration.get(Configuration.CHECKPOINT_PATH);
        Object checkpointDefault = configuration.get(Configuration.CHECKPOINT_DEFAULT);

        Objects.requireNonNull(checkpointPath, String.format("Configuration required: %s", Configuration.CHECKPOINT_PATH));

        int threads = Integer.parseInt(configuration.getOrDefault(Configuration.REPLICATOR_THREADS, "1").toString());
        int tasks = Integer.parseInt(configuration.getOrDefault(Configuration.REPLICATOR_TASKS, "1").toString());

        boolean overrideCheckpointStartPosition = Boolean.parseBoolean(configuration.getOrDefault(Configuration.OVERRIDE_CHECKPOINT_START_POSITION, false).toString());
        String overrideCheckpointBinLogFileName = configuration.getOrDefault(Configuration.OVERRIDE_CHECKPOINT_BINLOG_FILENAME, "").toString();
        long overrideCheckpointBinlogPosition   = Long.parseLong(configuration.getOrDefault(Configuration.OVERRIDE_CHECKPOINT_BINLOG_POSITION, "0").toString());
        String overrideCheckpointGtidSet        = configuration.getOrDefault(Configuration.OVERRIDE_CHECKPOINT_GTID_SET, "").toString();


        this.checkpointPath = checkpointPath.toString();

        this.checkpointDefault = (checkpointDefault != null) ? (checkpointDefault.toString()) : (null);

        this.webServer = WebServer.build(configuration);

        this.metrics = Metrics.build(configuration, webServer.getServer());

        this.errorCounter = MetricRegistry.name(
                String.valueOf(configuration.getOrDefault(Metrics.Configuration.BASE_PATH, "replicator")),
                "error"
        );

        this.coordinator        = Coordinator.build(configuration);
        this.supplier           = Supplier.build(configuration);
        this.augmenter          = Augmenter.build(configuration);
        this.augmenterFilter    = AugmenterFilter.build(configuration);
        this.seeker             = Seeker.build(configuration);
        this.partitioner        = Partitioner.build(configuration);
        this.applier            = Applier.build(configuration);

        this.checkPointDelay = new AtomicLong(0L);

        this.checkpointApplier = CheckpointApplier.build(configuration,
                this.coordinator,
                this.checkpointPath,
            safeCheckpoint -> this.checkPointDelay.set((System.currentTimeMillis() - safeCheckpoint.getTimestamp()) / 1000)
        );

        this.metrics.register(METRIC_COORDINATOR_DELAY, (Gauge<Long>) () -> this.checkPointDelay.get());

        // --------------------------------------------------------------------
        // Setup streams/pipelines:
        //
        // Notes:
        //
        //  Supplying the input data has three modes:
        //      1. Short Circuit Push:
        //          - Input data for this stream/data-pipeline will be provided by
        //            by calling the push() method of the pipeline instance, which will not
        //            use any queues but will directly pass the item to the process()
        //            method.
        //      2. Queued Push:
        //          - If pipeline has a queue then the push() method will default to
        //            adding items to that queue and then internal consumer is automatically
        //            initialized as the poller of that queue
        //      3. Pull:<commit_msg>If the checkPointApplier never been called, then checkPointDelay retains the last delayed value. Fixing the bug. Store the last safecheckpoint value in variable. Compute the delay when the metric register collect metric<commit_after>package com.booking.replication;

import com.booking.replication.applier.Applier;
import com.booking.replication.applier.Partitioner;
import com.booking.replication.applier.Seeker;
import com.booking.replication.augmenter.Augmenter;
import com.booking.replication.augmenter.AugmenterFilter;
import com.booking.replication.augmenter.model.event.AugmentedEvent;
import com.booking.replication.checkpoint.CheckpointApplier;
import com.booking.replication.commons.checkpoint.Binlog;
import com.booking.replication.commons.checkpoint.Checkpoint;
import com.booking.replication.commons.checkpoint.ForceRewindException;
import com.booking.replication.commons.map.MapFlatter;
import com.booking.replication.commons.metrics.Metrics;
import com.booking.replication.controller.WebServer;
import com.booking.replication.coordinator.Coordinator;
import com.booking.replication.streams.Streams;
import com.booking.replication.supplier.Supplier;
import com.booking.replication.supplier.model.RawEvent;
import com.booking.utils.BootstrapReplicator;

import com.codahale.metrics.Gauge;
import com.codahale.metrics.MetricRegistry;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;

import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.DefaultParser;
import org.apache.commons.cli.HelpFormatter;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;

public class Replicator {

    public interface Configuration {
        String CHECKPOINT_PATH      = "checkpoint.path";
        String CHECKPOINT_DEFAULT   = "checkpoint.default";
        String REPLICATOR_THREADS   = "replicator.threads";
        String REPLICATOR_TASKS     = "replicator.tasks";

        String OVERRIDE_CHECKPOINT_START_POSITION   = "override.checkpoint.start.position";
        String OVERRIDE_CHECKPOINT_BINLOG_FILENAME  = "override.checkpoint.binLog.filename";
        String OVERRIDE_CHECKPOINT_BINLOG_POSITION  = "override.checkpoint.binLog.position";
        String OVERRIDE_CHECKPOINT_GTID_SET         = "override.checkpoint.gtidSet";
    }

    private static final Logger LOG = LogManager.getLogger(Replicator.class);
    private static final String COMMAND_LINE_SYNTAX = "java -jar mysql-replicator-<version>.jar";

    private final String checkpointPath;
    private final String checkpointDefault;
    private final Coordinator coordinator;
    private final Supplier supplier;
    private final Augmenter augmenter;
    private final AugmenterFilter augmenterFilter;
    private final Seeker seeker;
    private final Partitioner partitioner;
    private final Applier applier;
    private final Metrics<?> metrics;
    private final String errorCounter;
    private final CheckpointApplier checkpointApplier;
    private final WebServer webServer;
    private final AtomicLong lastSafeCheckpoint;

    private final Streams<Collection<AugmentedEvent>, Collection<AugmentedEvent>> destinationStream;
    private final Streams<RawEvent, Collection<AugmentedEvent>> sourceStream;

    private final String METRIC_COORDINATOR_DELAY               = MetricRegistry.name("coordinator", "delay");
    private final String METRIC_STREAM_DESTINATION_QUEUE_SIZE   = MetricRegistry.name("streams", "destination", "queue", "size");
    private final String METRIC_STREAM_SOURCE_QUEUE_SIZE        = MetricRegistry.name("streams", "source", "queue", "size");

    public Replicator(final Map<String, Object> configuration) {

        Object checkpointPath = configuration.get(Configuration.CHECKPOINT_PATH);
        Object checkpointDefault = configuration.get(Configuration.CHECKPOINT_DEFAULT);

        Objects.requireNonNull(checkpointPath, String.format("Configuration required: %s", Configuration.CHECKPOINT_PATH));

        int threads = Integer.parseInt(configuration.getOrDefault(Configuration.REPLICATOR_THREADS, "1").toString());
        int tasks = Integer.parseInt(configuration.getOrDefault(Configuration.REPLICATOR_TASKS, "1").toString());

        boolean overrideCheckpointStartPosition = Boolean.parseBoolean(configuration.getOrDefault(Configuration.OVERRIDE_CHECKPOINT_START_POSITION, false).toString());
        String overrideCheckpointBinLogFileName = configuration.getOrDefault(Configuration.OVERRIDE_CHECKPOINT_BINLOG_FILENAME, "").toString();
        long overrideCheckpointBinlogPosition   = Long.parseLong(configuration.getOrDefault(Configuration.OVERRIDE_CHECKPOINT_BINLOG_POSITION, "0").toString());
        String overrideCheckpointGtidSet        = configuration.getOrDefault(Configuration.OVERRIDE_CHECKPOINT_GTID_SET, "").toString();


        this.checkpointPath = checkpointPath.toString();

        this.checkpointDefault = (checkpointDefault != null) ? (checkpointDefault.toString()) : (null);

        this.webServer = WebServer.build(configuration);

        this.metrics = Metrics.build(configuration, webServer.getServer());

        this.errorCounter = MetricRegistry.name(
                String.valueOf(configuration.getOrDefault(Metrics.Configuration.BASE_PATH, "replicator")),
                "error"
        );

        this.coordinator        = Coordinator.build(configuration);
        this.supplier           = Supplier.build(configuration);
        this.augmenter          = Augmenter.build(configuration);
        this.augmenterFilter    = AugmenterFilter.build(configuration);
        this.seeker             = Seeker.build(configuration);
        this.partitioner        = Partitioner.build(configuration);
        this.applier            = Applier.build(configuration);

        this.lastSafeCheckpoint = new AtomicLong(0L);

        this.checkpointApplier = CheckpointApplier.build(configuration,
                this.coordinator,
                this.checkpointPath,
                safeCheckpoint -> this.lastSafeCheckpoint.set(safeCheckpoint.getTimestamp())
        );

        this.metrics.register(METRIC_COORDINATOR_DELAY, (Gauge<Long>) () -> (System.currentTimeMillis() - this.lastSafeCheckpoint.get()) / 1000);

        // --------------------------------------------------------------------
        // Setup streams/pipelines:
        //
        // Notes:
        //
        //  Supplying the input data has three modes:
        //      1. Short Circuit Push:
        //          - Input data for this stream/data-pipeline will be provided by
        //            by calling the push() method of the pipeline instance, which will not
        //            use any queues but will directly pass the item to the process()
        //            method.
        //      2. Queued Push:
        //          - If pipeline has a queue then the push() method will default to
        //            adding items to that queue and then internal consumer is automatically
        //            initialized as the poller of that queue
        //      3. Pull:
{'fraction_non_alphanumeric': 0.09745591600484588, 'fraction_numerical': 0.0017498990442859066, 'mean_word_length': 3.5454267360048943, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 6}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6890480', 'n_tokens_mistral': 4042, 'n_tokens_neox': 3900, 'n_words': 794}