<commit_before>package org.keedio.flume.sink.azure.eventhub;

import java.util.List;

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.sink.AbstractSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Throwables;
import com.microsoft.eventhubs.client.EventHubClient;
import com.microsoft.eventhubs.client.EventHubException;
import com.microsoft.eventhubs.client.EventHubSender;

public class EventHubSink extends AbstractSink implements Configurable {

	private static final Logger logger = LoggerFactory.getLogger(EventHubSink.class);
	
	private EventHubSinkCounter counter;
	private EventHubSinkUtil eventHubSinkUtil;
	EventHubSender sender;
	
	private List<byte[]> messageList;
	
	@Override
	public Status process() throws EventDeliveryException {
		
		Status result = Status.READY;
		Channel channel = getChannel();
		Event event = null;
		Transaction transaction = null;
		
		transaction = channel.getTransaction();
	    transaction.begin();
		
		try{
			/*
			event = channel.take();
			if(event == null) {
	            result = Status.BACKOFF;
	            //counter.incrementBatchEmptyCount();
	          } else {
	        	byte[] eventBody = event.getBody();
	  			sender.send(eventBody);
	            //counter.incrementBatchUnderflowCount();
	          }
			transaction.commit();
			*/
			
			
		    long processedEvents = 0;
			
			for (; processedEvents < eventHubSinkUtil.getBatchSize(); processedEvents += 1) {
		        event = channel.take();
	
		        if (event == null) {
		          // no events available in channel
		          if(processedEvents == 0) {
		            result = Status.BACKOFF;
		            counter.incrementBatchEmptyCount();
		          } else {
		            counter.incrementBatchUnderflowCount();
		          }
		          break;
		        }
		        
		        byte[] eventBody = event.getBody();
		        messageList.add(eventBody);
		    }
			
			// publish batch and commit.
			if (processedEvents > 0) {
				long startTime = System.nanoTime();
				sender.send(messageList);
				long endTime = System.nanoTime();
				counter.addToKafkaEventSendTimer((endTime - startTime)
						/ (1000 * 1000));
				counter.addToEventDrainSuccessCount(Long.valueOf(messageList
						.size()));
			}
			

			transaction.commit();
			
		} catch (Exception ex) {
		      String errorMsg = "Failed to publish events";
		      logger.error("Failed to publish events", ex);
		      result = Status.BACKOFF;
		      if (transaction != null) {
		        try {
		          transaction.rollback();
		          //counter.incrementRollbackCount();
		        } catch (Exception e) {
		          logger.error("Transaction rollback failed", e);
		          throw Throwables.propagate(e);
		        }
		      }
		      throw new EventDeliveryException(errorMsg, ex);
		    } finally {
		      if (transaction != null) {
		        transaction.close();
		      }
		    }
		return result;	
	}

	@Override
	public synchronized void start() {
		// instantiate the producer
		//ProducerConfig config = new ProducerConfig(kafkaProps);
		//producer = new Producer<String, byte[]>(config);
		sender = eventHubSinkUtil.createSender();<commit_msg>Sending events one by one OK<commit_after>package org.keedio.flume.sink.azure.eventhub;

import java.util.List;

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.sink.AbstractSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Throwables;
import com.microsoft.eventhubs.client.EventHubClient;
import com.microsoft.eventhubs.client.EventHubException;
import com.microsoft.eventhubs.client.EventHubSender;

public class EventHubSink extends AbstractSink implements Configurable {

	private static final Logger logger = LoggerFactory.getLogger(EventHubSink.class);
	
	private EventHubSinkCounter counter;
	private EventHubSinkUtil eventHubSinkUtil;
	EventHubSender sender;
	
	private List<byte[]> messageList;
	
	@Override
	public Status process() throws EventDeliveryException {
		
		Status result = Status.READY;
		Channel channel = getChannel();
		Event event = null;
		Transaction transaction = null;
		
		transaction = channel.getTransaction();
	    transaction.begin();
		
		try{
			
			event = channel.take();
			if(event == null) {
	            result = Status.BACKOFF;
	            //counter.incrementBatchEmptyCount();
	          } else {
	        	byte[] eventBody = event.getBody();
	  			sender.send(eventBody);
	            //counter.incrementBatchUnderflowCount();
	          }
			transaction.commit();
			
			
			/*
		    long processedEvents = 0;
			
			for (; processedEvents < eventHubSinkUtil.getBatchSize(); processedEvents += 1) {
		        event = channel.take();
	
		        if (event == null) {
		          // no events available in channel
		          if(processedEvents == 0) {
		            result = Status.BACKOFF;
		            counter.incrementBatchEmptyCount();
		          } else {
		            counter.incrementBatchUnderflowCount();
		          }
		          break;
		        }
		        
		        byte[] eventBody = event.getBody();
		        messageList.add(eventBody);
		    }
			
			// publish batch and commit.
			if (processedEvents > 0) {
				long startTime = System.nanoTime();
				sender.send(messageList);
				long endTime = System.nanoTime();
				counter.addToKafkaEventSendTimer((endTime - startTime)
						/ (1000 * 1000));
				counter.addToEventDrainSuccessCount(Long.valueOf(messageList
						.size()));
			}
			

			transaction.commit();
			*/
		} catch (Exception ex) {
		      String errorMsg = "Failed to publish events";
		      logger.error("Failed to publish events", ex);
		      result = Status.BACKOFF;
		      if (transaction != null) {
		        try {
		          transaction.rollback();
		          //counter.incrementRollbackCount();
		        } catch (Exception e) {
		          logger.error("Transaction rollback failed", e);
		          throw Throwables.propagate(e);
		        }
		      }
		      throw new EventDeliveryException(errorMsg, ex);
		    } finally {
		      if (transaction != null) {
		        transaction.close();
		      }
		    }
		return result;	
	}

	@Override
	public synchronized void start() {
		// instantiate the producer
		//ProducerConfig config = new ProducerConfig(kafkaProps);
		//producer = new Producer<String, byte[]>(config);
		sender = eventHubSinkUtil.createSender();
{'fraction_non_alphanumeric': 0.10518583810783018, 'fraction_numerical': 0.0041134126634347, 'mean_word_length': 2.6097560975609757, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, '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': '4442238', 'n_tokens_mistral': 2229, 'n_tokens_neox': 1868, 'n_words': 406}