{"action":"created","issue":{"url":"https://api.github.com/repos/edenhill/librdkafka/issues/2030","repository_url":"https://api.github.com/repos/edenhill/librdkafka","labels_url":"https://api.github.com/repos/edenhill/librdkafka/issues/2030/labels{/name}","comments_url":"https://api.github.com/repos/edenhill/librdkafka/issues/2030/comments","events_url":"https://api.github.com/repos/edenhill/librdkafka/issues/2030/events","html_url":"https://github.com/edenhill/librdkafka/issues/2030","id":365123577,"node_id":"MDU6SXNzdWUzNjUxMjM1Nzc=","number":2030,"title":"the produced message seems nerver sent to server, the outq_len() always none zero and nerver decrease","user":{"login":"yandaren","id":2383743,"node_id":"MDQ6VXNlcjIzODM3NDM=","avatar_url":"https://avatars2.githubusercontent.com/u/2383743?v=4","gravatar_id":"","url":"https://api.github.com/users/yandaren","html_url":"https://github.com/yandaren","followers_url":"https://api.github.com/users/yandaren/followers","following_url":"https://api.github.com/users/yandaren/following{/other_user}","gists_url":"https://api.github.com/users/yandaren/gists{/gist_id}","starred_url":"https://api.github.com/users/yandaren/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/yandaren/subscriptions","organizations_url":"https://api.github.com/users/yandaren/orgs","repos_url":"https://api.github.com/users/yandaren/repos","events_url":"https://api.github.com/users/yandaren/events{/privacy}","received_events_url":"https://api.github.com/users/yandaren/received_events","type":"User","site_admin":false},"labels":[{"id":14337450,"node_id":"MDU6TGFiZWwxNDMzNzQ1MA==","url":"https://api.github.com/repos/edenhill/librdkafka/labels/duplicate","name":"duplicate","color":"cccccc","default":true}],"state":"closed","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":5,"created_at":"2018-09-29T11:18:27Z","updated_at":"2018-10-08T14:56:47Z","closed_at":"2018-10-08T14:28:54Z","author_association":"NONE","body":"Read the FAQ first: https://github.com/edenhill/librdkafka/wiki/FAQ\r\n\r\n\r\n\r\nDescription\r\n===========\r\nin my test code, the topic can create success, and can get metadata success, but can't produce message success, the producer->outq_len() alway same and nerver decease to 0. seems that the message nerver sent to the server, alway in the sent queue.\r\n\r\n```\r\n#include <iostream>\r\n#include <string>\r\n#include <list>\r\n#include <stdint.h>\r\n#include <rdkafkacpp.h>\r\n\r\nstatic bool run = true;\r\nstatic bool exit_eof = false;\r\n\r\nvoid dump_config(RdKafka::Conf* conf) {\r\n    std::list<std::string> *dump = conf->dump();\r\n\r\n    printf(\"config dump(%d):\\n\", (int32_t)dump->size());\r\n    for (auto it = dump->begin(); it != dump->end(); ) {\r\n        std::string name = *it++;\r\n        std::string value = *it++;\r\n        printf(\"%s = %s\\n\", name.c_str(), value.c_str());\r\n    }\r\n\r\n    printf(\"---------------------------------------------\\n\");\r\n}\r\n\r\nclass my_event_cb : public RdKafka::EventCb {\r\npublic:\r\n    void event_cb(RdKafka::Event &event) override {\r\n        switch (event.type())\r\n        {\r\n        case RdKafka::Event::EVENT_ERROR:\r\n            std::cerr << \"ERROR (\" << RdKafka::err2str(event.err()) << \"): \" <<\r\n                event.str() << std::endl;\r\n            if (event.err() == RdKafka::ERR__ALL_BROKERS_DOWN)\r\n                run = false;\r\n            break;\r\n\r\n        case RdKafka::Event::EVENT_STATS:\r\n            std::cerr << \"\\\"STATS\\\": \" << event.str() << std::endl;\r\n            break;\r\n\r\n        case RdKafka::Event::EVENT_LOG:\r\n            fprintf(stderr, \"LOG-%i-%s: %s\\n\",\r\n                event.severity(), event.fac().c_str(), event.str().c_str());\r\n            break;\r\n\r\n        default:\r\n            std::cerr << \"EVENT \" << event.type() <<\r\n                \" (\" << RdKafka::err2str(event.err()) << \"): \" <<\r\n                event.str() << std::endl;\r\n            break;\r\n        }\r\n    }\r\n};\r\n\r\nclass my_hash_partitioner_cb : public RdKafka::PartitionerCb {\r\npublic:\r\n    int32_t partitioner_cb(const RdKafka::Topic *topic, const std::string *key,\r\n        int32_t partition_cnt, void *msg_opaque) override {\r\n        return djb_hash(key->c_str(), key->size()) % partition_cnt;\r\n    }\r\nprivate:\r\n    static inline unsigned int djb_hash(const char *str, size_t len) {\r\n        unsigned int hash = 5381;\r\n        for (size_t i = 0; i < len; i++)\r\n            hash = ((hash << 5) + hash) + str[i];\r\n        return hash;\r\n    }\r\n};\r\n\r\nnamespace producer_ts {\r\n\r\nclass my_delivery_report_cb : public RdKafka::DeliveryReportCb {\r\npublic:\r\n    void dr_cb(RdKafka::Message& message) override {\r\n        printf(\"message delivery %d bytes, error:%s, key: %s\\n\",\r\n            (int32_t)message.len(), message.errstr().c_str(), message.key() ? message.key()->c_str() : \"\");\r\n    }\r\n};\r\n\r\nvoid producer_test() {\r\n    printf(\"producer test\\n\");\r\n\r\n    int32_t partition = RdKafka::Topic::PARTITION_UA;\r\n\r\n    printf(\"input brokers list(127.0.0.1:9092;127.0.0.1:9093;127.0.0.1:9094):\\n\");\r\n    std::string broker_list;\r\n\r\n    //std::cin >> broker_list;\r\n    broker_list = \"127.0.0.1:9092\";\r\n\r\n    printf(\"input partition:\");\r\n\r\n    //std::cin >> partition;\r\n    partition = 0;\r\n\r\n    // config \r\n    RdKafka::Conf* global_conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);\r\n    RdKafka::Conf* topic_conf = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);\r\n\r\n    my_hash_partitioner_cb          hash_partitioner;\r\n    my_event_cb                     event_cb;\r\n    my_delivery_report_cb           delivery_cb;\r\n  \r\n\r\n    std::string err_string;\r\n    if (topic_conf->set(\"partitioner_cb\", &hash_partitioner, err_string) != RdKafka::Conf::CONF_OK) {\r\n        printf(\"set partitioner_cb error: %s\\n\", err_string.c_str());\r\n        return;\r\n    }\r\n\r\n    global_conf->set(\"metadata.broker.list\", broker_list, err_string);\r\n    global_conf->set(\"event_cb\", &event_cb, err_string);\r\n    global_conf->set(\"dr_cb\", &delivery_cb, err_string);\r\n    global_conf->set(\"debug\", \"all\", err_string);\r\n\r\n    //dump_config(global_conf);\r\n    //dump_config(topic_conf);\r\n\r\n\r\n    // create producer\r\n    RdKafka::Producer* producer = RdKafka::Producer::create(global_conf, err_string);\r\n    if (!producer) {\r\n        printf(\"failed to create producer, %s\\n\", err_string.c_str());\r\n        return;\r\n    }\r\n\r\n    printf(\"created producer %s\\n\", producer->name().c_str());\r\n\r\n    std::string topic_name;\r\n    while (true) {\r\n\r\n        printf(\"input topic to create:\\n\");\r\n        std::cin >> topic_name;\r\n\r\n        // create topic\r\n        RdKafka::Topic* topic =\r\n            RdKafka::Topic::create(producer, topic_name, topic_conf, err_string);\r\n\r\n        if (!topic) {\r\n            printf(\"try create topic[%s] failed, %s\\n\",\r\n                topic_name.c_str(), err_string.c_str());\r\n            return;\r\n        }\r\n\r\n        printf(\">\");\r\n        for (std::string line; run && std::getline(std::cin, line); ) {\r\n            if (line.empty()) {\r\n                producer->poll(0);\r\n                continue;\r\n            }\r\n\r\n            if (line == \"quit\") {\r\n                break;\r\n            }\r\n\r\n            RdKafka::ErrorCode res = producer->produce(topic, partition,\r\n                RdKafka::Producer::RK_MSG_COPY,\r\n                (char*)line.c_str(), line.size(), NULL, NULL);\r\n\r\n            if (res != RdKafka::ERR_NO_ERROR) {\r\n                printf(\"produce failed, %s\\n\", RdKafka::err2str(res).c_str());\r\n            }\r\n            else {\r\n                printf(\"produced msg, bytes %d\\n\", (int32_t)line.size());\r\n            }\r\n\r\n            // do socket io\r\n            producer->poll(0);\r\n\r\n            printf(\"outq_len: %d\\n\", producer->outq_len());\r\n\r\n            //producer->flush(1000);\r\n\r\n            //while (run && producer->outq_len()) {\r\n            //    printf(\"wait for write queue( size %d) write finish\\n\", producer->outq_len());\r\n            //    producer->poll(1000);\r\n            //}\r\n\r\n            printf(\">\");\r\n        }\r\n\r\n        delete topic;\r\n\r\n        if (!run) {\r\n            break;\r\n        }\r\n    }\r\n\r\n    run = true;\r\n\r\n    while (run && producer->outq_len()) {\r\n        printf(\"wait for write queue( size %d) write finish\\n\", producer->outq_len());\r\n        producer->poll(1000);\r\n    }\r\n\r\n    delete producer;\r\n}\r\n}\r\n```\r\n================\r\nthe logs like this\r\n```\r\n>1\r\nproduced msg, bytes 1\r\noutq_len: 1\r\n>2\r\nproduced msg, bytes 1\r\noutq_len: 2\r\n>3\r\nproduced msg, bytes 1\r\noutq_len: 3\r\n```\r\n\r\n```\r\nLOG-7-TOPPAR: [thrd:HSH-D-3146.wowshga.internal:9092/0]: xxx:9092/0: kafka_cpp_test [0] 3 message(s) in xmit queue (0 added from partition queue)\r\nLOG-7-TOPPAR: [thrd:HSH-D-3146.wowshga.internal:9092/0]: xxx:9092/0: kafka_cpp_test [0] 3 message(s) in xmit queue (0 added from partition queue)\r\nLOG-7-TOPPAR: [thrd:HSH-D-3146.wowshga.internal:9092/0]: xxx:9092/0: kafka_cpp_test [0] 3 message(s) in xmit queue (0 added from partition queue)\r\n```\r\n\r\nthe outq_len just increase when i send q produce request, and then stay the same.\r\n\r\nChecklist\r\n=========\r\n\r\n**IMPORTANT**: We will close issues where the checklist has not been completed.\r\n\r\nPlease provide the following information:\r\n\r\n - [ ] librdkafka version (release number or git tag): `v0.11.5`\r\n - [ ] Apache Kafka version: `0.11.0.3`\r\n - [ ] librdkafka client configuration: `<REPLACE with e.g., message.timeout.ms=123, auto.reset.offset=earliest, ..>`\r\n - [ ] Operating system: `Win64`\r\n - [ ] Provide logs (with `debug=..` as necessary) from librdkafka\r\n - [ ] Provide broker log excerpts\r\n - [ ] Critical issue\r\n\r\n"},"comment":{"url":"https://api.github.com/repos/edenhill/librdkafka/issues/comments/427867286","html_url":"https://github.com/edenhill/librdkafka/issues/2030#issuecomment-427867286","issue_url":"https://api.github.com/repos/edenhill/librdkafka/issues/2030","id":427867286,"node_id":"MDEyOklzc3VlQ29tbWVudDQyNzg2NzI4Ng==","user":{"login":"yandaren","id":2383743,"node_id":"MDQ6VXNlcjIzODM3NDM=","avatar_url":"https://avatars2.githubusercontent.com/u/2383743?v=4","gravatar_id":"","url":"https://api.github.com/users/yandaren","html_url":"https://github.com/yandaren","followers_url":"https://api.github.com/users/yandaren/followers","following_url":"https://api.github.com/users/yandaren/following{/other_user}","gists_url":"https://api.github.com/users/yandaren/gists{/gist_id}","starred_url":"https://api.github.com/users/yandaren/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/yandaren/subscriptions","organizations_url":"https://api.github.com/users/yandaren/orgs","repos_url":"https://api.github.com/users/yandaren/repos","events_url":"https://api.github.com/users/yandaren/events{/privacy}","received_events_url":"https://api.github.com/users/yandaren/received_events","type":"User","site_admin":false},"created_at":"2018-10-08T14:56:47Z","updated_at":"2018-10-08T14:56:47Z","author_association":"NONE","body":"OK，thanks，i'll try it later"}}
{"actor":{"display_login":"yandaren"}}