<commit_before>package org.apache.flume.source;

import org.apache.flume.Channel;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.PollableSource;
import org.apache.flume.Transaction;
import org.apache.flume.event.EventBuilder;

public class SequenceGeneratorSource extends AbstractSource implements
    PollableSource {

  private long sequence;

  public SequenceGeneratorSource() {
    sequence = 0;
  }

  @Override
  public Status process() throws EventDeliveryException {
    Channel channel = getChannel();
    Transaction transaction = channel.getTransaction();

    try {
      transaction.begin();
      channel.put(EventBuilder.withBody(String.valueOf(sequence++).getBytes()));
      transaction.commit();
    } catch (Exception e) {
      transaction.rollback();
    } finally {
      transaction.close();
    }

    return Status.READY;
  }

}<commit_msg>- Added logging and metrics counters to Seq source.<commit_after>package org.apache.flume.source;

import org.apache.flume.Channel;
import org.apache.flume.CounterGroup;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.PollableSource;
import org.apache.flume.Transaction;
import org.apache.flume.event.EventBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class SequenceGeneratorSource extends AbstractSource implements
    PollableSource {

  private static final Logger logger = LoggerFactory
      .getLogger(SequenceGeneratorSource.class);

  private long sequence;
  private CounterGroup counterGroup;

  public SequenceGeneratorSource() {
    sequence = 0;
    counterGroup = new CounterGroup();
  }

  @Override
  public Status process() throws EventDeliveryException {
    Channel channel = getChannel();
    Transaction transaction = channel.getTransaction();

    try {
      transaction.begin();
      channel.put(EventBuilder.withBody(String.valueOf(sequence++).getBytes()));
      transaction.commit();

      counterGroup.incrementAndGet("events.successful");
    } catch (Exception e) {
      transaction.rollback();
      counterGroup.incrementAndGet("events.failed");
    } finally {
      transaction.close();
    }

    return Status.READY;
  }

  @Override
  public void start() {
    logger.info("Sequence generator source starting");

    super.start();

    logger.debug("Sequence generator source started");
  }

  @Override
  public void stop() {
    logger.info("Sequence generator source stopping");

    super.stop();

    logger.info("Sequence generator source stopped. Metrics:{}", counterGroup);
  }

}
{'fraction_non_alphanumeric': 0.104199066874028, 'fraction_numerical': 0.0015552099533437014, 'mean_word_length': 4.005836575875486, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '4743704', 'n_tokens_mistral': 710, 'n_tokens_neox': 682, 'n_words': 157}