Producer dies when producing to a just created topic #4904
Comments
|
I think I've been running into this exact issue. Although logs are a bit different, the referenced #4905 seems to fix the issue - I've been running my test on repeat for a few dozens of minutes without a crash now, while before I would get a crash in about a minute. Let me attach part of the question that I intended to ask before I found this bug report, since it might help others finding this issue. The topic I create has two partitions and it is assigned to two brokers on creation. I was not able to reproduce the problem with a single-partition topic, but I didn't try very hard since I had case that could be repeated very quickly. For testing, I use small Kafka cluster with four nodes using docker running locally on my development machine. Sometimes, it happens that messages are not written to the topic. This seems to happen randomly. Whenever this happens, a message like the following is logged: I tried to add a while loop calling I was able to workaround the problem temporarily by adding a The essence of the code I use to create the topic follows. I just simplified error checks to keep it more concise, none of the handled errors actually happen. char errstr_[512];
rd_kafka_resp_err_t err_;
// some broker assignments
std::vector< std::int32_t > brokerassignments_{1, 2};
// prepare a topic description with some broker assignment
std::unique_ptr< rd_kafka_queue_t, decltype(&rd_kafka_queue_destroy) > queue_(
rd_kafka_queue_new(rk_),
&rd_kafka_queue_destroy);
std::unique_ptr< rd_kafka_NewTopic_t, decltype(&rd_kafka_NewTopic_destroy) > newtopic_(
rd_kafka_NewTopic_new(
"topic",
2,
-1,
errstr_,
std::size(errstr_)),
&rd_kafka_NewTopic_destroy);
for(std::size_t partitionid_(0); partitionid_ < brokerassignments_.size(); ++partitionid_) {
std::int32_t broker_ids_[] = { brokerassignments_[partitionid_] };
if(rd_kafka_NewTopic_set_replica_assignment(
newtopic_.get(),
static_cast< std::int32_t >(partitionid_),
broker_ids_,
std::size(broker_ids_),
errstr_,
std::size(errstr_)) != != RD_KAFKA_RESP_ERR_NO_ERROR) {
throw std::runtime_error("NewTopic error");
}
}
// create the topic
rd_kafka_NewTopic_t* newtopics_[] = { newtopic_.get() };
rd_kafka_CreateTopics(rk_, newtopics_, std::size(newtopics_), nullptr, queue_.get());
// wait for the result
std::unique_ptr< rd_kafka_event_t, decltype(&rd_kafka_event_destroy) > event_(
rd_kafka_queue_poll(queue_.get(), pimpl->timeout),
&rd_kafka_event_destroy);
if(!event_) {
throw std::runtime_error("timeout");
}
// check the results
if(rd_kafka_event_error(event_.get()) != RD_KAFKA_RESP_ERR_NO_ERROR) {
throw std::runtime_error("CreateTopics error");
}
// success, but we still check that the topic was create successfully just to be sure
const rd_kafka_CreateTopics_result_t* createresult_(rd_kafka_event_CreateTopics_result(event_.get()));
size_t topicnum_;
const rd_kafka_topic_result_t** topicresult_(rd_kafka_CreateTopics_result_topics(createresult_, &topicnum_));
for(std::size_t i_(0); i_ < topicnum_; ++i_) {
if(rd_kafka_topic_result_error(topicresult_[i_]) != RD_KAFKA_RESP_ERR_NO_ERROR) {
throw std::runtime_error("some error");
}
} |
Description
It can occasionally happen, that if we create a topic and immediately after we start producing to that topic the producer can fail with an "unknown topic or partition" message. That problem occurs regardless of
topic.metadata.propagation.max.ms.How to reproduce
testsdirectory, a few of them fail withunknown topic or partitionerror, e.g.:logs:
We can observe, that the state of topic
my-topic-1067changed fromunknowntoexistsand after a second it changed fromexiststonotexists.Checklist
IMPORTANT: We will close issues where the checklist has not been completed.
Please provide the following information:
2.6.03.8.0debug=..as necessary) from librdkafkaubuntu:22.04(x64)The text was updated successfully, but these errors were encountered: