<commit_before>package com.netflix.suro.sink.kafka;

import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.core.type.TypeReference;
import com.google.common.base.Preconditions;
import com.google.common.collect.Maps;
import com.netflix.servo.monitor.Monitors;
import com.netflix.suro.message.Message;
import com.netflix.suro.message.MessageContainer;
import com.netflix.suro.queue.MemoryQueue4Sink;
import com.netflix.suro.queue.MessageQueue4Sink;
import com.netflix.suro.sink.Sink;
import com.netflix.suro.sink.ThreadPoolQueuedSink;

import kafka.metrics.KafkaMetricsReporter$;
import kafka.producer.*;
import kafka.utils.VerifiableProperties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.Metric;
import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Kafka 0.8.2 Sink, using new Java-native producer, rather than Scala produer.
 * Requests are re-queued indefinitely if they fail.
 * 
 * The configuration parameters for the new kafka producer are listed in:
 * http://kafka.apache.org/documentation.html#newproducerconfigs
 *
 * @author jbae
 * @author starzia
 */
public class KafkaSinkV2 extends ThreadPoolQueuedSink implements Sink {
    public final static String TYPE = "KafkaV2";

    private String clientId;
    private final Map<String, String> keyTopicMap;

    private final KafkaProducer producer;
    private long msgId = 0;
    private AtomicInteger receivedCount = new AtomicInteger(0);
    private AtomicInteger sentCount = new AtomicInteger(0);
    private AtomicInteger sentByteCount = new AtomicInteger(0);
    /** number of times a message send failed without retrying */
    private AtomicInteger droppedCount = new AtomicInteger(0);
    /** number of times a message send failed but was requeued */
    private AtomicInteger requeuedCount = new AtomicInteger(0);


    private final DefaultPartitioner partitioner = new DefaultPartitioner(null); // old Scala partitioner

    @JsonCreator
    public KafkaSinkV2(
            @JsonProperty("queue4Sink") MessageQueue4Sink queue4Sink,
            @JsonProperty("client.id") String clientId,
            @JsonProperty("metadata.broker.list") String bootstrapServers,
            @JsonProperty("compression.codec") String codec,
            @JsonProperty("send.buffer.bytes") int sendBufferBytes,
            @JsonProperty("batchSize") int batchSize,
            @JsonProperty("batchTimeout") int batchTimeout,
            @JsonProperty("request.timeout.ms") int requestTimeout,
            @JsonProperty("kafka.etc") Properties etcProps,
            @JsonProperty("keyTopicMap") Map<String, String> keyTopicMap,
            @JsonProperty("jobQueueSize") int jobQueueSize,
            @JsonProperty("corePoolSize") int corePoolSize,
            @JsonProperty("maxPoolSize") int maxPoolSize,
            @JsonProperty("jobTimeout") long jobTimeout<commit_msg>Changed counters from ints to longs to prevent overflow<commit_after>package com.netflix.suro.sink.kafka;

import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.core.type.TypeReference;
import com.google.common.base.Preconditions;
import com.google.common.collect.Maps;
import com.netflix.servo.monitor.Monitors;
import com.netflix.suro.message.Message;
import com.netflix.suro.message.MessageContainer;
import com.netflix.suro.queue.MemoryQueue4Sink;
import com.netflix.suro.queue.MessageQueue4Sink;
import com.netflix.suro.sink.Sink;
import com.netflix.suro.sink.ThreadPoolQueuedSink;

import kafka.metrics.KafkaMetricsReporter$;
import kafka.producer.*;
import kafka.utils.VerifiableProperties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.Metric;
import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Kafka 0.8.2 Sink, using new Java-native producer, rather than Scala produer.
 * Requests are re-queued indefinitely if they fail.
 * 
 * The configuration parameters for the new kafka producer are listed in:
 * http://kafka.apache.org/documentation.html#newproducerconfigs
 *
 * @author jbae
 * @author starzia
 */
public class KafkaSinkV2 extends ThreadPoolQueuedSink implements Sink {
    public final static String TYPE = "KafkaV2";

    private String clientId;
    private final Map<String, String> keyTopicMap;

    private final KafkaProducer producer;
    private long msgId = 0;
    private AtomicLong receivedCount = new AtomicLong(0);
    private AtomicLong sentCount = new AtomicLong(0);
    private AtomicLong sentByteCount = new AtomicLong(0);
    /** number of times a message send failed without retrying */
    private AtomicLong droppedCount = new AtomicLong(0);
    /** number of times a message send failed but was requeued */
    private AtomicLong requeuedCount = new AtomicLong(0);


    private final DefaultPartitioner partitioner = new DefaultPartitioner(null); // old Scala partitioner

    @JsonCreator
    public KafkaSinkV2(
            @JsonProperty("queue4Sink") MessageQueue4Sink queue4Sink,
            @JsonProperty("client.id") String clientId,
            @JsonProperty("metadata.broker.list") String bootstrapServers,
            @JsonProperty("compression.codec") String codec,
            @JsonProperty("send.buffer.bytes") int sendBufferBytes,
            @JsonProperty("batchSize") int batchSize,
            @JsonProperty("batchTimeout") int batchTimeout,
            @JsonProperty("request.timeout.ms") int requestTimeout,
            @JsonProperty("kafka.etc") Properties etcProps,
            @JsonProperty("keyTopicMap") Map<String, String> keyTopicMap,
            @JsonProperty("jobQueueSize") int jobQueueSize,
            @JsonProperty("corePoolSize") int corePoolSize,
            @JsonProperty("maxPoolSize") int maxPoolSize,
            @JsonProperty("jobTimeout") long jobTimeout
{'fraction_non_alphanumeric': 0.09370816599732262, 'fraction_numerical': 0.0050572661014428085, 'mean_word_length': 4.357768924302789, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, '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': '3873570', 'n_tokens_mistral': 1934, 'n_tokens_neox': 1886, 'n_words': 447}