<commit_before>package org.elasticsearch.kafka.indexer.jobs;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.WakeupException;
import org.elasticsearch.kafka.indexer.FailedEventsLogger;
import org.elasticsearch.kafka.indexer.exception.IndexerESNotRecoverableException;
import org.elasticsearch.kafka.indexer.exception.IndexerESRecoverableException;
import org.elasticsearch.kafka.indexer.service.IMessageHandler;
import org.elasticsearch.kafka.indexer.service.OffsetLoggingCallbackImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

/**
 * @author marinapopova Apr 13, 2016
 */
public class ConsumerWorker implements Runnable {

	private static final Logger logger = LoggerFactory.getLogger(ConsumerWorker.class);
	private IMessageHandler messageHandler;
	private final KafkaConsumer<String, String> consumer;
	private final String kafkaTopic;
	private final int consumerId;
	// interval in MS to poll Kafka brokers for messages, in case there were no
	// messages during the previous interval
	private long pollIntervalMs;
	private OffsetLoggingCallbackImpl offsetLoggingCallback;

	public ConsumerWorker(int consumerId, String consumerInstanceName, String kafkaTopic, Properties kafkaProperties,
			long pollIntervalMs, IMessageHandler messageHandler) {
		this.messageHandler = messageHandler;
		kafkaProperties.put(ConsumerConfig.CLIENT_ID_CONFIG, consumerInstanceName + "-" + consumerId);

		this.consumerId = consumerId;
		this.kafkaTopic = kafkaTopic;
		this.pollIntervalMs = pollIntervalMs;
		consumer = new KafkaConsumer<>(kafkaProperties);
		offsetLoggingCallback = new OffsetLoggingCallbackImpl();
		logger.info(
				"Created ConsumerWorker with properties: consumerId={}, consumerInstanceName={}, kafkaTopic={}, kafkaProperties={}",
				consumerId, consumerInstanceName, kafkaTopic, kafkaProperties);
	}

