<commit_before>// Copyright (c) 2017 Lanceolata

#include <unistd.h>
#include <signal.h>
#include "log2kafka/errmsg_handle.h"
#include "log2kafka/offset_table.h"
#include "log2kafka/topic_conf.h"
#include "log2kafka/produce.h"
#include "log2kafka/inotify.h"
#include "kafka/kafka_producer.h"
#include "util/configparser.h"
#include "easylogging++.h"

// Init logging
INITIALIZE_EASYLOGGINGPP

using namespace log2hdfs;

// Running flag
static bool running = true;
// conf file path
static char *conf_path = NULL;

static std::shared_ptr<ErrmsgHandle> handle;
static std::shared_ptr<OffsetTable> table;
static std::unordered_map<std::string,
    std::shared_ptr<TopicConf>> topic_confs;
static std::unique_ptr<Produce> produce;
static std::unique_ptr<Inotify> inotify;

// logging roll handler
void RolloutHandler(const char* filename, std::size_t size) {
  std::stringstream stream;
  stream << filename << "." << time(NULL);
  rename(filename, stream.str().c_str());
}

// kafka err callback
void err_cb(rd_kafka_t* rk, int err, const char* reason, void* opaque) {
  LOG(WARNING) << "rdkafka error cb name:" << rd_kafka_name(rk)
               << " err:" << rd_kafka_err2str((rd_kafka_resp_err_t)err)
               << " reason:" << reason;
}

// kafka deliver message callback
void dr_msg_cb(rd_kafka_t* rk, const rd_kafka_message_t* rkmessage,
               void* opaque) {
  if (rkmessage->err == 0)
    return;

  // handle produce failed message
  const std::string topic = rd_kafka_topic_name(rkmessage->rkt);
  char *payload = static_cast<char *>(rkmessage->payload);
  size_t len = rkmessage->len;

  std::string msg(payload, len);
  if (handle) {
    handle->ArchiveMsg(topic, msg);
  } else {
    LOG(WARNING) << "dr_msg_cb topic:" << topic << " msg[" << payload << "]";<commit_msg>dr msg callback add warning log<commit_after>// Copyright (c) 2017 Lanceolata

#include <unistd.h>
#include <signal.h>
#include "log2kafka/errmsg_handle.h"
#include "log2kafka/offset_table.h"
#include "log2kafka/topic_conf.h"
#include "log2kafka/produce.h"
#include "log2kafka/inotify.h"
#include "kafka/kafka_producer.h"
#include "util/configparser.h"
#include "easylogging++.h"

// Init logging
INITIALIZE_EASYLOGGINGPP

using namespace log2hdfs;

// Running flag
static bool running = true;
// conf file path
static char *conf_path = NULL;

static std::shared_ptr<ErrmsgHandle> handle;
static std::shared_ptr<OffsetTable> table;
static std::unordered_map<std::string,
    std::shared_ptr<TopicConf>> topic_confs;
static std::unique_ptr<Produce> produce;
static std::unique_ptr<Inotify> inotify;

// logging roll handler
void RolloutHandler(const char* filename, std::size_t size) {
  std::stringstream stream;
  stream << filename << "." << time(NULL);
  rename(filename, stream.str().c_str());
}

// kafka err callback
void err_cb(rd_kafka_t* rk, int err, const char* reason, void* opaque) {
  LOG(WARNING) << "rdkafka error cb name:" << rd_kafka_name(rk)
               << " err:" << rd_kafka_err2str((rd_kafka_resp_err_t)err)
               << " reason:" << reason;
}

// kafka deliver message callback
void dr_msg_cb(rd_kafka_t* rk, const rd_kafka_message_t* rkmessage,
               void* opaque) {
  if (rkmessage->err == 0)
    return;

  // handle produce failed message
  const std::string topic = rd_kafka_topic_name(rkmessage->rkt);
  char *payload = static_cast<char *>(rkmessage->payload);
  size_t len = rkmessage->len;

  std::string msg(payload, len);
  if (handle) {
    handle->ArchiveMsg(topic, msg);
    LOG(WARNING) << "dr_msg_cb topic[" << topic << "] partition["
                 << rkmessage->partition << "] failed with error["
                 << rd_kafka_err2str(rkmessage->err) << "]";
  } else {
    LOG(WARNING) << "dr_msg_cb topic[" << topic << "] msg[" << payload << "]";
{'fraction_non_alphanumeric': 0.1452249408050513, 'fraction_numerical': 0.006577216521967903, 'mean_word_length': 4.400568181818182, 'pattern_counts': {'":': 0, '<': 91, '<?xml version=': 0, '>': 33, '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': '6688763', 'n_tokens_mistral': 1377, 'n_tokens_neox': 1336, 'n_words': 303}