<commit_before>package ru.yandex.opentsdb.flume;

import com.google.common.base.Throwables;
import com.stumbleupon.async.Callback;
import com.stumbleupon.async.Deferred;
import com.yammer.metrics.core.Meter;
import net.opentsdb.core.TSDB;
import net.opentsdb.core.Tags;
import net.opentsdb.core.WritableDataPoints;
import org.apache.flume.Channel;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.Transaction;
import org.apache.flume.conf.Configurable;
import org.apache.flume.instrumentation.ChannelCounter;
import org.apache.flume.instrumentation.SinkCounter;
import org.apache.flume.sink.AbstractSink;
import org.hbase.async.HBaseClient;
import org.hbase.async.HBaseRpc;
import org.hbase.async.PleaseThrottleException;
import org.hbase.async.PutRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Sink, capable to do writes into opentsdb.
 * This version expects full command in event body.
 *
 * @author <NAME>
 */
public class OpenTSDBSink2 extends AbstractSink implements Configurable {

  private final Logger logger = LoggerFactory.getLogger(OpenTSDBSink2.class);
  public static final Charset UTF8 = Charset.forName("UTF-8");

  private SinkCounter sinkCounter;
  private ChannelCounter channelCounter;
  private int batchSize;
  private Semaphore statesPermitted;

  private TSDB tsdb;
  private HBaseClient hbaseClient;
  private String zkquorum;
  private String zkpath;
  private String seriesTable;
  private String uidsTable;
  private Meter pointsCounter;

  /**
   * State object holds result of asynchronous operations.
   * Implements throttle, when hbase can't handle incoming
   * rate of data points.
   * This object should be added as callback to handle throttles.
   */
  private class State implements Callback<Object, Object> {
    volatile boolean throttle = false;
    volatile Exception failure;
    private long stime = System.currentTimeMillis();
    private AtomicInteger inFlight = new AtomicInteger(0);
    private Semaphore signal = new Semaphore(0);
    private final Callback<Void, State> callback;
    private final int count;

    private State(List<EventData> data) throws Exception {
      this(data, null);
    }

    private State(List<EventData> data, Callback<Void, State> callback) throws Exception {
      this.callback = callback;
      this.count = data.size();
      writeDataPoints(data);
    }

    public Object call(final Object arg) throws Exception {
      if (arg instanceof PleaseThrottleException) {
        final PleaseThrottleException e = (PleaseThrottleException) arg;
        logger.warn("Need to throttle, HBase isn't keeping up.", e);
        throttle = true;
        final HBaseRpc rpc = e.getFailedRpc();
        if (rpc instanceof PutRequest) {
          hbaseClient.put((PutRequest) rpc);  // Don't lose edits.
        }
        return null;
      } else if (arg instanceof Exception) {
        int now = inFlight.decrementAndGet();
        if (callback != null && now == 0)
          callback.call(this);
        failure = (Exception) arg;
      } else {
        pointsCounter.mark();
        int now = inFlight.decrementAndGet();
        if (callback != null && now == 0)
          callback.call(this);
      }
      return arg;
    }

    /**
     * Main entry method for adding points
     *
     * @param data points list
     */
    private void writeDataPoints(List<EventData> data) throws Exception {
      if (data.size() < 1)<commit_msg>remove copypaste<commit_after>package ru.yandex.opentsdb.flume;

import com.google.common.base.Throwables;
import com.stumbleupon.async.Callback;
import com.stumbleupon.async.Deferred;
import com.yammer.metrics.core.Meter;
import net.opentsdb.core.TSDB;
import net.opentsdb.core.Tags;
import net.opentsdb.core.WritableDataPoints;
import org.apache.flume.Channel;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.Transaction;
import org.apache.flume.conf.Configurable;
import org.apache.flume.instrumentation.ChannelCounter;
import org.apache.flume.instrumentation.SinkCounter;
import org.apache.flume.sink.AbstractSink;
import org.hbase.async.HBaseClient;
import org.hbase.async.HBaseRpc;
import org.hbase.async.PleaseThrottleException;
import org.hbase.async.PutRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Sink, capable to do writes into opentsdb.
 * This version expects full command in event body.
 *
 * @author <NAME>
 */
public class OpenTSDBSink2 extends AbstractSink implements Configurable {

  private final Logger logger = LoggerFactory.getLogger(OpenTSDBSink2.class);
  public static final Charset UTF8 = Charset.forName("UTF-8");

  private SinkCounter sinkCounter;
  private ChannelCounter channelCounter;
  private int batchSize;
  private Semaphore statesPermitted;

  private TSDB tsdb;
  private HBaseClient hbaseClient;
  private String zkquorum;
  private String zkpath;
  private String seriesTable;
  private String uidsTable;
  private Meter pointsCounter;

  /**
   * State object holds result of asynchronous operations.
   * Implements throttle, when hbase can't handle incoming
   * rate of data points.
   * This object should be added as callback to handle throttles.
   */
  private class State implements Callback<Object, Object> {
    volatile boolean throttle = false;
    volatile Exception failure;
    private long stime = System.currentTimeMillis();
    private AtomicInteger inFlight = new AtomicInteger(0);
    private Semaphore signal = new Semaphore(0);
    private final Callback<Void, State> callback;
    private final int count;

    private State(List<EventData> data) throws Exception {
      this(data, null);
    }

    private State(List<EventData> data, Callback<Void, State> callback) throws Exception {
      this.callback = callback;
      this.count = data.size();
      writeDataPoints(data);
    }

    public Object call(final Object arg) throws Exception {
      if (arg instanceof PleaseThrottleException) {
        final PleaseThrottleException e = (PleaseThrottleException) arg;
        logger.warn("Need to throttle, HBase isn't keeping up.", e);
        throttle = true;
        final HBaseRpc rpc = e.getFailedRpc();
        if (rpc instanceof PutRequest) {
          hbaseClient.put((PutRequest) rpc);  // Don't lose edits.
          return null;
        }
      } else if (arg instanceof Exception) {
        failure = (Exception) arg;
      } else {
        pointsCounter.mark();
      }
      int now = inFlight.decrementAndGet();
      if (callback != null && now == 0)
        callback.call(this);
      return null;
    }

    /**
     * Main entry method for adding points
     *
     * @param data points list
     */
    private void writeDataPoints(List<EventData> data) throws Exception {
      if (data.size() < 1)
{'fraction_non_alphanumeric': 0.09289902280130294, 'fraction_numerical': 0.002736156351791531, 'mean_word_length': 3.7063151440833844, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 17, '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': '7484713', 'n_tokens_mistral': 2291, 'n_tokens_neox': 2269, 'n_words': 615}