<commit_before>package com.booking.replication.pipeline;

import static com.codahale.metrics.MetricRegistry.name;

import com.booking.replication.Configuration;
import com.booking.replication.Constants;
import com.booking.replication.Coordinator;
import com.booking.replication.Metrics;
import com.booking.replication.applier.Applier;
import com.booking.replication.augmenter.AugmentedRowsEvent;
import com.booking.replication.augmenter.AugmentedSchemaChangeEvent;
import com.booking.replication.augmenter.EventAugmenter;
import com.booking.replication.checkpoints.LastVerifiedBinlogFile;
import com.booking.replication.queues.ReplicatorQueues;
import com.booking.replication.schema.HBaseSchemaManager;
import com.booking.replication.schema.TableNameMapper;
import com.booking.replication.schema.exception.TableMapException;

import com.google.code.or.binlog.BinlogEventV4;
import com.google.code.or.binlog.impl.event.*;
import com.google.code.or.common.util.MySQLConstants;
import com.google.common.base.Joiner;

import com.codahale.metrics.Gauge;
import com.codahale.metrics.Meter;
import com.codahale.metrics.MetricRegistry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.URISyntaxException;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.regex.Matcher;
import java.util.regex.Pattern;


/**
 * Pipeline Orchestrator.
 *
 * <p>Manages data flow from event producer into the applier.
 * Also manages persistence of metadata necessary for the replicator features.</p>
 *
 * <p>On each event handles:
 *      1. schema version management
 *      2  augmenting events with schema info
 *      3. sending of events to applier.
 * </p>
 */
public class PipelineOrchestrator extends Thread {

    public  final  Configuration      configuration;
    private final  Applier            applier;
    private final  ReplicatorQueues   queues;
    private static EventAugmenter     eventAugmenter;
    private static HBaseSchemaManager hBaseSchemaManager;

    public CurrentTransactionMetadata currentTransactionMetadata;

    private final ConcurrentHashMap<Integer, BinlogPositionInfo> binlogPositionLastKnownInfo;

    private volatile boolean running = false;

    private volatile boolean replicatorShutdownRequested = false;

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

    private static final Meter XIDCounter           = Metrics.registry.meter(name("events", "XIDCounter"));
    private static final Meter deleteEventCounter   = Metrics.registry.meter(name("events", "deleteEventCounter"));
    private static final Meter insertEventCounter   = Metrics.registry.meter(name("events", "insertEventCounter"));
    private static final Meter commitQueryCounter   = Metrics.registry.meter(name("events", "commitQueryCounter"));
    private static final Meter updateEventCounter   = Metrics.registry.meter(name("events", "updateEventCounter"));
    private static final Meter heartBeatCounter     = Metrics.registry.meter(name("events", "heartBeatCounter"));
    private static final Meter eventsReceivedCounter    = Metrics.registry.meter(name("events", "eventsReceivedCounter"));
    private static final Meter eventsProcessedCounter   = Metrics.registry.meter(name("events", "eventsProcessedCounter"));
    private static final Meter eventsSkippedCounter     = Metrics.registry.meter(name("events", "eventsSkippedCounter"));

    private static final int BUFFER_FLUSH_INTERVAL = 30000; // <- force buffer flush every 30 sec

    private static final int DEFAULT_VERSIONS_FOR_MIRRORED_TABLES = 1000;

    private HashMap<String,Boolean> rotateEventAllreadySeenForBinlogFile = new HashMap<>();

    /**
     * Fake microsecond counter.
     *
     * <p>This is a special feature that
     * requires some explanation</p>
     *
     * <p>MySQL binlog events have second-precision timestamps. This
     * obviously means that we can't have microsecond precision,
     * but that is not the intention here. The idea is to at least
     * preserve the information about ordering of events,
     * especially if one ID has multiple events within the same
     * second. We want to know what was their order. That is the
     * main purpose of this counter.</p>
     */
    private static long fakeMicrosecondCounter = 0;

