{"action":"created","issue":{"url":"https://api.github.com/repos/edenhill/librdkafka/issues/411","labels_url":"https://api.github.com/repos/edenhill/librdkafka/issues/411/labels{/name}","comments_url":"https://api.github.com/repos/edenhill/librdkafka/issues/411/comments","events_url":"https://api.github.com/repos/edenhill/librdkafka/issues/411/events","html_url":"https://github.com/edenhill/librdkafka/issues/411","id":115642440,"number":411,"title":"Consumed wrong message with unexpected offset ","user":{"login":"jimmy1911","id":2586763,"avatar_url":"https://avatars.githubusercontent.com/u/2586763?v=3","gravatar_id":"","url":"https://api.github.com/users/jimmy1911","html_url":"https://github.com/jimmy1911","followers_url":"https://api.github.com/users/jimmy1911/followers","following_url":"https://api.github.com/users/jimmy1911/following{/other_user}","gists_url":"https://api.github.com/users/jimmy1911/gists{/gist_id}","starred_url":"https://api.github.com/users/jimmy1911/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/jimmy1911/subscriptions","organizations_url":"https://api.github.com/users/jimmy1911/orgs","repos_url":"https://api.github.com/users/jimmy1911/repos","events_url":"https://api.github.com/users/jimmy1911/events{/privacy}","received_events_url":"https://api.github.com/users/jimmy1911/received_events","type":"User","site_admin":false},"labels":[{"url":"https://api.github.com/repos/edenhill/librdkafka/labels/consumer","name":"consumer","color":"009800"}],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":4,"created_at":"2015-11-07T07:05:46Z","updated_at":"2015-11-07T09:42:38Z","closed_at":null,"body":"I have produced 100,000 messages into topic rt_mem01 partiton 0.But when I consume this  partition, kafka return message with unexpected offset. Here are some logs of my program:\r\n\r\ncreate topic offset:15         // consumer->start()   offset\r\nRead msg at offset 15        // message->offset()\r\nmsg\r\n     10.10.38.70rt_mem01*set2\r\nkey_15\r\ncreate topic offset:16            // consume-> start offset\r\nRead msg at offset 15371     // consumed message  with wrong offset \r\nmsg\r\n     10.10.38.70rt_mem01*set2\r\n\r\n     key_15371\r\ncreate topic offset:17\r\nRead msg at offset 17            //next time it is right\r\nmsg\r\n     10.10.38.70rt_mem01*set2\r\nkey_17\r\ncreate topic offset:18\r\nRead msg at offset 15373      //wrong again \r\nmsg\r\n     10.10.38.70rt_mem01*set2\r\n\r\n     key_15373\r\n\r\nIn order to find the problem, I wrote a more simple program to debug and unlucky the result is the same. This progrom creates a topic when consume one message. I don't know is this no appropriate. And when I modified the example/rdkafka_example.cpp in this mode to consume, the result is the same as the log above. My program is shown below: \r\n\r\nint init_kafka(){\r\n    brokers = \"10.10.38.70:9501,10.10.38.70:9502,10.10.38.70:9503\";\r\n    /*\r\n     *Create configuration objects\r\n     */\r\n    consumer_conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL); //global variable\r\n    consumer_tconf = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC); //global variable\r\n    producer_conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL); //global variable\r\n    producer_tconf = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC); //global variable\r\n    consumer_conf->set(\"metadata.broker.list\", FLAGS_brokers, errstr);\r\n    consumer_conf->set(\"queued.min.messages\", \"1000000\", errstr);\r\n    //consumer_conf->set(\"statistics.interval.ms\", \"60*1000\", errstr);\r\n    consumer_conf->set(\"event_cb\", &ex_event_cb, errstr);\r\n\r\n    producer_conf->set(\"metadata.broker.list\", FLAGS_brokers, errstr);\r\n    producer_conf->set(\"queue.buffering.max.messages\", \"1000000\", errstr);\r\n    //producer_conf->set(\"statistics.interval.ms\", \"60*1000\", errstr);\r\n    producer_conf->set(\"event_cb\", &ex_event_cb, errstr);\r\n    producer_conf->set(\"dr_cb\", &ex_dr_cb, errstr);\r\n    producer = RdKafka::Producer::create(producer_conf, errstr);//global variable\r\n    if (!producer) {\r\n        std::cerr << \"Failed to create producer: \" << errstr << std::endl;\r\n        exit(1);\r\n    }\r\n\r\n   consumer = RdKafka::Consumer::create(consumer_conf, errstr);//global variable\r\n    if (!consumer) {\r\n        std::cerr << \"Failed to create consumer: \" << errstr << std::endl;\r\n        exit(1);\r\n    }\r\n   /*only consume 1000 message*/\r\n    for (int64_t k =0;k<1000;k++){\r\n        /*create a topic when consume every message */\r\n        RdKafka::Topic *topic = RdKafka::Topic::create(consumer, string(\"rt_mem01\"),consumer_tconf, errstr);\r\n        if (!topic) {\r\n            std::cerr << \"Failed to create topic: \" << errstr << std::endl;\r\n            exit(1);\r\n        }\r\n        cout<<\"create topic offset:\"<<(int64_t)(k)<<endl;\r\n        RdKafka::ErrorCode resp = consumer->start(topic, 0, (int64_t)(k));\r\n        if (resp != RdKafka::ERR_NO_ERROR) {\r\n            std::cerr << \"Failed to start consumer: \" <<RdKafka::err2str(resp) << std::endl;\r\n            exit(1);\r\n        }        \r\n        RdKafka::Message *msg = consumer->consume(topic, 0, 1000);\r\n        std::cout << \"Read msg at offset \" << msg->offset() << std::endl;\r\n        delete msg;\r\n        consumer->poll(0);\r\n        }\r\n        consumer->stop(topic, 0);\r\n        consumer->poll(1000);\r\n        delete topic;\r\n    }\r\n}\r\n\r\nDid I do something wrong with consume?Or the problem occured when produce message?\r\n"},"comment":{"url":"https://api.github.com/repos/edenhill/librdkafka/issues/comments/154680647","html_url":"https://github.com/edenhill/librdkafka/issues/411#issuecomment-154680647","issue_url":"https://api.github.com/repos/edenhill/librdkafka/issues/411","id":154680647,"user":{"login":"edenhill","id":524990,"avatar_url":"https://avatars.githubusercontent.com/u/524990?v=3","gravatar_id":"","url":"https://api.github.com/users/edenhill","html_url":"https://github.com/edenhill","followers_url":"https://api.github.com/users/edenhill/followers","following_url":"https://api.github.com/users/edenhill/following{/other_user}","gists_url":"https://api.github.com/users/edenhill/gists{/gist_id}","starred_url":"https://api.github.com/users/edenhill/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/edenhill/subscriptions","organizations_url":"https://api.github.com/users/edenhill/orgs","repos_url":"https://api.github.com/users/edenhill/repos","events_url":"https://api.github.com/users/edenhill/events{/privacy}","received_events_url":"https://api.github.com/users/edenhill/received_events","type":"User","site_admin":false},"created_at":"2015-11-07T09:42:38Z","updated_at":"2015-11-07T09:42:38Z","body":"oct15 is in alpha state, but it should compile. Could you provide the compiler output?\r\n"}}