Getting topic from metadata

[ThalesCarl]

Hello everyone.

I am new user of librdkafka and kafka in general, therefore if this is a dumb question, let me know. That said, I need a way to get a topic from the metadata.

Suppose I have a broker running with some topics on it and I want to create a new topic. But first, I need to check if the broker already has this topic. I have successfully listed all the topics in my broker using the rd_kafka_metada function and I was able to verify whether the broker has or not the topic I want to create. In this situation, if the topic already has the topic, I want a way to assign this topic to a rd_kafka_topic_t variable.

Here is a MWE of how my code is structured so far (suppose I have already created a producer_ and a topic_conf_ variable, which are member of my class) :

std::string topic = "topic_abc";
const struct rd_kafka_metadata* metadata;
rd_kafka_resp_err_t errstr = rd_kafka_metadata(producer_, 0, NULL, &metadata, 5000);
if(errstr != RD_KAFKA_RESP_ERR_NO_ERROR)
{
    std::cout << "Failed to retrieve metadata: " << std::endl;
}
rd_kafka_topic_t* rkt = NULL;
bool new_topic = true;
for (int i = 0 ; i < metadata->topic_cnt ; i++) {
    const struct rd_kafka_metadata_topic* t = &metadata->topics[i];
    if(t -> topic == topic) {
        std::cout <<"Topic " << topic + " already exists!" << std::endl;
        new_topic = false;
    } 
}

if(new_topic) {
    rkt = (rd_kafka_topic_new(producer_, topic.c_str(), topic_conf_));
} else {
    // Assign here the topic from the metadata to the rkt variable
}

 rd_kafka_metadata_destroy(metadata);

Thanks in advance.
PS: I am using Kafka version 2.7 and librdkafka version 1.5.2 running on a centos7

[ThalesCarl]

@edenhill

[hinxx]

I just go and "re-create" the topics on every app startup. Not sure if that is desired but it seems to work.
This is what I do:

    rd_kafka_conf_res_t ret;
    char errstr[512];

    rd_kafka_t *adminProducer;
    rd_kafka_conf_t *conf = rd_kafka_conf_new();

    // a comma-separated list of host or host:port (default port 9092)
    ret = rd_kafka_conf_set(conf, "bootstrap.servers", mBrokers, errstr, sizeof(errstr));
    if (ret != RD_KAFKA_CONF_OK) {
        E("rd_kafka_conf_set() bootstrap.servers failed: %s\n", errstr);
        rd_kafka_conf_destroy(conf);
        D("<<\n");
        return false;
    }

    adminProducer = rd_kafka_new(RD_KAFKA_PRODUCER, conf, errstr, sizeof(errstr));
    if (! adminProducer) {
        E("rd_kafka_new() failed: %s\n", errstr);
        rd_kafka_conf_destroy(conf);
        D("<<\n");
        return false;
    }

    D("created admin producer %s\n", rd_kafka_name(adminProducer));

    rd_kafka_queue_t *queue = rd_kafka_queue_new(adminProducer);
    // set timeout (optional)
    rd_kafka_AdminOptions_t *options = rd_kafka_AdminOptions_new(adminProducer, RD_KAFKA_ADMIN_OP_CREATETOPICS);
    if (rd_kafka_AdminOptions_set_request_timeout(options, 30 * 1000 /* 30s */, errstr, sizeof(errstr))) {
        E("rd_kafka_AdminOptions_set_request_timeout() failed: %s\n", errstr);
        rd_kafka_destroy(adminProducer);
        adminProducer = NULL;
        D("<<\n");
        return false;
    }

    size_t nrTopics = 5;
    rd_kafka_NewTopic_t *topics[5] = {0};
    topics[0] = rd_kafka_NewTopic_new(mDataTopic, 1, 1, errstr, sizeof(errstr));
    topics[1] = rd_kafka_NewTopic_new(mUserTopic, 1, 1, errstr, sizeof(errstr));
    topics[2] = rd_kafka_NewTopic_new(mStatusTopic, 1, 1, errstr, sizeof(errstr));
    topics[3] = rd_kafka_NewTopic_new(mStoreTopic, 1, 1, errstr, sizeof(errstr));
    topics[4] = rd_kafka_NewTopic_new(mStatisticsTopic, 1, 1, errstr, sizeof(errstr));

    rd_kafka_CreateTopics(adminProducer, topics, nrTopics, options, queue);
    rd_kafka_AdminOptions_destroy(options);
    // wait for results
    rd_kafka_event_t *event = rd_kafka_queue_poll(queue, 30 * 1000 /* 30s */);
    if (! event) {
        rd_kafka_resp_err_t err = rd_kafka_last_error();
        E("rd_kafka_queue_poll() failed: %s\n", rd_kafka_err2str(err));
    } else if (rd_kafka_event_error(event)) {
        E("rd_kafka_CreateTopics() failed: %s\n", rd_kafka_event_error_string(event));
    } else {
        nrTopics = 0;
        const rd_kafka_topic_result_t **result = rd_kafka_CreateTopics_result_topics(event, &nrTopics);
        I("%ld topics created\n", nrTopics);
        for (size_t i = 0; i < nrTopics; i++) {
            I("[%ld] topic %s created\n", i, rd_kafka_topic_result_name(result[i]));
        }
        
    }
    rd_kafka_event_destroy(event);
    rd_kafka_queue_destroy(queue);
    rd_kafka_NewTopic_destroy_array(topics, nrTopics);
    rd_kafka_destroy(adminProducer);