<commit_before>package com.booking.replication.applier.hbase;

import com.booking.replication.applier.Applier;
import com.booking.replication.applier.hbase.schema.HBaseRowKeyMapper;
import com.booking.replication.applier.hbase.schema.HBaseSchemaManager;
import com.booking.replication.applier.hbase.schema.SchemaTransitionException;
import com.booking.replication.applier.hbase.writer.HBaseApplierWriter;
import com.booking.replication.applier.hbase.writer.HBaseTimeMachineWriter;
import com.booking.replication.augmenter.model.event.*;
import com.booking.replication.augmenter.model.row.AugmentedRow;
import com.booking.replication.augmenter.model.schema.SchemaSnapshot;
import com.booking.replication.augmenter.util.AugmentedEventRowExtractor;
import com.booking.replication.commons.metrics.Metrics;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.hadoop.hbase.client.Put;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.io.IOException;
import java.security.NoSuchAlgorithmException;
import java.time.Instant;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import static java.util.stream.Collectors.toList;

public class HBaseApplier implements Applier {

    private static final Logger LOG = LogManager.getLogger(com.booking.replication.applier.hbase.HBaseApplier.class);

    private static final ObjectMapper MAPPER = new ObjectMapper();
    private final Metrics<?> metrics;
    private int FLUSH_BUFFER_SIZE       = 1000;
    private int BUFFER_FLUSH_TIME_LIMIT = 30;
    private final boolean dryRun;

    private HBaseSchemaManager hbaseSchemaManager;
    private HBaseApplierWriter hBaseApplierWriter;

    private Map<String, Object> configuration;
    private final StorageConfig storageConfig;

    org.apache.hadoop.conf.Configuration hbaseConfig;

    public interface Configuration {
        String HBASE_ZOOKEEPER_QUORUM   = "applier.hbase.zookeeper.quorum";
        String REPLICATED_SCHEMA_NAME   = "applier.hbase.sourcedb.name";
        String TARGET_NAMESPACE         = "applier.hbase.destination.namespace";
        String SCHEMA_HISTORY_NAMESPACE = "applier.hbase.schema.history.namespace";
        String INITIAL_SNAPSHOT_MODE    = "applier.hbase.initial.snapshot";
        String HBASE_USE_SNAPPY         = "applier.hbase.snappy";
        String DRYRUN                   = "applier.hbase.dryrun";
        String PAYLOAD_TABLE_NAME       = "applier.hbase.payload.table.name";
    }

    @SuppressWarnings("unused")
    public HBaseApplier(Map<String, Object> configuration) {

        this.configuration = configuration;

        this.dryRun = (boolean) configuration.get(Configuration.DRYRUN);

        this.metrics = Metrics.getInstance(configuration);
        this.storageConfig = StorageConfig.build(configuration);

        hbaseConfig = storageConfig.getConfig();

        try {
            hbaseSchemaManager = new HBaseSchemaManager(configuration);
            LOG.info("Created HBaseSchemaManager.");

            hBaseApplierWriter = new HBaseTimeMachineWriter(hbaseConfig, hbaseSchemaManager,configuration);
            LOG.info("Created HBaseApplierWriter.");

        } catch (IOException | NoSuchAlgorithmException e) {
            LOG.error(e.getMessage(), e);
        }

        if (hbaseSchemaManager == null) {
            throw new RuntimeException("Failed to initialize HBaseSchemaManager");
        }

        if (hBaseApplierWriter == null) {
            throw new RuntimeException("Failed to initialize HBaseTimeMachineWriter");
        }

    }