	@Override
	public void run() {
		try {
			logger.info("Starting ConsumerWorker, consumerId={}", consumerId);
			consumer.subscribe(Arrays.asList(kafkaTopic), offsetLoggingCallback);
			while (true) {
				boolean isPollFirstRecord = true;
				int numProcessedMessages = 0;
				int numSkippedIndexingMessages = 0;
				int numMessagesInBatch = 0;
				long offsetOfNextBatch = 0;

				logger.debug("consumerId={}; about to call consumer.poll() ...", consumerId);
				ConsumerRecords<String, String> records = consumer.poll(pollIntervalMs);
				Map<Integer, Long> partitionOffsetMap = new HashMap<>();

				// processing messages and adding them to ES batch
				for (ConsumerRecord<String, String> record : records) {
					numMessagesInBatch++;
					Map<String, Object> data = new HashMap<>();
					data.put("partition", record.partition());
					data.put("offset", record.offset());
					data.put("value", record.value());

					logger.debug("consumerId={}; recieved record: {}", consumerId, data);
					if (isPollFirstRecord) {
						isPollFirstRecord = false;
						logger.info("Start offset for partition {} in this poll : {}", record.partition(),
								record.offset());
					}

					try {
						String processedMessage = messageHandler.transformMessage(record.value(), record.offset());
						messageHandler.addMessageToBatch(processedMessage);
						partitionOffsetMap.put(record.partition(), record.offset());
						numProcessedMessages++;
					} catch (Exception e) {
						numSkippedIndexingMessages++;

						logger.error("ERROR processing message {} - skipping it: {}", record.offset(), record.value(),
								e);
						FailedEventsLogger.logFailedToTransformEvent(record.offset(), e.getMessage(), record.value());
					}

				}

				logger.info(
						"Total # of messages in this batch: {}; "
								+ "# of successfully transformed and added to Index: {}; # of skipped from indexing: {}; offsetOfNextBatch: {}",
						numMessagesInBatch, numProcessedMessages, numSkippedIndexingMessages, offsetOfNextBatch);

				// push to ES whole batch
				boolean moveToNextBatch = false;
				if (!records.isEmpty()) {				
					moveToNextBatch = postToElasticSearch();
				}
				<commit_msg>Fix logging for Additional benchmark information<commit_after>package org.elasticsearch.kafka.indexer.jobs;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.WakeupException;
import org.elasticsearch.kafka.indexer.FailedEventsLogger;
import org.elasticsearch.kafka.indexer.exception.IndexerESNotRecoverableException;
import org.elasticsearch.kafka.indexer.exception.IndexerESRecoverableException;
import org.elasticsearch.kafka.indexer.service.IMessageHandler;
import org.elasticsearch.kafka.indexer.service.OffsetLoggingCallbackImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

/**
 * @author marinapopova Apr 13, 2016
 */
public class ConsumerWorker implements Runnable {

	private static final Logger logger = LoggerFactory.getLogger(ConsumerWorker.class);
	private IMessageHandler messageHandler;
	private final KafkaConsumer<String, String> consumer;
	private final String kafkaTopic;
	private final int consumerId;
	// interval in MS to poll Kafka brokers for messages, in case there were no
	// messages during the previous interval
	private long pollIntervalMs;
	private OffsetLoggingCallbackImpl offsetLoggingCallback;

	public ConsumerWorker(int consumerId, String consumerInstanceName, String kafkaTopic, Properties kafkaProperties,
			long pollIntervalMs, IMessageHandler messageHandler) {
		this.messageHandler = messageHandler;
		kafkaProperties.put(ConsumerConfig.CLIENT_ID_CONFIG, consumerInstanceName + "-" + consumerId);

		this.consumerId = consumerId;
		this.kafkaTopic = kafkaTopic;
		this.pollIntervalMs = pollIntervalMs;
		consumer = new KafkaConsumer<>(kafkaProperties);
		offsetLoggingCallback = new OffsetLoggingCallbackImpl();
		logger.info(
				"Created ConsumerWorker with properties: consumerId={}, consumerInstanceName={}, kafkaTopic={}, kafkaProperties={}",
				consumerId, consumerInstanceName, kafkaTopic, kafkaProperties);
	}

	@Override
	public void run() {
		try {
			logger.info("Starting ConsumerWorker, consumerId={}", consumerId);
			consumer.subscribe(Arrays.asList(kafkaTopic), offsetLoggingCallback);
			while (true) {
				boolean isPollFirstRecord = true;
				int numProcessedMessages = 0;
				int numSkippedIndexingMessages = 0;
				int numMessagesInBatch = 0;
				long pollStartMillis = 0L;

				logger.debug("consumerId={}; about to call consumer.poll() ...", consumerId);
				ConsumerRecords<String, String> records = consumer.poll(pollIntervalMs);
				Map<Integer, Long> partitionOffsetMap = new HashMap<>();

				// processing messages and adding them to ES batch
				for (ConsumerRecord<String, String> record : records) {
					numMessagesInBatch++;
					Map<String, Object> data = new HashMap<>();
					data.put("partition", record.partition());
					data.put("offset", record.offset());
					data.put("value", record.value());

					logger.debug("consumerId={}; recieved record: {}", consumerId, data);
					if (isPollFirstRecord) {
						isPollFirstRecord = false;
						logger.info("Start offset for partition {} in this poll : {}", record.partition(), record.offset());
						pollStartMillis = System.currentTimeMillis();
					}

					try {
						String processedMessage = messageHandler.transformMessage(record.value(), record.offset());
						messageHandler.addMessageToBatch(processedMessage);
						partitionOffsetMap.put(record.partition(), record.offset());
						numProcessedMessages++;
					} catch (Exception e) {
						numSkippedIndexingMessages++;

						logger.error("ERROR processing message {} - skipping it: {}", record.offset(), record.value(),
								e);
						FailedEventsLogger.logFailedToTransformEvent(record.offset(), e.getMessage(), record.value());
					}


				}
				long timeBeforePost = System.currentTimeMillis();

				// push to ES whole batch
				boolean moveToNextBatch = false;
				if (!records.isEmpty()) {				
					moveToNextBatch = postToElasticSearch();
					long timeToPost = System.currentTimeMillis() ;
					double perMessageTimeMillis = (double) (timeToPost - pollStartMillis) / numProcessedMessages;
                    logger.info("Previous poll snapshot: total-messages: {}, messages-processed: {}, messages-skipped: {}" +
                                    ", time-to-create-batch: {} ms, time-to-post-millis: {} ms, time-to-post-millis/messages-processed: {} ms"
                            , numMessagesInBatch
							, numProcessedMessages
							, numSkippedIndexingMessages
							, timeBeforePost-pollStartMillis
							, timeToPost-pollStartMillis
							, perMessageTimeMillis) ;
				}
				
{'fraction_non_alphanumeric': 0.10749185667752444, 'fraction_numerical': 0.002695720543636976, 'mean_word_length': 4.455882352941177, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6745017', 'n_tokens_mistral': 2874, 'n_tokens_neox': 2412, 'n_words': 564}