New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Feature/fetch 951 #4631
base: master
Are you sure you want to change the base?
Feature/fetch 951 #4631
Conversation
Completed the tags parsing in the PR #4668 |
src/rdkafka.c
Outdated
new_partition_internals[index].id = mdi->topics[i].partitions[metadata_partition_index].id; | ||
new_partition_internals->leader_epoch = mdi->topics[i].partitions[metadata_partition_index].leader_epoch; | ||
new_partition_internals->racks_cnt = mdi->topics[i].partitions[metadata_partition_index].racks_cnt; | ||
new_partition_internals->racks = rd_malloc( | ||
sizeof(char *) * new_partition_internals->racks_cnt); | ||
RD_LIST_FOREACH(rack, mdi->topics[i].partitions[metadata_partition_index].racks, k) { | ||
new_partition_internals->racks[j] = | ||
rack; /* no duplication */ | ||
} | ||
new_partitions[index].err = mdi->metadata.topics[i].partitions[metadata_partition_index].err; | ||
new_partitions[index].replica_cnt = mdi->metadata.topics[i].partitions[metadata_partition_index].replica_cnt; | ||
new_partitions[index].replicas = mdi->metadata.topics[i].partitions[metadata_partition_index].replicas; | ||
new_partitions[index].id = mdi->metadata.topics[i].partitions[metadata_partition_index].id; | ||
new_partitions[index].isr_cnt = mdi->metadata.topics[i].partitions[metadata_partition_index].isr_cnt; | ||
new_partitions[index].isrs = mdi->metadata.topics[i].partitions[metadata_partition_index].isrs; | ||
new_partitions[index].leader = mdi->metadata.topics[i].partitions[metadata_partition_index].leader; |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We need to merge with the existing info already present for known partitions, can't directly copy the new metadata directly.
new_partitions = rd_malloc(new_partition_cnt*sizeof(rd_kafka_metadata_partition_t)); | ||
cache_partition_index = metadata_partition_index = 0; | ||
for(index=0;index<new_partition_cnt;index++){ | ||
if((metadata_partition_index == mdi->metadata.topics[i].partition_cnt) || ((cache_partition_index != partition_size) && (partition_internals[cache_partition_index]->id < mdi->topics[i].partitions[metadata_partition_index].id))){ |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Can this be simplified?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Code isn't compiling. Can you make sure the code compiles for making the review easier?
int index,metadata_partition_index,cache_partition_index; | ||
char* rack; | ||
|
||
rd_list_t additional_brokers = rd_list_new(10, rd_free); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Should be a pointer. And why was the initial count of 10 chosen?
/* If the topic is in the cache, we need to merge both the information, | ||
* as broker only has updated information for a known partition or complete information for a new partition | ||
* and the cache has complete information of an un-updated partition */ | ||
partition_size = rkmce->rkmce_mtopic.partition_cnt; |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Better to rename to num_partitions or something like that
2275cee
to
31e6c44
Compare
No description provided.