I am fairly new to using the rdkafkacpp.h Kafka library for C++.
I referred to a few available online resources, on how to set up a Kafka consumer on Windows, using the standard Kafka environment setup .bat files provided in: https://kafka.apache.org/quickstart (I just played around and was able to test that a message send from the producer terminal came up on the consumer terminal)
I have read through basic theory on Kafka and glanced through the documentation at: https://docs.confluent.io/4.0.0/clients/librdkafka
After this basic reading, I tried writing up sample code using the inbuilt library functions, as shown below. However, I receive a runtime exception at the line of code where I am trying to 'set the bootstrap-server' property of 'conf'. The exception is an access violation exception. "Exception thrown at 0x00007FFB878EC3F9 (msvcr120.dll) in mykafka.exe: 0xC0000005: Access violation reading location 0x0000008E5A900000."
I suspect the order of Kafka consumer 'implementation' using the library functions has missed some steps, or has to be re-ordered.
I am trying to keep the implementation simple, just a single producer on my local machine (localhost:9092), and just this single consumer (mykafka.exe). Also, a topic "quickstart-events" had been started on a terminal.
Any help is greatly appreciated!
P.S: Visual Studio 2019 is used for this code dev.
#include <iostream>
#include "..\include\librdkafka\rdkafkacpp.h"
#include <process.h> // to use exit()
using namespace std;
using namespace RdKafka;
int configAsKafkaConsumer()
{
string host = "localhost:9092";
string errstr;
/* Set properties */
cout << "Inside configAsKafkaConsumer()" << endl;
// Create configuration objects
RdKafka::Conf* conf = NULL;
conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
if (conf != NULL)
{
cout << "conf != NULL" << endl;
}
// THIS IS WHERE I'M GETTING THE RUNTIME EXCEPTION!!
if (conf->set("bootstrap.servers", "localhost:9092", errstr) != RdKafka::Conf::CONF_OK)
{
cerr << "Failed to set config of broker: " << errstr << endl;
exit(1);
}
if (conf->set("client.id", host, errstr) != RdKafka::Conf::CONF_OK)
{
cout << "client.id:" << endl;
exit(1);
}
if (conf->set("group.id", "foo", errstr) != RdKafka::Conf::CONF_OK)
{
cout << "group.id:" << endl;
exit(1);
}
// Create a consumer handle
Consumer* ConsumerHandle = RdKafka::Consumer::create(conf, errstr);
string errstr;
string prodTopic = "quickstart-events";
cout << "Creating topic handle" << endl;
RdKafka::Conf* tconf = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);
// Create topic handle.
RdKafka::Topic* topic = RdKafka::Topic::create(ConsumerHandle, prodTopic,
tconf, errstr);
if (!topic)
{
std::cerr << "Failed to create topic: " << errstr << std::endl;
exit(1);
}
cout << "Starting the consumer handle" << endl;
ConsumerHandle->start(topic, 1, 0);
cout << "Consuming the message" << endl;
Message* msg = ConsumerHandle->consume(topic, 1, 10000);
cout << "Message is: " << msg->payload() << endl;
}
Aucun commentaire:
Enregistrer un commentaire