<commit_before>/*
 * librdkafka - Apache Kafka C library
 *
 * Copyright (c) 2012-2015, <NAME>
 * All rights reserved.
 * 
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met: 
 * 
 * 1. Redistributions of source code must retain the above copyright notice,
 *    this list of conditions and the following disclaimer. 
 * 2. Redistributions in binary form must reproduce the above copyright notice,
 *    this list of conditions and the following disclaimer in the documentation
 *    and/or other materials provided with the distribution. 
 * 
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE 
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 */

#include "rdkafka_int.h"
#include "rdkafka_request.h"
#include "rdkafka_broker.h"
#include "rdkafka_offset.h"
#include "rdkafka_topic.h"
#include "rdkafka_partition.h"


/**
 * Kafka protocol request and response handling.
 * All of this code runs in the broker thread and uses op queues for
 * propagating results back to the various sub-systems operating in
 * other threads.
 */

#define RD_KAFKA_ERR_ACTION_PERMANENT   0x1 /* Permanent error */
#define RD_KAFKA_ERR_ACTION_IGNORE      0x2 /* Error can be ignored */
#define RD_KAFKA_ERR_ACTION_REFRESH     0x4 /* Refresh state (e.g., metadata) */
#define RD_KAFKA_ERR_ACTION_INFORM      0x8 /* Inform application about err */

static int rd_kafka_err_action (rd_kafka_broker_t *rkb,
                                rd_kafka_resp_err_t err,
                                rd_kafka_buf_t *rkbuf,
                                rd_kafka_buf_t *request) {
        int actions = 0;

        if (err && rkb && request)
                rd_rkb_dbg(rkb, BROKER, "REQERR",
                           "%sRequest failed: %s",
                           rd_kafka_ApiKey2str(request->rkbuf_reqhdr.ApiKey),
                           rd_kafka_err2str(err));

        // FIXME: ILLEGAL_GENERATION
        switch (err)
        {
        case RD_KAFKA_RESP_ERR_NO_ERROR:
                break;
        case RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PART:
        case RD_KAFKA_RESP_ERR_LEADER_NOT_AVAILABLE:
        case RD_KAFKA_RESP_ERR_NOT_LEADER_FOR_PARTITION:
        case RD_KAFKA_RESP_ERR_BROKER_NOT_AVAILABLE:
        case RD_KAFKA_RESP_ERR_REPLICA_NOT_AVAILABLE:
        case RD_KAFKA_RESP_ERR_GROUP_COORDINATOR_NOT_AVAILABLE:
        case RD_KAFKA_RESP_ERR_NOT_COORDINATOR_FOR_GROUP:
        case RD_KAFKA_RESP_ERR__WAIT_COORD:
                /* Request metadata information update */
                actions |= RD_KAFKA_ERR_ACTION_REFRESH;
                break;
        case RD_KAFKA_RESP_ERR__DESTROY:
        default:
                actions |= RD_KAFKA_ERR_ACTION_PERMANENT;
                break;
        }

        return actions;
}


static void rd_kafka_assignor_handle_Metadata (rd_kafka_broker_t *rkb,
                                               rd_kafka_resp_err_t err,
                                               rd_kafka_buf_t *rkbuf,
                                               rd_kafka_buf_t *request,
                                               void *opaque);

/**
 * Send GroupCoordinatorRequest
 */
void rd_kafka_GroupCoordinatorRequest (rd_kafka_broker_t *rkb,
                                       const rd_kafkap_str_t *cgrp,
                                       rd_kafka_q_t *replyq,
                                       rd_kafka_resp_cb_t *resp_cb,
                                       void *opaque) {
        rd_kafka_buf_t *rkbuf;

        rkbuf = rd_kafka_buf_new(rkb->rkb_rk, 1, 0);
        rd_kafka_buf_push_kstr(rkbuf, cgrp);

        rd_kafka_broker_buf_enq_replyq(rkb, RD_KAFKAP_GroupCoordinator,
                                       rkbuf, replyq, resp_cb, opaque);
}




/**
 * Parses and handles Offset replies.
 * Returns the parsed Offset in '*Offsetp'.
 * Returns 0 on success, else an error.
 */
