<commit_before>#include <iostream>

#include "KafkaEventPublisher.h"

/**
 * Set up the configuration for the publisher and initialise it
 *
 * @param broker_str - the IP or hostname of the broker
 * @param topic_str - the name of the datastream (topic) to publish the data to
 */
void KafkaEventPublisher::setUp(const std::string &broker_str,
                                const std::string &topic_str) {

  std::cout << "Setting up Kafka producer" << std::endl;

  std::string error_str;

  RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
  RdKafka::Conf *tconf = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);

  conf->set("metadata.broker.list", broker_str, error_str);
  conf->set("message.max.bytes", "10000000", error_str);
  conf->set("fetch.message.max.bytes", "10000000", error_str);
  conf->set("replica.fetch.max.bytes", "10000000", error_str);
  //if (conf->set("compression.codec", "snappy", error_str) !=
  //    RdKafka::Conf::CONF_OK) {
  //  std::cerr << error_str << std::endl;
  //  exit(1);
  //}

  // Create producer
  m_producer_ptr = std::unique_ptr<RdKafka::Producer>(
      RdKafka::Producer::create(conf, error_str));
  if (!m_producer_ptr.get()) {
    std::cerr << "Failed to create producer: " << error_str << std::endl;
    exit(1);
  }

  // Create topic handle
  m_topic_ptr = std::unique_ptr<RdKafka::Topic>(RdKafka::Topic::create(
      m_producer_ptr.get(), topic_str, tconf, error_str));
  if (!m_topic_ptr.get()) {
    std::cerr << "Failed to create topic: " << error_str << std::endl;
    exit(1);
  }
}

/**
 * Publish the provided message to the datastream
 *
 * @param buf - pointer to the message buffer
 * @param messageSize - the size of the message in bytes
 */
void KafkaEventPublisher::sendMessage(char *buf, size_t messageSize) {
  // using -1 as the partition number will cause rdkafka to distribute messages
  // across multiple partitions to load balance (if the topic has multiple
  // partitions)
  auto resp = m_producer_ptr->produce(
      m_topic_ptr.get(), -1, RdKafka::Producer::RK_MSG_COPY, buf, messageSize,
      NULL, NULL);

  if (resp != RdKafka::ERR_NO_ERROR) {
    std::cerr << "% Produce failed: " << RdKafka::err2str(resp) << std::endl;
    std::cerr << "message size was " << messageSize << std::endl;
  }

  m_producer_ptr->poll(0);
}
<commit_msg>when producer message queue is full, wait 1000ms then try to send same message again<commit_after>#include <iostream>

#include "KafkaEventPublisher.h"

/**
 * Set up the configuration for the publisher and initialise it
 *
 * @param broker_str - the IP or hostname of the broker
 * @param topic_str - the name of the datastream (topic) to publish the data to
 */
void KafkaEventPublisher::setUp(const std::string &broker_str,
                                const std::string &topic_str) {

  std::cout << "Setting up Kafka producer" << std::endl;

  std::string error_str;

  RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
  RdKafka::Conf *tconf = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);

  conf->set("metadata.broker.list", broker_str, error_str);
  conf->set("message.max.bytes", "10000000", error_str);
  conf->set("fetch.message.max.bytes", "10000000", error_str);
  conf->set("replica.fetch.max.bytes", "10000000", error_str);
  //if (conf->set("compression.codec", "snappy", error_str) !=
  //    RdKafka::Conf::CONF_OK) {
  //  std::cerr << error_str << std::endl;
  //  exit(1);
  //}

  // Create producer
  m_producer_ptr = std::unique_ptr<RdKafka::Producer>(
      RdKafka::Producer::create(conf, error_str));
  if (!m_producer_ptr.get()) {
    std::cerr << "Failed to create producer: " << error_str << std::endl;
    exit(1);
  }

  // Create topic handle
  m_topic_ptr = std::unique_ptr<RdKafka::Topic>(RdKafka::Topic::create(
      m_producer_ptr.get(), topic_str, tconf, error_str));
  if (!m_topic_ptr.get()) {
    std::cerr << "Failed to create topic: " << error_str << std::endl;
    exit(1);
  }
}

/**
 * Publish the provided message to the datastream
 *
 * @param buf - pointer to the message buffer
 * @param messageSize - the size of the message in bytes
 */
void KafkaEventPublisher::sendMessage(char *buf, size_t messageSize) {

  RdKafka::ErrorCode resp;
  do {
    // using -1 as the partition number will cause rdkafka to distribute messages
    // across multiple partitions to load balance (if the topic has multiple
    // partitions)
    resp = m_producer_ptr->produce(
        m_topic_ptr.get(), -1, RdKafka::Producer::RK_MSG_COPY, buf, messageSize,
        NULL, NULL);

    if (resp != RdKafka::ERR_NO_ERROR) {
      if (resp != RdKafka::ERR__QUEUE_FULL) {
        std::cerr << "% Produce failed: " << RdKafka::err2str(resp) << std::endl;
        std::cerr << "message size was " << messageSize << std::endl;
      }
      // This blocking poll call should give Kafka some time for the problem to be resolved
      // for example for messages to leave the queue if it is full
      m_producer_ptr->poll(1000);
    } else {
      m_producer_ptr->poll(0);
    }
  } while (resp == RdKafka::ERR__QUEUE_FULL);
}

{'fraction_non_alphanumeric': 0.14853801169590644, 'fraction_numerical': 0.01364522417153996, 'mean_word_length': 3.781919850885368, 'pattern_counts': {'":': 0, '<': 73, '<?xml version=': 0, '>': 24, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 6, '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': '240825', 'n_tokens_mistral': 1808, 'n_tokens_neox': 1721, 'n_words': 431}