    /**
     * Logic of Operation:
     * <p>
     * The Streams implementation runs multiple threads and will partition the
     * tasks by transaction id.
     * <p>
     * The Augmenter will send grouped transactions to the Applier.apply(). In case of
     * large transactions it will send lists of 1000 rows.
     * <p>
     * The apply() method returns Boolean. If it returns true, this is a signal for
     * Coordinator that rows are committed. In case of small transactions, applier
     * will internally buffer them and apply will return false until buffer is
     * large enough. Once buffer is large enough it will flushThreadBuffer the buffer and
     * return true (on success).
     * <p>
     * In short:
     * - return false means that list of rows is buffered<commit_msg>Introducing new configurations for HBase Buffer flush size and buffer flush time. applier.hbase.buffer.size applier.hbase.buffer.time.limit<commit_after>package com.booking.replication.applier.hbase;

import com.booking.replication.applier.Applier;
import com.booking.replication.applier.hbase.schema.HBaseRowKeyMapper;
import com.booking.replication.applier.hbase.schema.HBaseSchemaManager;
import com.booking.replication.applier.hbase.schema.SchemaTransitionException;
import com.booking.replication.applier.hbase.writer.HBaseApplierWriter;
import com.booking.replication.applier.hbase.writer.HBaseTimeMachineWriter;
import com.booking.replication.augmenter.model.event.*;
import com.booking.replication.augmenter.model.row.AugmentedRow;
import com.booking.replication.augmenter.model.schema.SchemaSnapshot;
import com.booking.replication.augmenter.util.AugmentedEventRowExtractor;
import com.booking.replication.commons.metrics.Metrics;
import com.fasterxml.jackson.databind.ObjectMapper;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.io.IOException;
import java.security.NoSuchAlgorithmException;
import java.time.Instant;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import static java.util.stream.Collectors.toList;

public class HBaseApplier implements Applier {

    private static final Logger LOG = LogManager.getLogger(HBaseApplier.class);

    private static final int DEFAULT_FLUSH_BUFFER_SIZE        = 1000;
    private static final int DEFAULT_BUFFER_FLUSH_TIME_LIMIT  = 30;

    private static final ObjectMapper MAPPER = new ObjectMapper();
    private final Metrics<?> metrics;

    private final int FLUSH_BUFFER_SIZE;
    private final int BUFFER_FLUSH_TIME_LIMIT;

    private final boolean dryRun;

    private HBaseSchemaManager hbaseSchemaManager;
    private HBaseApplierWriter hBaseApplierWriter;

    private Map<String, Object> configuration;
    private final StorageConfig storageConfig;

    org.apache.hadoop.conf.Configuration hbaseConfig;

    public interface Configuration {
        String HBASE_ZOOKEEPER_QUORUM   = "applier.hbase.zookeeper.quorum";
        String REPLICATED_SCHEMA_NAME   = "applier.hbase.sourcedb.name";
        String TARGET_NAMESPACE         = "applier.hbase.destination.namespace";
        String SCHEMA_HISTORY_NAMESPACE = "applier.hbase.schema.history.namespace";
        String INITIAL_SNAPSHOT_MODE    = "applier.hbase.initial.snapshot";
        String HBASE_USE_SNAPPY         = "applier.hbase.snappy";
        String DRYRUN                   = "applier.hbase.dryrun";
        String PAYLOAD_TABLE_NAME       = "applier.hbase.payload.table.name";
        String FLUSH_BUFFER_SIZE        = "applier.hbase.buffer.size";
        String BUFFER_FLUSH_TIME_LIMIT  = "applier.hbase.buffer.time.limit";
    }

    @SuppressWarnings("unused")
    public HBaseApplier(Map<String, Object> configuration) {

        this.configuration = configuration;

        this.dryRun = (boolean) configuration.get(Configuration.DRYRUN);

        if (configuration.containsKey(Configuration.FLUSH_BUFFER_SIZE)) {
            FLUSH_BUFFER_SIZE = (int) configuration.get(Configuration.FLUSH_BUFFER_SIZE);
        }else {
            FLUSH_BUFFER_SIZE = DEFAULT_FLUSH_BUFFER_SIZE;
        }

        LOG.info("HBase FLUSH_BUFFER_SIZE set to " + FLUSH_BUFFER_SIZE);

        if (configuration.containsKey(Configuration.BUFFER_FLUSH_TIME_LIMIT)) {
            BUFFER_FLUSH_TIME_LIMIT= (int) configuration.get(Configuration.BUFFER_FLUSH_TIME_LIMIT);
        }else {
            BUFFER_FLUSH_TIME_LIMIT = DEFAULT_BUFFER_FLUSH_TIME_LIMIT;
        }

        LOG.info("HBase BUFFER_FLUSH_TIME_LIMIT to " + BUFFER_FLUSH_TIME_LIMIT);

        this.metrics = Metrics.getInstance(configuration);
        this.storageConfig = StorageConfig.build(configuration);

        hbaseConfig = storageConfig.getConfig();

        try {
            hbaseSchemaManager = new HBaseSchemaManager(configuration);
            LOG.info("Created HBaseSchemaManager.");

            hBaseApplierWriter = new HBaseTimeMachineWriter(hbaseConfig, hbaseSchemaManager,configuration);
            LOG.info("Created HBaseApplierWriter.");

        } catch (IOException | NoSuchAlgorithmException e) {
            LOG.error(e.getMessage(), e);
        }

        if (hbaseSchemaManager == null) {
            throw new RuntimeException("Failed to initialize HBaseSchemaManager");
        }

        if (hBaseApplierWriter == null) {
            throw new RuntimeException("Failed to initialize HBaseTimeMachineWriter");
        }

    }

    /**
     * Logic of Operation:
     * <p>
     * The Streams implementation runs multiple threads and will partition the
     * tasks by transaction id.
     * <p>
     * The Augmenter will send grouped transactions to the Applier.apply(). In case of
     * large transactions it will send lists of 1000 rows.
     * <p>
     * The apply() method returns Boolean. If it returns true, this is a signal for
     * Coordinator that rows are committed. In case of small transactions, applier
     * will internally buffer them and apply will return false until buffer is
     * large enough. Once buffer is large enough it will flushThreadBuffer the buffer and
     * return true (on success).
     * <p>
     * In short:
     * - return false means that list of rows is buffered
{'fraction_non_alphanumeric': 0.08257252992493407, 'fraction_numerical': 0.002434570906877663, 'mean_word_length': 3.5370455591348366, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, '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': '381577', 'n_tokens_mistral': 2830, 'n_tokens_neox': 2759, 'n_words': 623}