    public void requestReplicatorShutdown() {
        replicatorShutdownRequested = true;
    }

    public boolean isReplicatorShutdownRequested() {
        return replicatorShutdownRequested;
    }

    public static void setFakeMicrosecondCounter(Long fakeMicrosecondCounter) {
        LOGGER.info(String.format(
                "Setting fake microsecond counter to: %s (was: %s)",
                fakeMicrosecondCounter,
                PipelineOrchestrator.fakeMicrosecondCounter)
        );
        PipelineOrchestrator.fakeMicrosecondCounter = fakeMicrosecondCounter;
    }

    public PipelineOrchestrator(
            ReplicatorQueues                  repQueues,
            ConcurrentHashMap<Integer, BinlogPositionInfo> chm,
            Configuration                     repcfg,
            Applier                           applier
    ) throws SQLException, URISyntaxException {
        queues = repQueues;
        configuration = repcfg;

        eventAugmenter = new EventAugmenter(repcfg);

        currentTransactionMetadata = new CurrentTransactionMetadata();

        if (configuration.getApplierType().equals("hbase")) {
            hBaseSchemaManager = new HBaseSchemaManager(repcfg.getHBaseQuorum());
        }

        this.applier = applier;

        binlogPositionLastKnownInfo = chm;

        LOGGER.info("Created consumer with binlogPositionLastKnownInfo position => { "
                + " binlogFileName => "
                +   binlogPositionLastKnownInfo.get(Constants.LAST_KNOWN_BINLOG_POSITION).getBinlogFilename()
                + ", binlogPosition => "
                +   binlogPositionLastKnownInfo.get(Constants.LAST_KNOWN_BINLOG_POSITION).getBinlogPosition()
                + " }"
        );

        Metrics.registry.register(MetricRegistry.name("events", "replicatorReplicationDelay"),
            new Gauge<Long>() {
                @Override
                public Long getValue() {
                    return replDelay;
                }
            });
    }

    public boolean isRunning() {
        return running;
    }

    public void setRunning(boolean running) {
        this.running = running;
    }