static rd_kafka_resp_err_t rd_kafka_handle_Offset (rd_kafka_broker_t *rkb,
                                                   rd_kafka_resp_err_t err,
                                                   rd_kafka_buf_t *rkbuf,
                                                   rd_kafka_buf_t *request,
                                                   rd_kafka_toppar_t *rktp,
                                                   int64_t *Offsetp) {
        const int log_decode_errors = 1;
        int16_t ErrorCode = 0;
        int32_t TopicArrayCnt;
        int64_t Offset = -1;
        int hit = 0;
        int actions;

        if (err) {
                ErrorCode = err;
                goto err;
        }

        rd_kafka_buf_read_i32(rkbuf, &TopicArrayCnt);
        while (TopicArrayCnt-- > 0) {
                rd_kafkap_str_t topic;
                int32_t PartArrayCnt;

                rd_kafka_buf_read_str(rkbuf, &topic);
                rd_kafka_buf_read_i32(rkbuf, &PartArrayCnt);

                while (PartArrayCnt-- > 0) {
                        int32_t partition;
                        int32_t OffsetArrayCnt;
                        int correct_partition;

                        rd_kafka_buf_read_i32(rkbuf, &partition);
                        rd_kafka_buf_read_i16(rkbuf, &ErrorCode);
                        rd_kafka_buf_read_i32(rkbuf, &OffsetArrayCnt);

                        correct_partition =
                                !rd_kafkap_str_cmp(&topic,
                                                   rktp->rktp_rkt->rkt_topic) &&
                                partition == rktp->rktp_partition;

                        while (OffsetArrayCnt-- > 0) {
                                rd_kafka_buf_read_i64(rkbuf, &Offset);

                                if (correct_partition) {
                                        hit = 1;
                                        goto done;
                                }
                        }
                }
        }

err:
        actions = rd_kafka_err_action(rkb, ErrorCode, rkbuf, request);
        if (actions & RD_KAFKA_ERR_ACTION_REFRESH) {
                /* Re-query for coordinator */
                rd_kafka_toppar_lock(rktp);
                rd_kafka_broker_metadata_req(rktp->rktp_leader, 0,
                                             rktp->rktp_rkt, NULL,
                                             "Offset request failed");
                rd_kafka_toppar_unlock(rktp);
                /* Schedule a retry */
                rd_kafka_buf_keep(request);
                rd_kafka_broker_buf_retry(request->rkbuf_rkb, request);
                return RD_KAFKA_RESP_ERR__IN_PROGRESS;
        }

done:
        if (!ErrorCode && !hit)<commit_msg>Dont rely on partition leader broker to be set when querying metadata (issue #518)<commit_after>/*
 * librdkafka - Apache Kafka C library
 *
 * Copyright (c) 2012-2015, <NAME>
 * All rights reserved.
 * 
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met: 
 * 
 * 1. Redistributions of source code must retain the above copyright notice,
 *    this list of conditions and the following disclaimer. 
 * 2. Redistributions in binary form must reproduce the above copyright notice,
 *    this list of conditions and the following disclaimer in the documentation
 *    and/or other materials provided with the distribution. 
 * 
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE 
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 */

#include "rdkafka_int.h"
#include "rdkafka_request.h"
#include "rdkafka_broker.h"
#include "rdkafka_offset.h"
#include "rdkafka_topic.h"
#include "rdkafka_partition.h"


/**
 * Kafka protocol request and response handling.
 * All of this code runs in the broker thread and uses op queues for
 * propagating results back to the various sub-systems operating in
 * other threads.
 */

#define RD_KAFKA_ERR_ACTION_PERMANENT   0x1 /* Permanent error */
#define RD_KAFKA_ERR_ACTION_IGNORE      0x2 /* Error can be ignored */
#define RD_KAFKA_ERR_ACTION_REFRESH     0x4 /* Refresh state (e.g., metadata) */
#define RD_KAFKA_ERR_ACTION_INFORM      0x8 /* Inform application about err */

static int rd_kafka_err_action (rd_kafka_broker_t *rkb,
                                rd_kafka_resp_err_t err,
                                rd_kafka_buf_t *rkbuf,
                                rd_kafka_buf_t *request) {
        int actions = 0;

        if (err && rkb && request)
                rd_rkb_dbg(rkb, BROKER, "REQERR",
                           "%sRequest failed: %s",
                           rd_kafka_ApiKey2str(request->rkbuf_reqhdr.ApiKey),
                           rd_kafka_err2str(err));

        // FIXME: ILLEGAL_GENERATION
        switch (err)
        {
        case RD_KAFKA_RESP_ERR_NO_ERROR:
                break;
        case RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PART:
        case RD_KAFKA_RESP_ERR_LEADER_NOT_AVAILABLE:
        case RD_KAFKA_RESP_ERR_NOT_LEADER_FOR_PARTITION:
        case RD_KAFKA_RESP_ERR_BROKER_NOT_AVAILABLE:
        case RD_KAFKA_RESP_ERR_REPLICA_NOT_AVAILABLE:
        case RD_KAFKA_RESP_ERR_GROUP_COORDINATOR_NOT_AVAILABLE:
        case RD_KAFKA_RESP_ERR_NOT_COORDINATOR_FOR_GROUP:
        case RD_KAFKA_RESP_ERR__WAIT_COORD:
                /* Request metadata information update */
                actions |= RD_KAFKA_ERR_ACTION_REFRESH;
                break;
        case RD_KAFKA_RESP_ERR__DESTROY:
        default:
                actions |= RD_KAFKA_ERR_ACTION_PERMANENT;
                break;
        }

        return actions;
}


static void rd_kafka_assignor_handle_Metadata (rd_kafka_broker_t *rkb,
                                               rd_kafka_resp_err_t err,
                                               rd_kafka_buf_t *rkbuf,
                                               rd_kafka_buf_t *request,
                                               void *opaque);

/**
 * Send GroupCoordinatorRequest
 */
void rd_kafka_GroupCoordinatorRequest (rd_kafka_broker_t *rkb,
                                       const rd_kafkap_str_t *cgrp,
                                       rd_kafka_q_t *replyq,
                                       rd_kafka_resp_cb_t *resp_cb,
                                       void *opaque) {
        rd_kafka_buf_t *rkbuf;

        rkbuf = rd_kafka_buf_new(rkb->rkb_rk, 1, 0);
        rd_kafka_buf_push_kstr(rkbuf, cgrp);

        rd_kafka_broker_buf_enq_replyq(rkb, RD_KAFKAP_GroupCoordinator,
                                       rkbuf, replyq, resp_cb, opaque);
}




/**
 * Parses and handles Offset replies.
 * Returns the parsed Offset in '*Offsetp'.
 * Returns 0 on success, else an error.
 */
static rd_kafka_resp_err_t rd_kafka_handle_Offset (rd_kafka_broker_t *rkb,
                                                   rd_kafka_resp_err_t err,
                                                   rd_kafka_buf_t *rkbuf,
                                                   rd_kafka_buf_t *request,
                                                   rd_kafka_toppar_t *rktp,
                                                   int64_t *Offsetp) {
        const int log_decode_errors = 1;
        int16_t ErrorCode = 0;
        int32_t TopicArrayCnt;
        int64_t Offset = -1;
        int hit = 0;
        int actions;

        if (err) {
                ErrorCode = err;
                goto err;
        }

        rd_kafka_buf_read_i32(rkbuf, &TopicArrayCnt);
        while (TopicArrayCnt-- > 0) {
                rd_kafkap_str_t topic;
                int32_t PartArrayCnt;

                rd_kafka_buf_read_str(rkbuf, &topic);
                rd_kafka_buf_read_i32(rkbuf, &PartArrayCnt);

                while (PartArrayCnt-- > 0) {
                        int32_t partition;
                        int32_t OffsetArrayCnt;
                        int correct_partition;

                        rd_kafka_buf_read_i32(rkbuf, &partition);
                        rd_kafka_buf_read_i16(rkbuf, &ErrorCode);
                        rd_kafka_buf_read_i32(rkbuf, &OffsetArrayCnt);

                        correct_partition =
                                !rd_kafkap_str_cmp(&topic,
                                                   rktp->rktp_rkt->rkt_topic) &&
                                partition == rktp->rktp_partition;

                        while (OffsetArrayCnt-- > 0) {
                                rd_kafka_buf_read_i64(rkbuf, &Offset);

                                if (correct_partition) {
                                        hit = 1;
                                        goto done;
                                }
                        }
                }
        }

err:
        actions = rd_kafka_err_action(rkb, ErrorCode, rkbuf, request);
        if (actions & RD_KAFKA_ERR_ACTION_REFRESH) {
                /* Re-query for leader */
                rd_kafka_topic_leader_query(rktp->rktp_rkt->rkt_rk,
					    rktp->rktp_rkt);
                /* Schedule a retry */
                rd_kafka_buf_keep(request);
                rd_kafka_broker_buf_retry(request->rkbuf_rkb, request);
                return RD_KAFKA_RESP_ERR__IN_PROGRESS;
        }

done:
        if (!ErrorCode && !hit)
{'fraction_non_alphanumeric': 0.06104472516217139, 'fraction_numerical': 0.008193922840559918, 'mean_word_length': 1.6233207952713595, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 28, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 66, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '881907', 'n_tokens_mistral': 4653, 'n_tokens_neox': 4130, 'n_words': 1017}