# 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.

1. Default constructor:
kafka_publisher::kafka_publisher(const std::string &brokers,
const std::string &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())));
}
}
1. Copy constructor:
kafka_publisher::kafka_publisher(const kafka_publisher &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);
}
1. Move constructor:
kafka_publisher::kafka_publisher(kafka_publisher &&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 &brokers,
const std::string &topic);
kafka_publisher(const kafka_publisher &p);
kafka_publisher(kafka_publisher &&m);

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

~kafka_publisher();

void publish(std::string &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.

void publish(std::string &message);

looks wrong. If you mean to take message by reference for efficiency, it should be const std::string& 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(&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 &brokers,
const std::string &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.

• When I was looking at the constructor, the m_errstr = new char[512]; in combination with those throws later on screamed "Memory leak!" right into my face. Any thoughts on that? (I personally try to avoid new like hell, so I'm not 100% sure). – AlexV Mar 25 at 18:01
• @Alex: Yes, definitely, memory leak. I hadn't noticed the new at all, during my superficial review. Personally I always forget whether the destructor could fix it up by delete m_errstr in this case, and have to go write a test case every time to remind me that the destructor is not run if the body of a constructor throws. (Destructors of base and member subobjects are run, but the top-level destructor will not have a chance to clean up m_errstr in OP's case.) – Quuxplusone Mar 25 at 18:07
• @Quuxplusone, I can't thank you enough for your time and effort. Pretty much all of your assumptions about the code is right. Rest of the methods I have only sketched and haven't properly implemented yet. I will follow your advice for future reviews. Your line by line explanation specially the semantics are extremely useful. If you come to Singapore, ping me. Beer's on me:) – swdon Mar 26 at 1:52
• @Alex Thank you for catching the m_errstr, I did feel really uncomfortable with that but I wasn't sure exactly what I was doing wrong. Thanks. – swdon Mar 26 at 1:53
• @swdon: I don't really know librdkafka, but since there seems to be a rd_kafka_conf_destroy(rd_kafka_conf_t *conf) it seems reasonable that you have to clean it up yourself. You will have to check back with the doc/examples or ask a question at Stack Overflow. – AlexV Mar 26 at 15:53

I have already discussed this in a comment below Quuxplusone's answer, but I thought I would make this up into an supplementary answer.

The constructor will leak m_errstr = new char[512]; if one of the exceptions is ever thrown. C++ only guarantees that the memory which was allocated for the object itself will be freed on exception. However, this will not cover memory allocated in the construction process itself. As was already mentioned, the destructor is not run if the body of the constructor throws. You can read about it on the ISO C++ FAQ or try it yourself using the example of Quuxplusone.

I also want to point out an observation on the copy constructor. You're using strcpy to transfer the error message to the copied object there. According to the documentation, using strcpy with an dest parameter not large enough to hold the value to copy will result in undefined behavior. Since there is no pre-allocation of m_errstr there, bad bad things may happen to you (up to and including time travel).

Fortunately the documentation also has an example how to fix this issue (code theirs):

#include <iostream>
#include <cstring>
#include <memory>

int main()
{
const char* src = "Take the test.";
//  src[0] = 'M'; // can't modify string literal
auto dst = std::make_unique<char[]>(std::strlen(src)+1); // +1 for the null terminator
std::strcpy(dst.get(), src);
dst[0] = 'M';
std::cout << src << '\n' << dst.get() << '\n';
}

As a bonus it shows how to use a smart pointer with char arrays, which can be a reasonable solution to the memory leak problem.

• Thanks so much Alex, for your help. Special thanks for making the extra effort to link the external sources. Would buy you that beer too :) – swdon Mar 26 at 9:33
• It seems kind of insane that cppreference's example would use make_unique<char[]> instead of, you know, std::string. For OP's purposes, I hope we all agree that std::string is the appropriate way to hold string data. const char *src = "Take the test."; std::string dst = src; dst[0] = 'M'; std::cout << dst << '\n'; (But since this is a bug in OP's completely-wrong copy ctor, maybe the point is moot. :) ) – Quuxplusone Mar 26 at 20:10