    @Override
    public void run() {
        setRunning(true);

        long timeOfLastEvent = System.currentTimeMillis();

        while (isRunning()) {
            try {
                if (queues.rawQueue.size() > 0) {
                    BinlogEventV4 event =
                            queues.rawQueue.poll(100, TimeUnit.MILLISECONDS);

                    if (event == null) {
                        LOGGER.warn("Poll timeout. Will sleep for 1s and try again.");
                        Thread.sleep(1000);
                        continue;
                    }

                    timeOfLastEvent = System.currentTimeMillis();
                    eventsReceivedCounter.mark();

                    if (!skipEvent(event)) {
                        calculateAndPropagateChanges(event);
                        eventsProcessedCounter.mark();
                    } else {
                        eventsSkippedCounter.mark();
                    }
                } else {
                    LOGGER.info("Pipeline report: no items in producer event rawQueue. Will sleep for 0.5s and check again.");
                    Thread.sleep(500);
                    long currentTime = System.currentTimeMillis();
                    long timeDiff = currentTime - timeOfLastEvent;
                    boolean forceFlush = (timeDiff > BUFFER_FLUSH_INTERVAL);
                    if (forceFlush) {
                        applier.forceFlush();
                    }
                }
            } catch (InterruptedException e) {
                LOGGER.error("InterruptedException, requesting replicator shutdown...", e);
                requestReplicatorShutdown();
            } catch (TableMapException e) {
                LOGGER.error("TableMapException, requesting replicator shutdown...",e);
                requestReplicatorShutdown();
            } catch (IOException e) {
                LOGGER.error("IOException, requesting replicator shutdown...",e);
                requestReplicatorShutdown();
            } catch (Exception e) {
                LOGGER.error("Exception, requesting replicator shutdown...",e);<commit_msg>Move verbose message to debug level.<commit_after>package com.booking.replication.pipeline;

import static com.codahale.metrics.MetricRegistry.name;

import com.booking.replication.Configuration;
import com.booking.replication.Constants;
import com.booking.replication.Coordinator;
import com.booking.replication.Metrics;
import com.booking.replication.applier.Applier;
import com.booking.replication.augmenter.AugmentedRowsEvent;
import com.booking.replication.augmenter.AugmentedSchemaChangeEvent;
import com.booking.replication.augmenter.EventAugmenter;
import com.booking.replication.checkpoints.LastVerifiedBinlogFile;
import com.booking.replication.queues.ReplicatorQueues;
import com.booking.replication.schema.HBaseSchemaManager;
import com.booking.replication.schema.TableNameMapper;
import com.booking.replication.schema.exception.TableMapException;

import com.google.code.or.binlog.BinlogEventV4;
import com.google.code.or.binlog.impl.event.*;
import com.google.code.or.common.util.MySQLConstants;
import com.google.common.base.Joiner;

import com.codahale.metrics.Gauge;
import com.codahale.metrics.Meter;
import com.codahale.metrics.MetricRegistry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.URISyntaxException;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.regex.Matcher;
import java.util.regex.Pattern;


/**
 * Pipeline Orchestrator.
 *
 * <p>Manages data flow from event producer into the applier.
 * Also manages persistence of metadata necessary for the replicator features.</p>
 *
 * <p>On each event handles:
 *      1. schema version management
 *      2  augmenting events with schema info
 *      3. sending of events to applier.
 * </p>
 */
public class PipelineOrchestrator extends Thread {

    public  final  Configuration      configuration;
    private final  Applier            applier;
    private final  ReplicatorQueues   queues;
    private static EventAugmenter     eventAugmenter;
    private static HBaseSchemaManager hBaseSchemaManager;

    public CurrentTransactionMetadata currentTransactionMetadata;

    private final ConcurrentHashMap<Integer, BinlogPositionInfo> binlogPositionLastKnownInfo;

    private volatile boolean running = false;

    private volatile boolean replicatorShutdownRequested = false;

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

    private static final Meter XIDCounter           = Metrics.registry.meter(name("events", "XIDCounter"));
    private static final Meter deleteEventCounter   = Metrics.registry.meter(name("events", "deleteEventCounter"));
    private static final Meter insertEventCounter   = Metrics.registry.meter(name("events", "insertEventCounter"));
    private static final Meter commitQueryCounter   = Metrics.registry.meter(name("events", "commitQueryCounter"));
    private static final Meter updateEventCounter   = Metrics.registry.meter(name("events", "updateEventCounter"));
    private static final Meter heartBeatCounter     = Metrics.registry.meter(name("events", "heartBeatCounter"));
    private static final Meter eventsReceivedCounter    = Metrics.registry.meter(name("events", "eventsReceivedCounter"));
    private static final Meter eventsProcessedCounter   = Metrics.registry.meter(name("events", "eventsProcessedCounter"));
    private static final Meter eventsSkippedCounter     = Metrics.registry.meter(name("events", "eventsSkippedCounter"));

    private static final int BUFFER_FLUSH_INTERVAL = 30000; // <- force buffer flush every 30 sec

    private static final int DEFAULT_VERSIONS_FOR_MIRRORED_TABLES = 1000;

    private HashMap<String,Boolean> rotateEventAllreadySeenForBinlogFile = new HashMap<>();

    /**
     * Fake microsecond counter.
     *
     * <p>This is a special feature that
     * requires some explanation</p>
     *
     * <p>MySQL binlog events have second-precision timestamps. This
     * obviously means that we can't have microsecond precision,
     * but that is not the intention here. The idea is to at least
     * preserve the information about ordering of events,
     * especially if one ID has multiple events within the same
     * second. We want to know what was their order. That is the
     * main purpose of this counter.</p>
     */
    private static long fakeMicrosecondCounter = 0;

    public void requestReplicatorShutdown() {
        replicatorShutdownRequested = true;
    }

    public boolean isReplicatorShutdownRequested() {
        return replicatorShutdownRequested;
    }

    public static void setFakeMicrosecondCounter(Long fakeMicrosecondCounter) {
        LOGGER.info(String.format(
                "Setting fake microsecond counter to: %s (was: %s)",
                fakeMicrosecondCounter,
                PipelineOrchestrator.fakeMicrosecondCounter)
        );
        PipelineOrchestrator.fakeMicrosecondCounter = fakeMicrosecondCounter;
    }

    public PipelineOrchestrator(
            ReplicatorQueues                  repQueues,
            ConcurrentHashMap<Integer, BinlogPositionInfo> chm,
            Configuration                     repcfg,
            Applier                           applier
    ) throws SQLException, URISyntaxException {
        queues = repQueues;
        configuration = repcfg;

        eventAugmenter = new EventAugmenter(repcfg);

        currentTransactionMetadata = new CurrentTransactionMetadata();

        if (configuration.getApplierType().equals("hbase")) {
            hBaseSchemaManager = new HBaseSchemaManager(repcfg.getHBaseQuorum());
        }

        this.applier = applier;

        binlogPositionLastKnownInfo = chm;

        LOGGER.info("Created consumer with binlogPositionLastKnownInfo position => { "
                + " binlogFileName => "
                +   binlogPositionLastKnownInfo.get(Constants.LAST_KNOWN_BINLOG_POSITION).getBinlogFilename()
                + ", binlogPosition => "
                +   binlogPositionLastKnownInfo.get(Constants.LAST_KNOWN_BINLOG_POSITION).getBinlogPosition()
                + " }"
        );

        Metrics.registry.register(MetricRegistry.name("events", "replicatorReplicationDelay"),
            new Gauge<Long>() {
                @Override
                public Long getValue() {
                    return replDelay;
                }
            });
    }

    public boolean isRunning() {
        return running;
    }

    public void setRunning(boolean running) {
        this.running = running;
    }

    @Override
    public void run() {
        setRunning(true);

        long timeOfLastEvent = System.currentTimeMillis();

        while (isRunning()) {
            try {
                if (queues.rawQueue.size() > 0) {
                    BinlogEventV4 event =
                            queues.rawQueue.poll(100, TimeUnit.MILLISECONDS);

                    if (event == null) {
                        LOGGER.warn("Poll timeout. Will sleep for 1s and try again.");
                        Thread.sleep(1000);
                        continue;
                    }

                    timeOfLastEvent = System.currentTimeMillis();
                    eventsReceivedCounter.mark();

                    if (!skipEvent(event)) {
                        calculateAndPropagateChanges(event);
                        eventsProcessedCounter.mark();
                    } else {
                        eventsSkippedCounter.mark();
                    }
                } else {
                    LOGGER.debug("Pipeline report: no items in producer event rawQueue. Will sleep for 0.5s and check again.");
                    Thread.sleep(500);
                    long currentTime = System.currentTimeMillis();
                    long timeDiff = currentTime - timeOfLastEvent;
                    boolean forceFlush = (timeDiff > BUFFER_FLUSH_INTERVAL);
                    if (forceFlush) {
                        applier.forceFlush();
                    }
                }
            } catch (InterruptedException e) {
                LOGGER.error("InterruptedException, requesting replicator shutdown...", e);
                requestReplicatorShutdown();
            } catch (TableMapException e) {
                LOGGER.error("TableMapException, requesting replicator shutdown...",e);
                requestReplicatorShutdown();
            } catch (IOException e) {
                LOGGER.error("IOException, requesting replicator shutdown...",e);
                requestReplicatorShutdown();
            } catch (Exception e) {
                LOGGER.error("Exception, requesting replicator shutdown...",e);
{'fraction_non_alphanumeric': 0.08284667353715791, 'fraction_numerical': 0.0037787701820680177, 'mean_word_length': 2.5962528309656165, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 39, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, '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': '4046457', 'n_tokens_mistral': 4627, 'n_tokens_neox': 4399, 'n_words': 1036}