KIP 951 - All phases - cherry picked

[anchitj]

No description provided.

[emasab]

Please add this fix too to the PR.
To avoid logging when the id doesn't change and to use rd_kafka_Uuid_base64str instead of rd_kafka_Uuid_str that allocates and leaks memory (and is only for debugging)

                if (rd_kafka_Uuid_cmp(mdit->topic_id, RD_KAFKA_UUID_ZERO) &&
                    rd_kafka_Uuid_cmp(mdit->topic_id, rkt->rkt_topic_id)) {
                        /* FIXME: an offset reset must be triggered.
                         * when rkt_topic_id wasn't zero.
                         * There are no problems
                         * in test 0107_topic_recreate if offsets in new
                         * topic are lower than in previous one,
                         * causing an out of range and an offset reset,
                         * but the rarer case where they're higher needs
                         * to be checked. */
                        rd_kafka_dbg(
                            rk, TOPIC | RD_KAFKA_DBG_METADATA, "METADATA",
                            "Topic %s changed id from %s to %s",
                            rkt->rkt_topic->str,
                            rd_kafka_Uuid_base64str(&rkt->rkt_topic_id),
                            rd_kafka_Uuid_base64str(&mdit->topic_id));
                        rkt->rkt_topic_id = mdit->topic_id;
                }

[emasab]

Please rebase with dependency upgrade commit merged

[emasab]

First comments about the code structure and naming

[emasab]

RD_KAFKA_OP_RES_HANDLED must be returned otherwise it's not destroyed. The decision on the fields to use should be done by the handling code

Suggested change

	rd_kafka_metadata_update_op(rk, rko->rko_u.metadata.mdi);
	res = rd_kafka_metadata_update_op(rk, rko);

[anchitj]

Updated

[anchitj]

I'm returning RD_KAFKA_OP_RES_HANDLED from rd_kafka_metadata_update_op not sure if that's correct

[emasab]

Suggested change

	typedef struct rd_kafkap_Fetch_reply_Partition_s {
	typedef struct rd_kafkap_Fetch_reply_Partition_s {

[anchitj]

Updated to rd_kafkap_Fetch_reply_tags_Partition_s

[emasab]

Suggested change

	mdp->leader;
	mdp->leader;
	rd_kafka_wrunlock(rk);

[emasab]

Add a
rd_kafka_broker_destroy(rkb);
here

[anchitj]

Updated and added rd_kafka_broker_destroy(rkb); after unlock

[emasab]

This needs to be done once per topic, so you can put it outside the for loop you can have a variable
partition_cache_changes that is incremented for each partition and use it in an if outside the partition loop

[emasab]

There the rkb is necessary just for logging purposes, you can use rd_kafka_broker_t *rkb = rk->rk_internal_rkb; in the if after the partition loop

[anchitj]

Added a counter partition_cache_changes which is incremented for each partition cache change, and if the counter> 0 then the rd_kafka_topic_metadata_update2 is called.

[emasab]

Add a
rd_kafka_broker_destroy(rkb);
here

[emasab]

Here it should just populate the topic tags, then the message isn't sent depending on the error code but whenever we have these tags

Suggested change

	if (request->rkbuf_reqhdr.ApiVersion >= 10 &&
	if (request->rkbuf_reqhdr.ApiVersion >= 10) {
	rd_kafkap_Produce_reply_tags_Topic_t *TopicTags =
	&ProduceTags.Topic;
	rd_kafkap_Produce_reply_tags_Partition_t *PartitionTags =
	&TopicTags->Partition;
	/* Partition tags count */
	TopicTags->TopicName = RD_KAFKAP_STR_DUP(&TopicName);
	PartitionTags->Partition = hdr.Partition;
	}

[emasab]

All that comes after must be simpler and without these conditions, this way:

        /* Partition tags */
        rd_kafka_buf_read_tags(rkbuf,
                               rd_kafkap_Produce_reply_tags_partition_parse,
                               &ProduceTags, &ProduceTags.Topic.Partition);

        /* Topic tags */
        rd_kafka_buf_skip_tags(rkbuf);

        if (request->rkbuf_reqhdr.ApiVersion >= 1) {
                int32_t Throttle_Time;
                rd_kafka_buf_read_i32(rkbuf, &Throttle_Time);

                rd_kafka_op_throttle_time(rkb, rkb->rkb_rk->rk_rep,
                                          Throttle_Time);
        }

        /* ProduceResponse tags */
        rd_kafka_buf_read_tags(rkbuf, rd_kafkap_Produce_reply_tags_parse,
                               &ProduceTags);

        rd_kafka_handle_Produce_metadata_update(rkb, &ProduceTags);

        rd_kafkap_Produce_reply_tags_destroy(&ProduceTags);
        return hdr.ErrorCode;

err_parse:
        rd_kafkap_Produce_reply_tags_destroy(&ProduceTags);
        return rkbuf->rkbuf_err;
err:
        rd_kafkap_Produce_reply_tags_destroy(&ProduceTags);
        return RD_KAFKA_RESP_ERR__BAD_MSG;

It's the rd_kafka_handle_Produce_metadata_update that sends an update op if the ProduceTags->leader_change_cnt is non zero.
rd_kafkap_Produce_reply_tags_destroy instead frees the TopicName and NodeEndpoints.NodeEndpoints if not NULL

[anchitj]

Updated. Would do the similar for fetch

[emasab]

Comments about creation of the metadata update operation

[emasab]

Changes about the fetch tags

[emasab]

LGTM! Thanks for the changes