Implementing a c++ Kafka producer based on librdkafka To practice my c++ I decided to implement a simple Kafka producer to wrap the producer in c in librdkafka library. I just want to get your opinion on the way I have implemented teh default, copy and move constructors and make sure what I'm doing is safe.


Default constructor:


kafka_publisher::kafka_publisher(const std::string &amp;brokers,
                                 const std::string &amp;topic)
{
    m_errstr = new char[512];
    m_config = rd_kafka_conf_new();

    if (rd_kafka_conf_set(m_config,
                          "bootstrap.servers",
                          brokers.c_str(),
                          m_errstr,
                          sizeof(m_errstr)) != RD_KAFKA_CONF_OK) {
        std::string msg = "Failed to initialize Kafka configuration. Caused by ";
        throw std::invalid_argument(msg.append(m_errstr));
    }

    m_publisher = rd_kafka_new(RD_KAFKA_PRODUCER,
                               m_config,
                               m_errstr,
                               sizeof(m_errstr));

    if (!m_publisher) {
        std::string msg = "Failed to initialize Kafka publisher. Caused by ";
        throw std::runtime_error(msg.append(m_errstr));
    }

    m_topic = rd_kafka_topic_new(m_publisher, topic.c_str(), nullptr);

    if (!m_topic) {
        std::string msg = "Failed to initialize Kafka topic. Caused by ";
        throw std::runtime_error(msg.append(rd_kafka_err2str(rd_kafka_last_error())));
    }
}



Copy constructor:


kafka_publisher::kafka_publisher(const kafka_publisher &amp;p)
: m_run(p.m_run),
  m_config(p.m_config),
  m_publisher(p.m_publisher),
  m_topic(p.m_topic)
{
    strcpy(m_errstr, p.m_errstr);
}



Move constructor:


kafka_publisher::kafka_publisher(kafka_publisher &amp;&amp;m)
: m_run(m.m_run),
  m_errstr(m.m_errstr),
  m_config(m.m_config),
  m_publisher(m.m_publisher),
  m_topic(m.m_topic)
{
    m.m_run = 0;
    m.m_errstr = nullptr;
    m.m_config = nullptr;
    m.m_publisher = nullptr;
    m.m_topic = nullptr;
}


In case if it's helpful, the declaration for this class is:

class kafka_publisher {
public:
    kafka_publisher(const std::string &amp;brokers,
                    const std::string &amp;topic);
    kafka_publisher(const kafka_publisher &amp;p);
    kafka_publisher(kafka_publisher &amp;&amp;m);

    kafka_publisher &amp;operator=(const kafka_publisher &amp;p);
    kafka_publisher &amp;operator=(kafka_publisher &amp;&amp;m);

    ~kafka_publisher();

    void publish(std::string &amp;message);
    void stop() { m_run = 0; };

private:
    int m_run = 1;
    char *m_errstr;

    rd_kafka_conf_t *m_config;
    rd_kafka_t *m_publisher;
    rd_kafka_topic_t *m_topic;
};


Really appreciate your time and effort and helping me improve my c++. Thank you.
It would be more useful (and typical) for CodeReview if you'd post the entire code for the class, starting with the class definition and then the method definitions, rather than starting with one of the methods and then two more and then the class definition.

For example, your declaration for

void publish(std::string &amp;message);


looks wrong. If you mean to take message by reference for efficiency, it should be const std::string&amp; message. On the other hand, if it is semantically important that publish is going to modify its argument (an "out-parameter"), it would be traditional to pass the argument by pointer, or (better) return it by value.

std::string s;
kp.publish(s);  // confusing — is `s` modified?
kp.publish(&amp;s);  // OK, informs the caller that something wacky is going on
s = kp.publish();  // best way to return a value


However, I strongly suspect from context that what's going on here is actually that publish wants to take ownership of s's heap-allocated buffer of chars. Ownership transfer is represented in C++ by plain old pass-by-value:

void publish(std::string s);

std::string s;
kp.publish(s);  // publish a copy of `s`, don't modify the original
kp.publish(std::move(s));  // publish `s` and let the original be trashed




The parts of the code you posted don't use m_run at all, so theoretically you could eliminate it.

At least, you should explain why it's an int. We only ever see it assigned the values 0 and 1, which strongly implies to me that m_run ought to be a bool.



Without seeing the code of your destructor, we can't tell whether your copy-constructor and move-constructor are doing the right thing or not.

But from what I know about librdkafka, I am reasonably confident that your copy-constructor is incorrect. Your destructor should be calling rd_kafka_topic_destroy to balance out the call to rd_kafka_topic_new in your constructor; so, if I write

kafka_publisher kp1(brokers, topicname);
kafka_publisher kp2 = kp1;
// destroy kp2
// destroy kp1


the two destructors will each call rd_kafka_topic_destroy, meaning you'll destroy the topic twice; but you only created it once! So you've got a double-free bug here.

My understanding is that librdkafka objects should be handled by unique_ptr in C++. "Copying" a pointer to a librdkafka object (such as a topic) is not generally possible. (Alternatively, you could use shared_ptr; but that seems like overkill.)



kafka_publisher(const std::string &amp;brokers,
                const std::string &amp;topic);


should be marked explicit.



std::string msg = "Failed to initialize Kafka configuration. Caused by ";
throw std::invalid_argument(msg.append(m_errstr));


strikes me as unnecessarily confusing. Try to avoid having multiple side-effects on the same line of C++ code. Either add lines until you have one side-effect per line:

std::string msg = "Failed to initialize Kafka configuration. Caused by ";
msg += m_errstr;
throw std::invalid_argument(msg);


or eliminate side-effects until you have one side-effect per line:

throw std::invalid_argument("Failed to initialize Kafka configuration. Caused by " + m_errstr);


I would also caution that rd_kafka_conf_set can probably fail for plenty of reasons that would not be best described as "invalid argument." In production code you'd probably just wrap up the return value in a custom exception type (throw KafkaException(return_code)) and let the higher-level code deal with it if it wanted to.