diff options
Diffstat (limited to 'fluent-bit/lib/librdkafka-2.1.0/src/rdkafka_request.c')
| -rw-r--r-- | fluent-bit/lib/librdkafka-2.1.0/src/rdkafka_request.c | 5378 |
1 files changed, 5378 insertions, 0 deletions
diff --git a/fluent-bit/lib/librdkafka-2.1.0/src/rdkafka_request.c b/fluent-bit/lib/librdkafka-2.1.0/src/rdkafka_request.c new file mode 100644 index 000000000..12d9eb30e --- /dev/null +++ b/fluent-bit/lib/librdkafka-2.1.0/src/rdkafka_request.c @@ -0,0 +1,5378 @@ +/* + * librdkafka - Apache Kafka C library + * + * Copyright (c) 2012-2015, Magnus Edenhill + * 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 <stdarg.h> + +#include "rdkafka_int.h" +#include "rdkafka_request.h" +#include "rdkafka_broker.h" +#include "rdkafka_offset.h" +#include "rdkafka_topic.h" +#include "rdkafka_partition.h" +#include "rdkafka_metadata.h" +#include "rdkafka_msgset.h" +#include "rdkafka_idempotence.h" +#include "rdkafka_txnmgr.h" +#include "rdkafka_sasl.h" + +#include "rdrand.h" +#include "rdstring.h" +#include "rdunittest.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. + */ + + +/* RD_KAFKA_ERR_ACTION_.. to string map */ +static const char *rd_kafka_actions_descs[] = { + "Permanent", "Ignore", "Refresh", "Retry", + "Inform", "Special", "MsgNotPersisted", "MsgPossiblyPersisted", + "MsgPersisted", NULL, +}; + +const char *rd_kafka_actions2str(int actions) { + static RD_TLS char actstr[128]; + return rd_flags2str(actstr, sizeof(actstr), rd_kafka_actions_descs, + actions); +} + + +/** + * @brief Decide action(s) to take based on the returned error code. + * + * The optional var-args is a .._ACTION_END terminated list + * of action,error tuples which overrides the general behaviour. + * It is to be read as: for \p error, return \p action(s). + * + * @warning \p request, \p rkbuf and \p rkb may be NULL. + */ +int rd_kafka_err_action(rd_kafka_broker_t *rkb, + rd_kafka_resp_err_t err, + const rd_kafka_buf_t *request, + ...) { + va_list ap; + int actions = 0; + int exp_act; + + if (!err) + return 0; + + /* Match explicitly defined error mappings first. */ + va_start(ap, request); + while ((exp_act = va_arg(ap, int))) { + int exp_err = va_arg(ap, int); + + if (err == exp_err) + actions |= exp_act; + } + va_end(ap); + + /* Explicit error match. */ + if (actions) { + if (err && rkb && request) + rd_rkb_dbg( + rkb, BROKER, "REQERR", + "%sRequest failed: %s: explicit actions %s", + rd_kafka_ApiKey2str(request->rkbuf_reqhdr.ApiKey), + rd_kafka_err2str(err), + rd_kafka_actions2str(actions)); + + return actions; + } + + /* Default error matching */ + switch (err) { + case RD_KAFKA_RESP_ERR_NO_ERROR: + break; + 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_COORDINATOR_NOT_AVAILABLE: + case RD_KAFKA_RESP_ERR_NOT_COORDINATOR: + case RD_KAFKA_RESP_ERR__WAIT_COORD: + /* Request metadata information update */ + actions |= RD_KAFKA_ERR_ACTION_REFRESH | + RD_KAFKA_ERR_ACTION_MSG_NOT_PERSISTED; + break; + + case RD_KAFKA_RESP_ERR_KAFKA_STORAGE_ERROR: + /* Request metadata update and retry */ + actions |= RD_KAFKA_ERR_ACTION_REFRESH | + RD_KAFKA_ERR_ACTION_RETRY | + RD_KAFKA_ERR_ACTION_MSG_NOT_PERSISTED; + break; + + case RD_KAFKA_RESP_ERR__TRANSPORT: + case RD_KAFKA_RESP_ERR__TIMED_OUT: + case RD_KAFKA_RESP_ERR_REQUEST_TIMED_OUT: + case RD_KAFKA_RESP_ERR_NOT_ENOUGH_REPLICAS_AFTER_APPEND: + actions |= RD_KAFKA_ERR_ACTION_RETRY | + RD_KAFKA_ERR_ACTION_MSG_POSSIBLY_PERSISTED; + break; + + case RD_KAFKA_RESP_ERR_NOT_ENOUGH_REPLICAS: + /* Client-side wait-response/in-queue timeout */ + case RD_KAFKA_RESP_ERR__TIMED_OUT_QUEUE: + actions |= RD_KAFKA_ERR_ACTION_RETRY | + RD_KAFKA_ERR_ACTION_MSG_NOT_PERSISTED; + break; + + case RD_KAFKA_RESP_ERR__PURGE_INFLIGHT: + actions |= RD_KAFKA_ERR_ACTION_PERMANENT | + RD_KAFKA_ERR_ACTION_MSG_POSSIBLY_PERSISTED; + break; + + case RD_KAFKA_RESP_ERR__BAD_MSG: + /* Buffer parse failures are typically a client-side bug, + * treat them as permanent failures. */ + actions |= RD_KAFKA_ERR_ACTION_PERMANENT | + RD_KAFKA_ERR_ACTION_MSG_POSSIBLY_PERSISTED; + break; + + case RD_KAFKA_RESP_ERR_COORDINATOR_LOAD_IN_PROGRESS: + actions |= RD_KAFKA_ERR_ACTION_RETRY; + break; + + case RD_KAFKA_RESP_ERR__DESTROY: + case RD_KAFKA_RESP_ERR_INVALID_SESSION_TIMEOUT: + case RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE: + case RD_KAFKA_RESP_ERR__PURGE_QUEUE: + default: + actions |= RD_KAFKA_ERR_ACTION_PERMANENT | + RD_KAFKA_ERR_ACTION_MSG_NOT_PERSISTED; + break; + } + + /* Fatal or permanent errors are not retriable */ + if (actions & + (RD_KAFKA_ERR_ACTION_FATAL | RD_KAFKA_ERR_ACTION_PERMANENT)) + actions &= ~RD_KAFKA_ERR_ACTION_RETRY; + + /* If no request buffer was specified, which might be the case + * in certain error call chains, mask out the retry action. */ + if (!request) + actions &= ~RD_KAFKA_ERR_ACTION_RETRY; + else if (request->rkbuf_reqhdr.ApiKey != RD_KAFKAP_Produce) + /* Mask out message-related bits for non-Produce requests */ + actions &= ~RD_KAFKA_ERR_ACTION_MSG_FLAGS; + + if (err && actions && rkb && request) + rd_rkb_dbg( + rkb, BROKER, "REQERR", "%sRequest failed: %s: actions %s", + rd_kafka_ApiKey2str(request->rkbuf_reqhdr.ApiKey), + rd_kafka_err2str(err), rd_kafka_actions2str(actions)); + + return actions; +} + + +/** + * @brief Read a list of topic+partitions+extra from \p rkbuf. + * + * @param rkbuf buffer to read from + * @param fields An array of fields to read from the buffer and set on + * the rktpar object, in the specified order, must end + * with RD_KAFKA_TOPIC_PARTITION_FIELD_END. + * + * @returns a newly allocated list on success, or NULL on parse error. + */ +rd_kafka_topic_partition_list_t *rd_kafka_buf_read_topic_partitions( + rd_kafka_buf_t *rkbuf, + size_t estimated_part_cnt, + const rd_kafka_topic_partition_field_t *fields) { + const int log_decode_errors = LOG_ERR; + int32_t TopicArrayCnt; + rd_kafka_topic_partition_list_t *parts = NULL; + + rd_kafka_buf_read_arraycnt(rkbuf, &TopicArrayCnt, RD_KAFKAP_TOPICS_MAX); + + parts = rd_kafka_topic_partition_list_new( + RD_MAX(TopicArrayCnt * 4, (int)estimated_part_cnt)); + + while (TopicArrayCnt-- > 0) { + rd_kafkap_str_t kTopic; + int32_t PartArrayCnt; + char *topic; + + rd_kafka_buf_read_str(rkbuf, &kTopic); + rd_kafka_buf_read_arraycnt(rkbuf, &PartArrayCnt, + RD_KAFKAP_PARTITIONS_MAX); + + RD_KAFKAP_STR_DUPA(&topic, &kTopic); + + while (PartArrayCnt-- > 0) { + int32_t Partition = -1, Epoch = -1234, + CurrentLeaderEpoch = -1234; + int64_t Offset = -1234; + int16_t ErrorCode = 0; + rd_kafka_topic_partition_t *rktpar; + int fi; + + /* + * Read requested fields + */ + for (fi = 0; + fields[fi] != RD_KAFKA_TOPIC_PARTITION_FIELD_END; + fi++) { + switch (fields[fi]) { + case RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION: + rd_kafka_buf_read_i32(rkbuf, + &Partition); + break; + case RD_KAFKA_TOPIC_PARTITION_FIELD_OFFSET: + rd_kafka_buf_read_i64(rkbuf, &Offset); + break; + case RD_KAFKA_TOPIC_PARTITION_FIELD_CURRENT_EPOCH: + rd_kafka_buf_read_i32( + rkbuf, &CurrentLeaderEpoch); + break; + case RD_KAFKA_TOPIC_PARTITION_FIELD_EPOCH: + rd_kafka_buf_read_i32(rkbuf, &Epoch); + break; + case RD_KAFKA_TOPIC_PARTITION_FIELD_ERR: + rd_kafka_buf_read_i16(rkbuf, + &ErrorCode); + break; + case RD_KAFKA_TOPIC_PARTITION_FIELD_METADATA: + rd_assert(!*"metadata not implemented"); + break; + case RD_KAFKA_TOPIC_PARTITION_FIELD_NOOP: + break; + case RD_KAFKA_TOPIC_PARTITION_FIELD_END: + break; + } + } + + rktpar = rd_kafka_topic_partition_list_add(parts, topic, + Partition); + /* Use dummy sentinel values that are unlikely to be + * seen from the broker to know if we are to set these + * fields or not. */ + if (Offset != -1234) + rktpar->offset = Offset; + if (Epoch != -1234) + rd_kafka_topic_partition_set_leader_epoch( + rktpar, Epoch); + if (CurrentLeaderEpoch != -1234) + rd_kafka_topic_partition_set_current_leader_epoch( + rktpar, CurrentLeaderEpoch); + rktpar->err = ErrorCode; + + + rd_kafka_buf_skip_tags(rkbuf); + } + + rd_kafka_buf_skip_tags(rkbuf); + } + + return parts; + +err_parse: + if (parts) + rd_kafka_topic_partition_list_destroy(parts); + + return NULL; +} + + +/** + * @brief Write a list of topic+partitions+offsets+extra to \p rkbuf + * + * @returns the number of partitions written to buffer. + * + * @remark The \p parts list MUST be sorted. + */ +int rd_kafka_buf_write_topic_partitions( + rd_kafka_buf_t *rkbuf, + const rd_kafka_topic_partition_list_t *parts, + rd_bool_t skip_invalid_offsets, + rd_bool_t only_invalid_offsets, + const rd_kafka_topic_partition_field_t *fields) { + size_t of_TopicArrayCnt; + size_t of_PartArrayCnt = 0; + int TopicArrayCnt = 0, PartArrayCnt = 0; + int i; + const char *prev_topic = NULL; + int cnt = 0; + + rd_assert(!only_invalid_offsets || + (only_invalid_offsets != skip_invalid_offsets)); + + /* TopicArrayCnt */ + of_TopicArrayCnt = rd_kafka_buf_write_arraycnt_pos(rkbuf); + + for (i = 0; i < parts->cnt; i++) { + const rd_kafka_topic_partition_t *rktpar = &parts->elems[i]; + int fi; + + if (rktpar->offset < 0) { + if (skip_invalid_offsets) + continue; + } else if (only_invalid_offsets) + continue; + + if (!prev_topic || strcmp(rktpar->topic, prev_topic)) { + /* Finish previous topic, if any. */ + if (of_PartArrayCnt > 0) { + rd_kafka_buf_finalize_arraycnt( + rkbuf, of_PartArrayCnt, PartArrayCnt); + /* Tags for previous topic struct */ + rd_kafka_buf_write_tags(rkbuf); + } + + + /* Topic */ + rd_kafka_buf_write_str(rkbuf, rktpar->topic, -1); + TopicArrayCnt++; + prev_topic = rktpar->topic; + /* New topic so reset partition count */ + PartArrayCnt = 0; + + /* PartitionArrayCnt: updated later */ + of_PartArrayCnt = + rd_kafka_buf_write_arraycnt_pos(rkbuf); + } + + + /* + * Write requested fields + */ + for (fi = 0; fields[fi] != RD_KAFKA_TOPIC_PARTITION_FIELD_END; + fi++) { + switch (fields[fi]) { + case RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION: + rd_kafka_buf_write_i32(rkbuf, + rktpar->partition); + break; + case RD_KAFKA_TOPIC_PARTITION_FIELD_OFFSET: + rd_kafka_buf_write_i64(rkbuf, rktpar->offset); + break; + case RD_KAFKA_TOPIC_PARTITION_FIELD_CURRENT_EPOCH: + rd_kafka_buf_write_i32( + rkbuf, + rd_kafka_topic_partition_get_current_leader_epoch( + rktpar)); + break; + case RD_KAFKA_TOPIC_PARTITION_FIELD_EPOCH: + rd_kafka_buf_write_i32( + rkbuf, + rd_kafka_topic_partition_get_leader_epoch( + rktpar)); + break; + case RD_KAFKA_TOPIC_PARTITION_FIELD_ERR: + rd_kafka_buf_write_i16(rkbuf, rktpar->err); + break; + case RD_KAFKA_TOPIC_PARTITION_FIELD_METADATA: + /* Java client 0.9.0 and broker <0.10.0 can't + * parse Null metadata fields, so as a + * workaround we send an empty string if + * it's Null. */ + if (!rktpar->metadata) + rd_kafka_buf_write_str(rkbuf, "", 0); + else + rd_kafka_buf_write_str( + rkbuf, rktpar->metadata, + rktpar->metadata_size); + break; + case RD_KAFKA_TOPIC_PARTITION_FIELD_NOOP: + break; + case RD_KAFKA_TOPIC_PARTITION_FIELD_END: + break; + } + } + + + if (fi > 1) + /* If there was more than one field written + * then this was a struct and thus needs the + * struct suffix tags written. */ + rd_kafka_buf_write_tags(rkbuf); + + PartArrayCnt++; + cnt++; + } + + if (of_PartArrayCnt > 0) { + rd_kafka_buf_finalize_arraycnt(rkbuf, of_PartArrayCnt, + PartArrayCnt); + /* Tags for topic struct */ + rd_kafka_buf_write_tags(rkbuf); + } + + rd_kafka_buf_finalize_arraycnt(rkbuf, of_TopicArrayCnt, TopicArrayCnt); + + return cnt; +} + + +/** + * @brief Send FindCoordinatorRequest. + * + * @param coordkey is the group.id for RD_KAFKA_COORD_GROUP, + * and the transactional.id for RD_KAFKA_COORD_TXN + */ +rd_kafka_resp_err_t +rd_kafka_FindCoordinatorRequest(rd_kafka_broker_t *rkb, + rd_kafka_coordtype_t coordtype, + const char *coordkey, + rd_kafka_replyq_t replyq, + rd_kafka_resp_cb_t *resp_cb, + void *opaque) { + rd_kafka_buf_t *rkbuf; + int16_t ApiVersion; + + ApiVersion = rd_kafka_broker_ApiVersion_supported( + rkb, RD_KAFKAP_FindCoordinator, 0, 2, NULL); + + if (coordtype != RD_KAFKA_COORD_GROUP && ApiVersion < 1) + return RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE; + + rkbuf = rd_kafka_buf_new_request(rkb, RD_KAFKAP_FindCoordinator, 1, + 1 + 2 + strlen(coordkey)); + + rd_kafka_buf_write_str(rkbuf, coordkey, -1); + + if (ApiVersion >= 1) + rd_kafka_buf_write_i8(rkbuf, (int8_t)coordtype); + + rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0); + + rd_kafka_broker_buf_enq_replyq(rkb, rkbuf, replyq, resp_cb, opaque); + + return RD_KAFKA_RESP_ERR_NO_ERROR; +} + + + +/** + * @brief Parses a ListOffsets reply. + * + * Returns the parsed offsets (and errors) in \p offsets which must have been + * initialized by caller. + * + * @returns 0 on success, else an error (\p offsets may be completely or + * partially updated, depending on the nature of the error, and per + * partition error codes should be checked by the caller). + */ +static rd_kafka_resp_err_t +rd_kafka_parse_ListOffsets(rd_kafka_buf_t *rkbuf, + rd_kafka_topic_partition_list_t *offsets) { + const int log_decode_errors = LOG_ERR; + int32_t TopicArrayCnt; + int16_t api_version; + rd_kafka_resp_err_t all_err = RD_KAFKA_RESP_ERR_NO_ERROR; + + api_version = rkbuf->rkbuf_reqhdr.ApiVersion; + + if (api_version >= 2) + rd_kafka_buf_read_throttle_time(rkbuf); + + /* NOTE: + * Broker may return offsets in a different constellation than + * in the original request .*/ + + rd_kafka_buf_read_i32(rkbuf, &TopicArrayCnt); + while (TopicArrayCnt-- > 0) { + rd_kafkap_str_t ktopic; + int32_t PartArrayCnt; + char *topic_name; + + rd_kafka_buf_read_str(rkbuf, &ktopic); + rd_kafka_buf_read_i32(rkbuf, &PartArrayCnt); + + RD_KAFKAP_STR_DUPA(&topic_name, &ktopic); + + while (PartArrayCnt-- > 0) { + int32_t kpartition; + int16_t ErrorCode; + int32_t OffsetArrayCnt; + int64_t Offset = -1; + int32_t LeaderEpoch = -1; + rd_kafka_topic_partition_t *rktpar; + + rd_kafka_buf_read_i32(rkbuf, &kpartition); + rd_kafka_buf_read_i16(rkbuf, &ErrorCode); + + if (api_version >= 1) { + int64_t Timestamp; + rd_kafka_buf_read_i64(rkbuf, &Timestamp); + rd_kafka_buf_read_i64(rkbuf, &Offset); + if (api_version >= 4) + rd_kafka_buf_read_i32(rkbuf, + &LeaderEpoch); + } else if (api_version == 0) { + rd_kafka_buf_read_i32(rkbuf, &OffsetArrayCnt); + /* We only request one offset so just grab + * the first one. */ + while (OffsetArrayCnt-- > 0) + rd_kafka_buf_read_i64(rkbuf, &Offset); + } else { + RD_NOTREACHED(); + } + + rktpar = rd_kafka_topic_partition_list_add( + offsets, topic_name, kpartition); + rktpar->err = ErrorCode; + rktpar->offset = Offset; + rd_kafka_topic_partition_set_leader_epoch(rktpar, + LeaderEpoch); + + if (ErrorCode && !all_err) + all_err = ErrorCode; + } + } + + return all_err; + +err_parse: + return rkbuf->rkbuf_err; +} + + + +/** + * @brief Parses and handles ListOffsets replies. + * + * Returns the parsed offsets (and errors) in \p offsets. + * \p offsets must be initialized by the caller. + * + * @returns 0 on success, else an error. \p offsets may be populated on error, + * depending on the nature of the error. + * On error \p actionsp (unless NULL) is updated with the recommended + * error actions. + */ +rd_kafka_resp_err_t +rd_kafka_handle_ListOffsets(rd_kafka_t *rk, + rd_kafka_broker_t *rkb, + rd_kafka_resp_err_t err, + rd_kafka_buf_t *rkbuf, + rd_kafka_buf_t *request, + rd_kafka_topic_partition_list_t *offsets, + int *actionsp) { + + int actions; + + if (!err) + err = rd_kafka_parse_ListOffsets(rkbuf, offsets); + if (!err) + return RD_KAFKA_RESP_ERR_NO_ERROR; + + actions = rd_kafka_err_action( + rkb, err, request, RD_KAFKA_ERR_ACTION_PERMANENT, + RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PART, + + RD_KAFKA_ERR_ACTION_REFRESH, + RD_KAFKA_RESP_ERR_NOT_LEADER_FOR_PARTITION, + + RD_KAFKA_ERR_ACTION_REFRESH, + RD_KAFKA_RESP_ERR_REPLICA_NOT_AVAILABLE, + + RD_KAFKA_ERR_ACTION_REFRESH, RD_KAFKA_RESP_ERR_KAFKA_STORAGE_ERROR, + + RD_KAFKA_ERR_ACTION_REFRESH, RD_KAFKA_RESP_ERR_OFFSET_NOT_AVAILABLE, + + RD_KAFKA_ERR_ACTION_REFRESH | RD_KAFKA_ERR_ACTION_RETRY, + RD_KAFKA_RESP_ERR_LEADER_NOT_AVAILABLE, + + RD_KAFKA_ERR_ACTION_REFRESH | RD_KAFKA_ERR_ACTION_RETRY, + RD_KAFKA_RESP_ERR_FENCED_LEADER_EPOCH, + + RD_KAFKA_ERR_ACTION_REFRESH | RD_KAFKA_ERR_ACTION_RETRY, + RD_KAFKA_RESP_ERR_UNKNOWN_LEADER_EPOCH, + + RD_KAFKA_ERR_ACTION_RETRY, RD_KAFKA_RESP_ERR__TRANSPORT, + + RD_KAFKA_ERR_ACTION_RETRY, RD_KAFKA_RESP_ERR_REQUEST_TIMED_OUT, + + + RD_KAFKA_ERR_ACTION_END); + + if (actionsp) + *actionsp = actions; + + if (rkb) + rd_rkb_dbg( + rkb, TOPIC, "OFFSET", "OffsetRequest failed: %s (%s)", + rd_kafka_err2str(err), rd_kafka_actions2str(actions)); + + if (actions & RD_KAFKA_ERR_ACTION_REFRESH) { + char tmp[256]; + /* Re-query for leader */ + rd_snprintf(tmp, sizeof(tmp), "ListOffsetsRequest failed: %s", + rd_kafka_err2str(err)); + rd_kafka_metadata_refresh_known_topics(rk, NULL, + rd_true /*force*/, tmp); + } + + if ((actions & RD_KAFKA_ERR_ACTION_RETRY) && + rd_kafka_buf_retry(rkb, request)) + return RD_KAFKA_RESP_ERR__IN_PROGRESS; + + return err; +} + + + +/** + * @brief Async maker for ListOffsetsRequest. + */ +static rd_kafka_resp_err_t +rd_kafka_make_ListOffsetsRequest(rd_kafka_broker_t *rkb, + rd_kafka_buf_t *rkbuf, + void *make_opaque) { + const rd_kafka_topic_partition_list_t *partitions = + (const rd_kafka_topic_partition_list_t *)make_opaque; + int i; + size_t of_TopicArrayCnt = 0, of_PartArrayCnt = 0; + const char *last_topic = ""; + int32_t topic_cnt = 0, part_cnt = 0; + int16_t ApiVersion; + + ApiVersion = rd_kafka_broker_ApiVersion_supported( + rkb, RD_KAFKAP_ListOffsets, 0, 5, NULL); + if (ApiVersion == -1) + return RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE; + + /* ReplicaId */ + rd_kafka_buf_write_i32(rkbuf, -1); + + /* IsolationLevel */ + if (ApiVersion >= 2) + rd_kafka_buf_write_i8(rkbuf, + rkb->rkb_rk->rk_conf.isolation_level); + + /* TopicArrayCnt */ + of_TopicArrayCnt = rd_kafka_buf_write_i32(rkbuf, 0); /* updated later */ + + for (i = 0; i < partitions->cnt; i++) { + const rd_kafka_topic_partition_t *rktpar = + &partitions->elems[i]; + + if (strcmp(rktpar->topic, last_topic)) { + /* Finish last topic, if any. */ + if (of_PartArrayCnt > 0) + rd_kafka_buf_update_i32(rkbuf, of_PartArrayCnt, + part_cnt); + + /* Topic */ + rd_kafka_buf_write_str(rkbuf, rktpar->topic, -1); + topic_cnt++; + last_topic = rktpar->topic; + /* New topic so reset partition count */ + part_cnt = 0; + + /* PartitionArrayCnt: updated later */ + of_PartArrayCnt = rd_kafka_buf_write_i32(rkbuf, 0); + } + + /* Partition */ + rd_kafka_buf_write_i32(rkbuf, rktpar->partition); + part_cnt++; + + if (ApiVersion >= 4) + /* CurrentLeaderEpoch */ + rd_kafka_buf_write_i32( + rkbuf, + rd_kafka_topic_partition_get_current_leader_epoch( + rktpar)); + + /* Time/Offset */ + rd_kafka_buf_write_i64(rkbuf, rktpar->offset); + + if (ApiVersion == 0) { + /* MaxNumberOfOffsets */ + rd_kafka_buf_write_i32(rkbuf, 1); + } + } + + if (of_PartArrayCnt > 0) { + rd_kafka_buf_update_i32(rkbuf, of_PartArrayCnt, part_cnt); + rd_kafka_buf_update_i32(rkbuf, of_TopicArrayCnt, topic_cnt); + } + + rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0); + + rd_rkb_dbg(rkb, TOPIC, "OFFSET", + "ListOffsetsRequest (v%hd, opv %d) " + "for %" PRId32 " topic(s) and %" PRId32 " partition(s)", + ApiVersion, rkbuf->rkbuf_replyq.version, topic_cnt, + partitions->cnt); + + return RD_KAFKA_RESP_ERR_NO_ERROR; +} + + +/** + * @brief Send ListOffsetsRequest for partitions in \p partitions. + */ +void rd_kafka_ListOffsetsRequest(rd_kafka_broker_t *rkb, + rd_kafka_topic_partition_list_t *partitions, + rd_kafka_replyq_t replyq, + rd_kafka_resp_cb_t *resp_cb, + void *opaque) { + rd_kafka_buf_t *rkbuf; + rd_kafka_topic_partition_list_t *make_parts; + + make_parts = rd_kafka_topic_partition_list_copy(partitions); + rd_kafka_topic_partition_list_sort_by_topic(make_parts); + + rkbuf = rd_kafka_buf_new_request( + rkb, RD_KAFKAP_ListOffsets, 1, + /* ReplicaId+IsolationLevel+TopicArrayCnt+Topic */ + 4 + 1 + 4 + 100 + + /* PartArrayCnt */ + 4 + + /* partition_cnt * Partition+Time+MaxNumOffs */ + (make_parts->cnt * (4 + 8 + 4))); + + /* Postpone creating the request contents until time to send, + * at which time the ApiVersion is known. */ + rd_kafka_buf_set_maker(rkbuf, rd_kafka_make_ListOffsetsRequest, + make_parts, + rd_kafka_topic_partition_list_destroy_free); + + rd_kafka_broker_buf_enq_replyq(rkb, rkbuf, replyq, resp_cb, opaque); +} + + +/** + * @brief OffsetForLeaderEpochResponse handler. + */ +rd_kafka_resp_err_t rd_kafka_handle_OffsetForLeaderEpoch( + rd_kafka_t *rk, + rd_kafka_broker_t *rkb, + rd_kafka_resp_err_t err, + rd_kafka_buf_t *rkbuf, + rd_kafka_buf_t *request, + rd_kafka_topic_partition_list_t **offsets) { + const int log_decode_errors = LOG_ERR; + int16_t ApiVersion; + + if (err) + goto err; + + ApiVersion = rkbuf->rkbuf_reqhdr.ApiVersion; + + if (ApiVersion >= 2) + rd_kafka_buf_read_throttle_time(rkbuf); + + const rd_kafka_topic_partition_field_t fields[] = { + RD_KAFKA_TOPIC_PARTITION_FIELD_ERR, + RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION, + ApiVersion >= 1 ? RD_KAFKA_TOPIC_PARTITION_FIELD_EPOCH + : RD_KAFKA_TOPIC_PARTITION_FIELD_NOOP, + RD_KAFKA_TOPIC_PARTITION_FIELD_OFFSET, + RD_KAFKA_TOPIC_PARTITION_FIELD_END}; + *offsets = rd_kafka_buf_read_topic_partitions(rkbuf, 0, fields); + if (!*offsets) + goto err_parse; + + return RD_KAFKA_RESP_ERR_NO_ERROR; + +err: + return err; + +err_parse: + err = rkbuf->rkbuf_err; + goto err; +} + + +/** + * @brief Send OffsetForLeaderEpochRequest for partition(s). + * + */ +void rd_kafka_OffsetForLeaderEpochRequest( + rd_kafka_broker_t *rkb, + rd_kafka_topic_partition_list_t *parts, + rd_kafka_replyq_t replyq, + rd_kafka_resp_cb_t *resp_cb, + void *opaque) { + rd_kafka_buf_t *rkbuf; + int16_t ApiVersion; + + ApiVersion = rd_kafka_broker_ApiVersion_supported( + rkb, RD_KAFKAP_OffsetForLeaderEpoch, 2, 2, NULL); + /* If the supported ApiVersions are not yet known, + * or this broker doesn't support it, we let this request + * succeed or fail later from the broker thread where the + * version is checked again. */ + if (ApiVersion == -1) + ApiVersion = 2; + + rkbuf = rd_kafka_buf_new_flexver_request( + rkb, RD_KAFKAP_OffsetForLeaderEpoch, 1, 4 + (parts->cnt * 64), + ApiVersion >= 4 /*flexver*/); + + /* Sort partitions by topic */ + rd_kafka_topic_partition_list_sort_by_topic(parts); + + /* Write partition list */ + const rd_kafka_topic_partition_field_t fields[] = { + RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION, + /* CurrentLeaderEpoch */ + RD_KAFKA_TOPIC_PARTITION_FIELD_CURRENT_EPOCH, + /* LeaderEpoch */ + RD_KAFKA_TOPIC_PARTITION_FIELD_EPOCH, + RD_KAFKA_TOPIC_PARTITION_FIELD_END}; + rd_kafka_buf_write_topic_partitions( + rkbuf, parts, rd_false /*include invalid offsets*/, + rd_false /*skip valid offsets */, fields); + + rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0); + + /* Let caller perform retries */ + rkbuf->rkbuf_max_retries = RD_KAFKA_REQUEST_NO_RETRIES; + + rd_kafka_broker_buf_enq_replyq(rkb, rkbuf, replyq, resp_cb, opaque); +} + + + +/** + * Generic handler for OffsetFetch responses. + * Offsets for included partitions will be propagated through the passed + * 'offsets' list. + * + * @param rkbuf response buffer, may be NULL if \p err is set. + * @param update_toppar update toppar's committed_offset + * @param add_part if true add partitions from the response to \p *offsets, + * else just update the partitions that are already + * in \p *offsets. + */ +rd_kafka_resp_err_t +rd_kafka_handle_OffsetFetch(rd_kafka_t *rk, + rd_kafka_broker_t *rkb, + rd_kafka_resp_err_t err, + rd_kafka_buf_t *rkbuf, + rd_kafka_buf_t *request, + rd_kafka_topic_partition_list_t **offsets, + rd_bool_t update_toppar, + rd_bool_t add_part, + rd_bool_t allow_retry) { + const int log_decode_errors = LOG_ERR; + int32_t TopicArrayCnt; + int64_t offset = RD_KAFKA_OFFSET_INVALID; + int16_t ApiVersion; + rd_kafkap_str_t metadata; + int retry_unstable = 0; + int i; + int actions; + int seen_cnt = 0; + + if (err) + goto err; + + ApiVersion = rkbuf->rkbuf_reqhdr.ApiVersion; + + if (ApiVersion >= 3) + rd_kafka_buf_read_throttle_time(rkbuf); + + if (!*offsets) + *offsets = rd_kafka_topic_partition_list_new(16); + + /* Set default offset for all partitions. */ + rd_kafka_topic_partition_list_set_offsets(rkb->rkb_rk, *offsets, 0, + RD_KAFKA_OFFSET_INVALID, + 0 /* !is commit */); + + rd_kafka_buf_read_arraycnt(rkbuf, &TopicArrayCnt, RD_KAFKAP_TOPICS_MAX); + for (i = 0; i < TopicArrayCnt; i++) { + rd_kafkap_str_t topic; + int32_t PartArrayCnt; + char *topic_name; + int j; + + rd_kafka_buf_read_str(rkbuf, &topic); + + rd_kafka_buf_read_arraycnt(rkbuf, &PartArrayCnt, + RD_KAFKAP_PARTITIONS_MAX); + + RD_KAFKAP_STR_DUPA(&topic_name, &topic); + + for (j = 0; j < PartArrayCnt; j++) { + int32_t partition; + rd_kafka_toppar_t *rktp; + rd_kafka_topic_partition_t *rktpar; + int32_t LeaderEpoch = -1; + int16_t err2; + + rd_kafka_buf_read_i32(rkbuf, &partition); + rd_kafka_buf_read_i64(rkbuf, &offset); + if (ApiVersion >= 5) + rd_kafka_buf_read_i32(rkbuf, &LeaderEpoch); + rd_kafka_buf_read_str(rkbuf, &metadata); + rd_kafka_buf_read_i16(rkbuf, &err2); + rd_kafka_buf_skip_tags(rkbuf); + + rktpar = rd_kafka_topic_partition_list_find( + *offsets, topic_name, partition); + if (!rktpar && add_part) + rktpar = rd_kafka_topic_partition_list_add( + *offsets, topic_name, partition); + else if (!rktpar) { + rd_rkb_dbg(rkb, TOPIC, "OFFSETFETCH", + "OffsetFetchResponse: %s [%" PRId32 + "] " + "not found in local list: ignoring", + topic_name, partition); + continue; + } + + seen_cnt++; + + rktp = rd_kafka_topic_partition_get_toppar( + rk, rktpar, rd_false /*no create on miss*/); + + /* broker reports invalid offset as -1 */ + if (offset == -1) + rktpar->offset = RD_KAFKA_OFFSET_INVALID; + else + rktpar->offset = offset; + + rd_kafka_topic_partition_set_leader_epoch(rktpar, + LeaderEpoch); + rktpar->err = err2; + + rd_rkb_dbg(rkb, TOPIC, "OFFSETFETCH", + "OffsetFetchResponse: %s [%" PRId32 + "] " + "offset %" PRId64 ", leader epoch %" PRId32 + ", metadata %d byte(s): %s", + topic_name, partition, offset, LeaderEpoch, + RD_KAFKAP_STR_LEN(&metadata), + rd_kafka_err2name(rktpar->err)); + + if (update_toppar && !err2 && rktp) { + /* Update toppar's committed offset */ + rd_kafka_toppar_lock(rktp); + rktp->rktp_committed_pos = + rd_kafka_topic_partition_get_fetch_pos( + rktpar); + rd_kafka_toppar_unlock(rktp); + } + + if (rktpar->err == + RD_KAFKA_RESP_ERR_UNSTABLE_OFFSET_COMMIT) + retry_unstable++; + + + if (rktpar->metadata) + rd_free(rktpar->metadata); + + if (RD_KAFKAP_STR_IS_NULL(&metadata)) { + rktpar->metadata = NULL; + rktpar->metadata_size = 0; + } else { + rktpar->metadata = RD_KAFKAP_STR_DUP(&metadata); + rktpar->metadata_size = + RD_KAFKAP_STR_LEN(&metadata); + } + + /* Loose ref from get_toppar() */ + if (rktp) + rd_kafka_toppar_destroy(rktp); + } + + rd_kafka_buf_skip_tags(rkbuf); + } + + if (ApiVersion >= 2) { + int16_t ErrorCode; + rd_kafka_buf_read_i16(rkbuf, &ErrorCode); + if (ErrorCode) { + err = ErrorCode; + goto err; + } + } + + +err: + if (!*offsets) + rd_rkb_dbg(rkb, TOPIC, "OFFFETCH", "OffsetFetch returned %s", + rd_kafka_err2str(err)); + else + rd_rkb_dbg(rkb, TOPIC, "OFFFETCH", + "OffsetFetch for %d/%d partition(s) " + "(%d unstable partition(s)) returned %s", + seen_cnt, (*offsets)->cnt, retry_unstable, + rd_kafka_err2str(err)); + + actions = + rd_kafka_err_action(rkb, err, request, RD_KAFKA_ERR_ACTION_END); + + if (actions & RD_KAFKA_ERR_ACTION_REFRESH) { + /* Re-query for coordinator */ + rd_kafka_cgrp_op(rkb->rkb_rk->rk_cgrp, NULL, RD_KAFKA_NO_REPLYQ, + RD_KAFKA_OP_COORD_QUERY, err); + } + + if (actions & RD_KAFKA_ERR_ACTION_RETRY || retry_unstable) { + if (allow_retry && rd_kafka_buf_retry(rkb, request)) + return RD_KAFKA_RESP_ERR__IN_PROGRESS; + /* FALLTHRU */ + } + + return err; + +err_parse: + err = rkbuf->rkbuf_err; + goto err; +} + + + +/** + * @brief Handle OffsetFetch response based on an RD_KAFKA_OP_OFFSET_FETCH + * rko in \p opaque. + * + * @param opaque rko wrapper for handle_OffsetFetch. + * + * The \c rko->rko_u.offset_fetch.partitions list will be filled in with + * the fetched offsets. + * + * A reply will be sent on 'rko->rko_replyq' with type RD_KAFKA_OP_OFFSET_FETCH. + * + * @remark \p rkb, \p rkbuf and \p request are optional. + * + * @remark The \p request buffer may be retried on error. + * + * @locality cgrp's broker thread + */ +void rd_kafka_op_handle_OffsetFetch(rd_kafka_t *rk, + rd_kafka_broker_t *rkb, + rd_kafka_resp_err_t err, + rd_kafka_buf_t *rkbuf, + rd_kafka_buf_t *request, + void *opaque) { + rd_kafka_op_t *rko = opaque; + rd_kafka_op_t *rko_reply; + rd_kafka_topic_partition_list_t *offsets; + + RD_KAFKA_OP_TYPE_ASSERT(rko, RD_KAFKA_OP_OFFSET_FETCH); + + if (err == RD_KAFKA_RESP_ERR__DESTROY) { + /* Termination, quick cleanup. */ + rd_kafka_op_destroy(rko); + return; + } + + offsets = rd_kafka_topic_partition_list_copy( + rko->rko_u.offset_fetch.partitions); + + /* If all partitions already had usable offsets then there + * was no request sent and thus no reply, the offsets list is + * good to go.. */ + if (rkbuf) { + /* ..else parse the response (or perror) */ + err = rd_kafka_handle_OffsetFetch( + rkb->rkb_rk, rkb, err, rkbuf, request, &offsets, + rd_false /*dont update rktp*/, rd_false /*dont add part*/, + /* Allow retries if replyq is valid */ + rd_kafka_op_replyq_is_valid(rko)); + if (err == RD_KAFKA_RESP_ERR__IN_PROGRESS) { + if (offsets) + rd_kafka_topic_partition_list_destroy(offsets); + return; /* Retrying */ + } + } + + rko_reply = + rd_kafka_op_new(RD_KAFKA_OP_OFFSET_FETCH | RD_KAFKA_OP_REPLY); + rko_reply->rko_err = err; + rko_reply->rko_u.offset_fetch.partitions = offsets; + rko_reply->rko_u.offset_fetch.do_free = 1; + if (rko->rko_rktp) + rko_reply->rko_rktp = rd_kafka_toppar_keep(rko->rko_rktp); + + rd_kafka_replyq_enq(&rko->rko_replyq, rko_reply, 0); + + rd_kafka_op_destroy(rko); +} + +/** + * Send OffsetFetchRequest for a consumer group id. + * + * Any partition with a usable offset will be ignored, if all partitions + * have usable offsets then no request is sent at all but an empty + * reply is enqueued on the replyq. + * + * @param group_id Request offset for this group id. + * @param parts (optional) List of topic partitions to request, + * or NULL to return all topic partitions associated with the + * group. + * @param require_stable_offsets Whether broker should return stable offsets + * (transaction-committed). + * @param timeout Optional timeout to set to the buffer. + */ +void rd_kafka_OffsetFetchRequest(rd_kafka_broker_t *rkb, + const char *group_id, + rd_kafka_topic_partition_list_t *parts, + rd_bool_t require_stable_offsets, + int timeout, + rd_kafka_replyq_t replyq, + rd_kafka_resp_cb_t *resp_cb, + void *opaque) { + rd_kafka_buf_t *rkbuf; + int16_t ApiVersion; + size_t parts_size = 0; + int PartCnt = -1; + + ApiVersion = rd_kafka_broker_ApiVersion_supported( + rkb, RD_KAFKAP_OffsetFetch, 0, 7, NULL); + + if (parts) { + parts_size = parts->cnt * 32; + } + + rkbuf = rd_kafka_buf_new_flexver_request( + rkb, RD_KAFKAP_OffsetFetch, 1, + /* GroupId + rd_kafka_buf_write_arraycnt_pos + + * Topics + RequireStable */ + 32 + 4 + parts_size + 1, ApiVersion >= 6 /*flexver*/); + + /* ConsumerGroup */ + rd_kafka_buf_write_str(rkbuf, group_id, -1); + + if (parts) { + /* Sort partitions by topic */ + rd_kafka_topic_partition_list_sort_by_topic(parts); + + /* Write partition list, filtering out partitions with valid + * offsets */ + const rd_kafka_topic_partition_field_t fields[] = { + RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION, + RD_KAFKA_TOPIC_PARTITION_FIELD_END}; + PartCnt = rd_kafka_buf_write_topic_partitions( + rkbuf, parts, rd_false /*include invalid offsets*/, + rd_false /*skip valid offsets */, fields); + } else { + rd_kafka_buf_write_arraycnt_pos(rkbuf); + } + + if (ApiVersion >= 7) { + /* RequireStable */ + rd_kafka_buf_write_i8(rkbuf, require_stable_offsets); + } + + if (PartCnt == 0) { + /* No partitions needs OffsetFetch, enqueue empty + * response right away. */ + rkbuf->rkbuf_replyq = replyq; + rkbuf->rkbuf_cb = resp_cb; + rkbuf->rkbuf_opaque = opaque; + rd_kafka_buf_callback(rkb->rkb_rk, rkb, 0, NULL, rkbuf); + return; + } + + if (timeout > rkb->rkb_rk->rk_conf.socket_timeout_ms) + rd_kafka_buf_set_abs_timeout(rkbuf, timeout + 1000, 0); + + rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0); + + if (parts) { + rd_rkb_dbg( + rkb, TOPIC | RD_KAFKA_DBG_CGRP | RD_KAFKA_DBG_CONSUMER, + "OFFSET", + "Group %s OffsetFetchRequest(v%d) for %d/%d partition(s)", + group_id, ApiVersion, PartCnt, parts->cnt); + } else { + rd_rkb_dbg( + rkb, TOPIC | RD_KAFKA_DBG_CGRP | RD_KAFKA_DBG_CONSUMER, + "OFFSET", + "Group %s OffsetFetchRequest(v%d) for all partitions", + group_id, ApiVersion); + } + + /* Let handler decide if retries should be performed */ + rkbuf->rkbuf_max_retries = RD_KAFKA_REQUEST_MAX_RETRIES; + + if (parts) { + rd_rkb_dbg(rkb, CGRP | RD_KAFKA_DBG_CONSUMER, "OFFSET", + "Fetch committed offsets for %d/%d partition(s)", + PartCnt, parts->cnt); + } else { + rd_rkb_dbg(rkb, CGRP | RD_KAFKA_DBG_CONSUMER, "OFFSET", + "Fetch committed offsets all the partitions"); + } + + rd_kafka_broker_buf_enq_replyq(rkb, rkbuf, replyq, resp_cb, opaque); +} + + + +/** + * @brief Handle per-partition OffsetCommit errors and returns actions flags. + */ +static int +rd_kafka_handle_OffsetCommit_error(rd_kafka_broker_t *rkb, + rd_kafka_buf_t *request, + const rd_kafka_topic_partition_t *rktpar) { + + /* These actions are mimicking AK's ConsumerCoordinator.java */ + + return rd_kafka_err_action( + rkb, rktpar->err, request, + + RD_KAFKA_ERR_ACTION_PERMANENT, + RD_KAFKA_RESP_ERR_GROUP_AUTHORIZATION_FAILED, + + RD_KAFKA_ERR_ACTION_PERMANENT, + RD_KAFKA_RESP_ERR_TOPIC_AUTHORIZATION_FAILED, + + + RD_KAFKA_ERR_ACTION_PERMANENT, + RD_KAFKA_RESP_ERR_OFFSET_METADATA_TOO_LARGE, + + RD_KAFKA_ERR_ACTION_PERMANENT, + RD_KAFKA_RESP_ERR_INVALID_COMMIT_OFFSET_SIZE, + + + RD_KAFKA_ERR_ACTION_RETRY, + RD_KAFKA_RESP_ERR_COORDINATOR_LOAD_IN_PROGRESS, + + RD_KAFKA_ERR_ACTION_RETRY, RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PART, + + + /* .._SPECIAL: mark coordinator dead, refresh and retry */ + RD_KAFKA_ERR_ACTION_REFRESH | RD_KAFKA_ERR_ACTION_RETRY | + RD_KAFKA_ERR_ACTION_SPECIAL, + RD_KAFKA_RESP_ERR_COORDINATOR_NOT_AVAILABLE, + + RD_KAFKA_ERR_ACTION_REFRESH | RD_KAFKA_ERR_ACTION_RETRY | + RD_KAFKA_ERR_ACTION_SPECIAL, + RD_KAFKA_RESP_ERR_NOT_COORDINATOR, + + /* Replicas possibly unavailable: + * Refresh coordinator (but don't mark as dead (!.._SPECIAL)), + * and retry */ + RD_KAFKA_ERR_ACTION_REFRESH | RD_KAFKA_ERR_ACTION_RETRY, + RD_KAFKA_RESP_ERR_REQUEST_TIMED_OUT, + + + /* FIXME: There are some cases in the Java code where + * this is not treated as a fatal error. */ + RD_KAFKA_ERR_ACTION_PERMANENT | RD_KAFKA_ERR_ACTION_FATAL, + RD_KAFKA_RESP_ERR_FENCED_INSTANCE_ID, + + + RD_KAFKA_ERR_ACTION_PERMANENT, + RD_KAFKA_RESP_ERR_REBALANCE_IN_PROGRESS, + + + RD_KAFKA_ERR_ACTION_PERMANENT, RD_KAFKA_RESP_ERR_UNKNOWN_MEMBER_ID, + + RD_KAFKA_ERR_ACTION_PERMANENT, RD_KAFKA_RESP_ERR_ILLEGAL_GENERATION, + + RD_KAFKA_ERR_ACTION_END); +} + + +/** + * @brief Handle OffsetCommit response. + * + * @remark \p offsets may be NULL if \p err is set + * + * @returns RD_KAFKA_RESP_ERR_NO_ERROR if all partitions were successfully + * committed, + * RD_KAFKA_RESP_ERR__IN_PROGRESS if a retry was scheduled, + * or any other error code if the request was not retried. + */ +rd_kafka_resp_err_t +rd_kafka_handle_OffsetCommit(rd_kafka_t *rk, + rd_kafka_broker_t *rkb, + rd_kafka_resp_err_t err, + rd_kafka_buf_t *rkbuf, + rd_kafka_buf_t *request, + rd_kafka_topic_partition_list_t *offsets, + rd_bool_t ignore_cgrp) { + const int log_decode_errors = LOG_ERR; + int32_t TopicArrayCnt; + int errcnt = 0; + int partcnt = 0; + int i; + int actions = 0; + + if (err) + goto err; + + if (rd_kafka_buf_ApiVersion(rkbuf) >= 3) + rd_kafka_buf_read_throttle_time(rkbuf); + + rd_kafka_buf_read_i32(rkbuf, &TopicArrayCnt); + for (i = 0; i < TopicArrayCnt; i++) { + rd_kafkap_str_t topic; + char *topic_str; + int32_t PartArrayCnt; + int j; + + rd_kafka_buf_read_str(rkbuf, &topic); + rd_kafka_buf_read_i32(rkbuf, &PartArrayCnt); + + RD_KAFKAP_STR_DUPA(&topic_str, &topic); + + for (j = 0; j < PartArrayCnt; j++) { + int32_t partition; + int16_t ErrorCode; + rd_kafka_topic_partition_t *rktpar; + + rd_kafka_buf_read_i32(rkbuf, &partition); + rd_kafka_buf_read_i16(rkbuf, &ErrorCode); + + rktpar = rd_kafka_topic_partition_list_find( + offsets, topic_str, partition); + + if (!rktpar) { + /* Received offset for topic/partition we didn't + * ask for, this shouldn't really happen. */ + continue; + } + + rktpar->err = ErrorCode; + if (ErrorCode) { + err = ErrorCode; + errcnt++; + + /* Accumulate actions for per-partition + * errors. */ + actions |= rd_kafka_handle_OffsetCommit_error( + rkb, request, rktpar); + } + + partcnt++; + } + } + + /* If all partitions failed use error code + * from last partition as the global error. */ + if (offsets && err && errcnt == partcnt) + goto err; + + goto done; + +err_parse: + err = rkbuf->rkbuf_err; + +err: + if (!actions) /* Transport/Request-level error */ + actions = rd_kafka_err_action(rkb, err, request, + + RD_KAFKA_ERR_ACTION_REFRESH | + RD_KAFKA_ERR_ACTION_SPECIAL | + RD_KAFKA_ERR_ACTION_RETRY, + RD_KAFKA_RESP_ERR__TRANSPORT, + + RD_KAFKA_ERR_ACTION_END); + + if (!ignore_cgrp && (actions & RD_KAFKA_ERR_ACTION_FATAL)) { + rd_kafka_set_fatal_error(rk, err, "OffsetCommit failed: %s", + rd_kafka_err2str(err)); + return err; + } + + if (!ignore_cgrp && (actions & RD_KAFKA_ERR_ACTION_REFRESH) && + rk->rk_cgrp) { + /* Mark coordinator dead or re-query for coordinator. + * ..dead() will trigger a re-query. */ + if (actions & RD_KAFKA_ERR_ACTION_SPECIAL) + rd_kafka_cgrp_coord_dead(rk->rk_cgrp, err, + "OffsetCommitRequest failed"); + else + rd_kafka_cgrp_coord_query(rk->rk_cgrp, + "OffsetCommitRequest failed"); + } + + if (!ignore_cgrp && actions & RD_KAFKA_ERR_ACTION_RETRY && + !(actions & RD_KAFKA_ERR_ACTION_PERMANENT) && + rd_kafka_buf_retry(rkb, request)) + return RD_KAFKA_RESP_ERR__IN_PROGRESS; + +done: + return err; +} + +/** + * @brief Send OffsetCommitRequest for a list of partitions. + * + * @param cgmetadata consumer group metadata. + * + * @param offsets - offsets to commit for each topic-partition. + * + * @returns 0 if none of the partitions in \p offsets had valid offsets, + * else 1. + */ +int rd_kafka_OffsetCommitRequest(rd_kafka_broker_t *rkb, + rd_kafka_consumer_group_metadata_t *cgmetadata, + rd_kafka_topic_partition_list_t *offsets, + rd_kafka_replyq_t replyq, + rd_kafka_resp_cb_t *resp_cb, + void *opaque, + const char *reason) { + rd_kafka_buf_t *rkbuf; + ssize_t of_TopicCnt = -1; + int TopicCnt = 0; + const char *last_topic = NULL; + ssize_t of_PartCnt = -1; + int PartCnt = 0; + int tot_PartCnt = 0; + int i; + int16_t ApiVersion; + int features; + + ApiVersion = rd_kafka_broker_ApiVersion_supported( + rkb, RD_KAFKAP_OffsetCommit, 0, 7, &features); + + rd_kafka_assert(NULL, offsets != NULL); + + rkbuf = rd_kafka_buf_new_request(rkb, RD_KAFKAP_OffsetCommit, 1, + 100 + (offsets->cnt * 128)); + + /* ConsumerGroup */ + rd_kafka_buf_write_str(rkbuf, cgmetadata->group_id, -1); + + /* v1,v2 */ + if (ApiVersion >= 1) { + /* ConsumerGroupGenerationId */ + rd_kafka_buf_write_i32(rkbuf, cgmetadata->generation_id); + /* ConsumerId */ + rd_kafka_buf_write_str(rkbuf, cgmetadata->member_id, -1); + } + + /* v7: GroupInstanceId */ + if (ApiVersion >= 7) + rd_kafka_buf_write_str(rkbuf, cgmetadata->group_instance_id, + -1); + + /* v2-4: RetentionTime */ + if (ApiVersion >= 2 && ApiVersion <= 4) + rd_kafka_buf_write_i64(rkbuf, -1); + + /* Sort offsets by topic */ + rd_kafka_topic_partition_list_sort_by_topic(offsets); + + /* TopicArrayCnt: Will be updated when we know the number of topics. */ + of_TopicCnt = rd_kafka_buf_write_i32(rkbuf, 0); + + for (i = 0; i < offsets->cnt; i++) { + rd_kafka_topic_partition_t *rktpar = &offsets->elems[i]; + + /* Skip partitions with invalid offset. */ + if (rktpar->offset < 0) + continue; + + if (last_topic == NULL || strcmp(last_topic, rktpar->topic)) { + /* New topic */ + + /* Finalize previous PartitionCnt */ + if (PartCnt > 0) + rd_kafka_buf_update_u32(rkbuf, of_PartCnt, + PartCnt); + + /* TopicName */ + rd_kafka_buf_write_str(rkbuf, rktpar->topic, -1); + /* PartitionCnt, finalized later */ + of_PartCnt = rd_kafka_buf_write_i32(rkbuf, 0); + PartCnt = 0; + last_topic = rktpar->topic; + TopicCnt++; + } + + /* Partition */ + rd_kafka_buf_write_i32(rkbuf, rktpar->partition); + PartCnt++; + tot_PartCnt++; + + /* Offset */ + rd_kafka_buf_write_i64(rkbuf, rktpar->offset); + + /* v6: KIP-101 CommittedLeaderEpoch */ + if (ApiVersion >= 6) + rd_kafka_buf_write_i32( + rkbuf, + rd_kafka_topic_partition_get_leader_epoch(rktpar)); + + /* v1: TimeStamp */ + if (ApiVersion == 1) + rd_kafka_buf_write_i64(rkbuf, -1); + + /* Metadata */ + /* Java client 0.9.0 and broker <0.10.0 can't parse + * Null metadata fields, so as a workaround we send an + * empty string if it's Null. */ + if (!rktpar->metadata) + rd_kafka_buf_write_str(rkbuf, "", 0); + else + rd_kafka_buf_write_str(rkbuf, rktpar->metadata, + rktpar->metadata_size); + } + + if (tot_PartCnt == 0) { + /* No topic+partitions had valid offsets to commit. */ + rd_kafka_replyq_destroy(&replyq); + rd_kafka_buf_destroy(rkbuf); + return 0; + } + + /* Finalize previous PartitionCnt */ + if (PartCnt > 0) + rd_kafka_buf_update_u32(rkbuf, of_PartCnt, PartCnt); + + /* Finalize TopicCnt */ + rd_kafka_buf_update_u32(rkbuf, of_TopicCnt, TopicCnt); + + rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0); + + rd_rkb_dbg(rkb, TOPIC, "OFFSET", + "Enqueue OffsetCommitRequest(v%d, %d/%d partition(s))): %s", + ApiVersion, tot_PartCnt, offsets->cnt, reason); + + rd_kafka_broker_buf_enq_replyq(rkb, rkbuf, replyq, resp_cb, opaque); + + return 1; +} + +/** + * @brief Construct and send OffsetDeleteRequest to \p rkb + * with the partitions in del_grpoffsets (DeleteConsumerGroupOffsets_t*) + * using \p options. + * + * The response (unparsed) will be enqueued on \p replyq + * for handling by \p resp_cb (with \p opaque passed). + * + * @remark Only one del_grpoffsets element is supported. + * + * @returns RD_KAFKA_RESP_ERR_NO_ERROR if the request was enqueued for + * transmission, otherwise an error code and errstr will be + * updated with a human readable error string. + */ +rd_kafka_resp_err_t +rd_kafka_OffsetDeleteRequest(rd_kafka_broker_t *rkb, + /** (rd_kafka_DeleteConsumerGroupOffsets_t*) */ + const rd_list_t *del_grpoffsets, + rd_kafka_AdminOptions_t *options, + char *errstr, + size_t errstr_size, + rd_kafka_replyq_t replyq, + rd_kafka_resp_cb_t *resp_cb, + void *opaque) { + rd_kafka_buf_t *rkbuf; + int16_t ApiVersion = 0; + int features; + const rd_kafka_DeleteConsumerGroupOffsets_t *grpoffsets = + rd_list_elem(del_grpoffsets, 0); + + rd_assert(rd_list_cnt(del_grpoffsets) == 1); + + ApiVersion = rd_kafka_broker_ApiVersion_supported( + rkb, RD_KAFKAP_OffsetDelete, 0, 0, &features); + if (ApiVersion == -1) { + rd_snprintf(errstr, errstr_size, + "OffsetDelete API (KIP-496) not supported " + "by broker, requires broker version >= 2.4.0"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE; + } + + rkbuf = rd_kafka_buf_new_request( + rkb, RD_KAFKAP_OffsetDelete, 1, + 2 + strlen(grpoffsets->group) + (64 * grpoffsets->partitions->cnt)); + + /* GroupId */ + rd_kafka_buf_write_str(rkbuf, grpoffsets->group, -1); + + const rd_kafka_topic_partition_field_t fields[] = { + RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION, + RD_KAFKA_TOPIC_PARTITION_FIELD_END}; + rd_kafka_buf_write_topic_partitions( + rkbuf, grpoffsets->partitions, + rd_false /*dont skip invalid offsets*/, rd_false /*any offset*/, + fields); + + rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0); + + rd_kafka_broker_buf_enq_replyq(rkb, rkbuf, replyq, resp_cb, opaque); + + return RD_KAFKA_RESP_ERR_NO_ERROR; +} + + + +/** + * @brief Write "consumer" protocol type MemberState for SyncGroupRequest to + * enveloping buffer \p rkbuf. + */ +static void +rd_kafka_group_MemberState_consumer_write(rd_kafka_buf_t *env_rkbuf, + const rd_kafka_group_member_t *rkgm) { + rd_kafka_buf_t *rkbuf; + rd_slice_t slice; + + rkbuf = rd_kafka_buf_new(1, 100); + rd_kafka_buf_write_i16(rkbuf, 0); /* Version */ + rd_assert(rkgm->rkgm_assignment); + const rd_kafka_topic_partition_field_t fields[] = { + RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION, + RD_KAFKA_TOPIC_PARTITION_FIELD_END}; + rd_kafka_buf_write_topic_partitions( + rkbuf, rkgm->rkgm_assignment, + rd_false /*don't skip invalid offsets*/, rd_false /* any offset */, + fields); + rd_kafka_buf_write_kbytes(rkbuf, rkgm->rkgm_userdata); + + /* Get pointer to binary buffer */ + rd_slice_init_full(&slice, &rkbuf->rkbuf_buf); + + /* Write binary buffer as Kafka Bytes to enveloping buffer. */ + rd_kafka_buf_write_i32(env_rkbuf, (int32_t)rd_slice_remains(&slice)); + rd_buf_write_slice(&env_rkbuf->rkbuf_buf, &slice); + + rd_kafka_buf_destroy(rkbuf); +} + +/** + * Send SyncGroupRequest + */ +void rd_kafka_SyncGroupRequest(rd_kafka_broker_t *rkb, + const rd_kafkap_str_t *group_id, + int32_t generation_id, + const rd_kafkap_str_t *member_id, + const rd_kafkap_str_t *group_instance_id, + const rd_kafka_group_member_t *assignments, + int assignment_cnt, + rd_kafka_replyq_t replyq, + rd_kafka_resp_cb_t *resp_cb, + void *opaque) { + rd_kafka_buf_t *rkbuf; + int i; + int16_t ApiVersion; + int features; + + ApiVersion = rd_kafka_broker_ApiVersion_supported( + rkb, RD_KAFKAP_SyncGroup, 0, 3, &features); + + rkbuf = rd_kafka_buf_new_request( + rkb, RD_KAFKAP_SyncGroup, 1, + RD_KAFKAP_STR_SIZE(group_id) + 4 /* GenerationId */ + + RD_KAFKAP_STR_SIZE(member_id) + + RD_KAFKAP_STR_SIZE(group_instance_id) + + 4 /* array size group_assignment */ + + (assignment_cnt * 100 /*guess*/)); + rd_kafka_buf_write_kstr(rkbuf, group_id); + rd_kafka_buf_write_i32(rkbuf, generation_id); + rd_kafka_buf_write_kstr(rkbuf, member_id); + if (ApiVersion >= 3) + rd_kafka_buf_write_kstr(rkbuf, group_instance_id); + rd_kafka_buf_write_i32(rkbuf, assignment_cnt); + + for (i = 0; i < assignment_cnt; i++) { + const rd_kafka_group_member_t *rkgm = &assignments[i]; + + rd_kafka_buf_write_kstr(rkbuf, rkgm->rkgm_member_id); + rd_kafka_group_MemberState_consumer_write(rkbuf, rkgm); + } + + /* This is a blocking request */ + rkbuf->rkbuf_flags |= RD_KAFKA_OP_F_BLOCKING; + rd_kafka_buf_set_abs_timeout( + rkbuf, + rkb->rkb_rk->rk_conf.group_session_timeout_ms + + 3000 /* 3s grace period*/, + 0); + + rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0); + + rd_kafka_broker_buf_enq_replyq(rkb, rkbuf, replyq, resp_cb, opaque); +} + + + +/** + * Send JoinGroupRequest + */ +void rd_kafka_JoinGroupRequest(rd_kafka_broker_t *rkb, + const rd_kafkap_str_t *group_id, + const rd_kafkap_str_t *member_id, + const rd_kafkap_str_t *group_instance_id, + const rd_kafkap_str_t *protocol_type, + const rd_list_t *topics, + rd_kafka_replyq_t replyq, + rd_kafka_resp_cb_t *resp_cb, + void *opaque) { + rd_kafka_buf_t *rkbuf; + rd_kafka_t *rk = rkb->rkb_rk; + rd_kafka_assignor_t *rkas; + int i; + int16_t ApiVersion = 0; + int features; + + ApiVersion = rd_kafka_broker_ApiVersion_supported( + rkb, RD_KAFKAP_JoinGroup, 0, 5, &features); + + + rkbuf = rd_kafka_buf_new_request( + rkb, RD_KAFKAP_JoinGroup, 1, + RD_KAFKAP_STR_SIZE(group_id) + 4 /* sessionTimeoutMs */ + + 4 /* rebalanceTimeoutMs */ + RD_KAFKAP_STR_SIZE(member_id) + + RD_KAFKAP_STR_SIZE(group_instance_id) + + RD_KAFKAP_STR_SIZE(protocol_type) + + 4 /* array count GroupProtocols */ + + (rd_list_cnt(topics) * 100)); + rd_kafka_buf_write_kstr(rkbuf, group_id); + rd_kafka_buf_write_i32(rkbuf, rk->rk_conf.group_session_timeout_ms); + if (ApiVersion >= 1) + rd_kafka_buf_write_i32(rkbuf, rk->rk_conf.max_poll_interval_ms); + rd_kafka_buf_write_kstr(rkbuf, member_id); + if (ApiVersion >= 5) + rd_kafka_buf_write_kstr(rkbuf, group_instance_id); + rd_kafka_buf_write_kstr(rkbuf, protocol_type); + rd_kafka_buf_write_i32(rkbuf, rk->rk_conf.enabled_assignor_cnt); + + RD_LIST_FOREACH(rkas, &rk->rk_conf.partition_assignors, i) { + rd_kafkap_bytes_t *member_metadata; + if (!rkas->rkas_enabled) + continue; + rd_kafka_buf_write_kstr(rkbuf, rkas->rkas_protocol_name); + member_metadata = rkas->rkas_get_metadata_cb( + rkas, rk->rk_cgrp->rkcg_assignor_state, topics, + rk->rk_cgrp->rkcg_group_assignment); + rd_kafka_buf_write_kbytes(rkbuf, member_metadata); + rd_kafkap_bytes_destroy(member_metadata); + } + + rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0); + + if (ApiVersion < 1 && + rk->rk_conf.max_poll_interval_ms > + rk->rk_conf.group_session_timeout_ms && + rd_interval(&rkb->rkb_suppress.unsupported_kip62, + /* at most once per day */ + (rd_ts_t)86400 * 1000 * 1000, 0) > 0) + rd_rkb_log(rkb, LOG_NOTICE, "MAXPOLL", + "Broker does not support KIP-62 " + "(requires Apache Kafka >= v0.10.1.0): " + "consumer configuration " + "`max.poll.interval.ms` (%d) " + "is effectively limited " + "by `session.timeout.ms` (%d) " + "with this broker version", + rk->rk_conf.max_poll_interval_ms, + rk->rk_conf.group_session_timeout_ms); + + + if (ApiVersion < 5 && rk->rk_conf.group_instance_id && + rd_interval(&rkb->rkb_suppress.unsupported_kip345, + /* at most once per day */ + (rd_ts_t)86400 * 1000 * 1000, 0) > 0) + rd_rkb_log(rkb, LOG_NOTICE, "STATICMEMBER", + "Broker does not support KIP-345 " + "(requires Apache Kafka >= v2.3.0): " + "consumer configuration " + "`group.instance.id` (%s) " + "will not take effect", + rk->rk_conf.group_instance_id); + + /* Absolute timeout */ + rd_kafka_buf_set_abs_timeout_force( + rkbuf, + /* Request timeout is max.poll.interval.ms + grace + * if the broker supports it, else + * session.timeout.ms + grace. */ + (ApiVersion >= 1 ? rk->rk_conf.max_poll_interval_ms + : rk->rk_conf.group_session_timeout_ms) + + 3000 /* 3s grace period*/, + 0); + + /* This is a blocking request */ + rkbuf->rkbuf_flags |= RD_KAFKA_OP_F_BLOCKING; + + rd_kafka_broker_buf_enq_replyq(rkb, rkbuf, replyq, resp_cb, opaque); +} + + + +/** + * Send LeaveGroupRequest + */ +void rd_kafka_LeaveGroupRequest(rd_kafka_broker_t *rkb, + const char *group_id, + const char *member_id, + rd_kafka_replyq_t replyq, + rd_kafka_resp_cb_t *resp_cb, + void *opaque) { + rd_kafka_buf_t *rkbuf; + int16_t ApiVersion = 0; + int features; + + ApiVersion = rd_kafka_broker_ApiVersion_supported( + rkb, RD_KAFKAP_LeaveGroup, 0, 1, &features); + + rkbuf = rd_kafka_buf_new_request(rkb, RD_KAFKAP_LeaveGroup, 1, 300); + + rd_kafka_buf_write_str(rkbuf, group_id, -1); + rd_kafka_buf_write_str(rkbuf, member_id, -1); + + rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0); + + /* LeaveGroupRequests are best-effort, the local consumer + * does not care if it succeeds or not, so the request timeout + * is shortened. + * Retries are not needed. */ + rd_kafka_buf_set_abs_timeout(rkbuf, 5000, 0); + rkbuf->rkbuf_max_retries = RD_KAFKA_REQUEST_NO_RETRIES; + + rd_kafka_broker_buf_enq_replyq(rkb, rkbuf, replyq, resp_cb, opaque); +} + + +/** + * Handler for LeaveGroup responses + * opaque must be the cgrp handle. + */ +void rd_kafka_handle_LeaveGroup(rd_kafka_t *rk, + rd_kafka_broker_t *rkb, + rd_kafka_resp_err_t err, + rd_kafka_buf_t *rkbuf, + rd_kafka_buf_t *request, + void *opaque) { + rd_kafka_cgrp_t *rkcg = opaque; + const int log_decode_errors = LOG_ERR; + int16_t ErrorCode = 0; + int actions; + + if (err) { + ErrorCode = err; + goto err; + } + + rd_kafka_buf_read_i16(rkbuf, &ErrorCode); + +err: + actions = rd_kafka_err_action(rkb, ErrorCode, request, + RD_KAFKA_ERR_ACTION_END); + + if (actions & RD_KAFKA_ERR_ACTION_REFRESH) { + /* Re-query for coordinator */ + rd_kafka_cgrp_op(rkcg, NULL, RD_KAFKA_NO_REPLYQ, + RD_KAFKA_OP_COORD_QUERY, ErrorCode); + } + + if (actions & RD_KAFKA_ERR_ACTION_RETRY) { + if (rd_kafka_buf_retry(rkb, request)) + return; + /* FALLTHRU */ + } + + if (ErrorCode) + rd_kafka_dbg(rkb->rkb_rk, CGRP, "LEAVEGROUP", + "LeaveGroup response: %s", + rd_kafka_err2str(ErrorCode)); + + return; + +err_parse: + ErrorCode = rkbuf->rkbuf_err; + goto err; +} + + + +/** + * Send HeartbeatRequest + */ +void rd_kafka_HeartbeatRequest(rd_kafka_broker_t *rkb, + const rd_kafkap_str_t *group_id, + int32_t generation_id, + const rd_kafkap_str_t *member_id, + const rd_kafkap_str_t *group_instance_id, + rd_kafka_replyq_t replyq, + rd_kafka_resp_cb_t *resp_cb, + void *opaque) { + rd_kafka_buf_t *rkbuf; + int16_t ApiVersion = 0; + int features; + + ApiVersion = rd_kafka_broker_ApiVersion_supported( + rkb, RD_KAFKAP_Heartbeat, 0, 3, &features); + + rd_rkb_dbg(rkb, CGRP, "HEARTBEAT", + "Heartbeat for group \"%s\" generation id %" PRId32, + group_id->str, generation_id); + + rkbuf = rd_kafka_buf_new_request(rkb, RD_KAFKAP_Heartbeat, 1, + RD_KAFKAP_STR_SIZE(group_id) + + 4 /* GenerationId */ + + RD_KAFKAP_STR_SIZE(member_id)); + + rd_kafka_buf_write_kstr(rkbuf, group_id); + rd_kafka_buf_write_i32(rkbuf, generation_id); + rd_kafka_buf_write_kstr(rkbuf, member_id); + if (ApiVersion >= 3) + rd_kafka_buf_write_kstr(rkbuf, group_instance_id); + + rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0); + + rd_kafka_buf_set_abs_timeout( + rkbuf, rkb->rkb_rk->rk_conf.group_session_timeout_ms, 0); + + rd_kafka_broker_buf_enq_replyq(rkb, rkbuf, replyq, resp_cb, opaque); +} + + + +/** + * @brief Construct and send ListGroupsRequest to \p rkb + * with the states (const char *) in \p states. + * Uses \p max_ApiVersion as maximum API version, + * pass -1 to use the maximum available version. + * + * The response (unparsed) will be enqueued on \p replyq + * for handling by \p resp_cb (with \p opaque passed). + * + * @return NULL on success, a new error instance that must be + * released with rd_kafka_error_destroy() in case of error. + */ +rd_kafka_error_t *rd_kafka_ListGroupsRequest(rd_kafka_broker_t *rkb, + int16_t max_ApiVersion, + const char **states, + size_t states_cnt, + rd_kafka_replyq_t replyq, + rd_kafka_resp_cb_t *resp_cb, + void *opaque) { + rd_kafka_buf_t *rkbuf; + int16_t ApiVersion = 0; + size_t i; + + if (max_ApiVersion < 0) + max_ApiVersion = 4; + + if (max_ApiVersion > ApiVersion) { + /* Remark: don't check if max_ApiVersion is zero. + * As rd_kafka_broker_ApiVersion_supported cannot be checked + * in the application thread reliably . */ + ApiVersion = rd_kafka_broker_ApiVersion_supported( + rkb, RD_KAFKAP_ListGroups, 0, max_ApiVersion, NULL); + } + + if (ApiVersion == -1) { + return rd_kafka_error_new( + RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE, + "ListGroupsRequest not supported by broker"); + } + + rkbuf = rd_kafka_buf_new_flexver_request( + rkb, RD_KAFKAP_ListGroups, 1, + /* rd_kafka_buf_write_arraycnt_pos + tags + StatesFilter */ + 4 + 1 + 32 * states_cnt, ApiVersion >= 3 /* is_flexver */); + + if (ApiVersion >= 4) { + size_t of_GroupsArrayCnt = + rd_kafka_buf_write_arraycnt_pos(rkbuf); + for (i = 0; i < states_cnt; i++) { + rd_kafka_buf_write_str(rkbuf, states[i], -1); + } + rd_kafka_buf_finalize_arraycnt(rkbuf, of_GroupsArrayCnt, i); + } + + rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0); + rd_kafka_broker_buf_enq_replyq(rkb, rkbuf, replyq, resp_cb, opaque); + return NULL; +} + +/** + * @brief Construct and send DescribeGroupsRequest to \p rkb + * with the groups (const char *) in \p groups. + * Uses \p max_ApiVersion as maximum API version, + * pass -1 to use the maximum available version. + * + * The response (unparsed) will be enqueued on \p replyq + * for handling by \p resp_cb (with \p opaque passed). + * + * @return NULL on success, a new error instance that must be + * released with rd_kafka_error_destroy() in case of error. + */ +rd_kafka_error_t *rd_kafka_DescribeGroupsRequest(rd_kafka_broker_t *rkb, + int16_t max_ApiVersion, + char **groups, + size_t group_cnt, + rd_kafka_replyq_t replyq, + rd_kafka_resp_cb_t *resp_cb, + void *opaque) { + rd_kafka_buf_t *rkbuf; + int16_t ApiVersion = 0; + size_t of_GroupsArrayCnt; + + if (max_ApiVersion < 0) + max_ApiVersion = 4; + + if (max_ApiVersion > ApiVersion) { + /* Remark: don't check if max_ApiVersion is zero. + * As rd_kafka_broker_ApiVersion_supported cannot be checked + * in the application thread reliably . */ + ApiVersion = rd_kafka_broker_ApiVersion_supported( + rkb, RD_KAFKAP_DescribeGroups, 0, max_ApiVersion, NULL); + } + + if (ApiVersion == -1) { + return rd_kafka_error_new( + RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE, + "DescribeGroupsRequest not supported by broker"); + } + + rkbuf = rd_kafka_buf_new_flexver_request( + rkb, RD_KAFKAP_DescribeGroups, 1, + 4 /* rd_kafka_buf_write_arraycnt_pos */ + + 1 /* IncludeAuthorizedOperations */ + 1 /* tags */ + + 32 * group_cnt /* Groups */, + rd_false); + + /* write Groups */ + of_GroupsArrayCnt = rd_kafka_buf_write_arraycnt_pos(rkbuf); + rd_kafka_buf_finalize_arraycnt(rkbuf, of_GroupsArrayCnt, group_cnt); + while (group_cnt-- > 0) + rd_kafka_buf_write_str(rkbuf, groups[group_cnt], -1); + + /* write IncludeAuthorizedOperations */ + if (ApiVersion >= 3) { + /* TODO: implement KIP-430 */ + rd_kafka_buf_write_bool(rkbuf, rd_false); + } + + rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0); + rd_kafka_broker_buf_enq_replyq(rkb, rkbuf, replyq, resp_cb, opaque); + return NULL; +} + +/** + * @brief Generic handler for Metadata responses + * + * @locality rdkafka main thread + */ +static void rd_kafka_handle_Metadata(rd_kafka_t *rk, + rd_kafka_broker_t *rkb, + rd_kafka_resp_err_t err, + rd_kafka_buf_t *rkbuf, + rd_kafka_buf_t *request, + void *opaque) { + rd_kafka_op_t *rko = opaque; /* Possibly NULL */ + struct rd_kafka_metadata *md = NULL; + const rd_list_t *topics = request->rkbuf_u.Metadata.topics; + int actions; + + rd_kafka_assert(NULL, err == RD_KAFKA_RESP_ERR__DESTROY || + thrd_is_current(rk->rk_thread)); + + /* Avoid metadata updates when we're terminating. */ + if (rd_kafka_terminating(rkb->rkb_rk) || + err == RD_KAFKA_RESP_ERR__DESTROY) { + /* Terminating */ + goto done; + } + + if (err) + goto err; + + if (!topics) + rd_rkb_dbg(rkb, METADATA, "METADATA", + "===== Received metadata: %s =====", + request->rkbuf_u.Metadata.reason); + else + rd_rkb_dbg(rkb, METADATA, "METADATA", + "===== Received metadata " + "(for %d requested topics): %s =====", + rd_list_cnt(topics), + request->rkbuf_u.Metadata.reason); + + err = rd_kafka_parse_Metadata(rkb, request, rkbuf, &md); + if (err) + goto err; + + if (rko && rko->rko_replyq.q) { + /* Reply to metadata requester, passing on the metadata. + * Reuse requesting rko for the reply. */ + rko->rko_err = err; + rko->rko_u.metadata.md = md; + + rd_kafka_replyq_enq(&rko->rko_replyq, rko, 0); + rko = NULL; + } else { + if (md) + rd_free(md); + } + + goto done; + +err: + actions = rd_kafka_err_action(rkb, err, request, + + RD_KAFKA_ERR_ACTION_RETRY, + RD_KAFKA_RESP_ERR__PARTIAL, + + RD_KAFKA_ERR_ACTION_END); + + if (actions & RD_KAFKA_ERR_ACTION_RETRY) { + if (rd_kafka_buf_retry(rkb, request)) + return; + /* FALLTHRU */ + } else { + rd_rkb_log(rkb, LOG_WARNING, "METADATA", + "Metadata request failed: %s: %s (%dms): %s", + request->rkbuf_u.Metadata.reason, + rd_kafka_err2str(err), + (int)(request->rkbuf_ts_sent / 1000), + rd_kafka_actions2str(actions)); + /* Respond back to caller on non-retriable errors */ + if (rko && rko->rko_replyq.q) { + rko->rko_err = err; + rko->rko_u.metadata.md = NULL; + rd_kafka_replyq_enq(&rko->rko_replyq, rko, 0); + rko = NULL; + } + } + + + + /* FALLTHRU */ + +done: + if (rko) + rd_kafka_op_destroy(rko); +} + + +/** + * @brief Construct MetadataRequest (does not send) + * + * \p topics is a list of topic names (char *) to request. + * + * !topics - only request brokers (if supported by broker, else + * all topics) + * topics.cnt==0 - all topics in cluster are requested + * topics.cnt >0 - only specified topics are requested + * + * @param reason - metadata request reason + * @param allow_auto_create_topics - allow broker-side auto topic creation. + * This is best-effort, depending on broker + * config and version. + * @param cgrp_update - Update cgrp in parse_Metadata (see comment there). + * @param rko - (optional) rko with replyq for handling response. + * Specifying an rko forces a metadata request even if + * there is already a matching one in-transit. + * + * If full metadata for all topics is requested (or all brokers, which + * results in all-topics on older brokers) and there is already a full request + * in transit then this function will return RD_KAFKA_RESP_ERR__PREV_IN_PROGRESS + * otherwise RD_KAFKA_RESP_ERR_NO_ERROR. If \p rko is non-NULL the request + * is sent regardless. + */ +rd_kafka_resp_err_t rd_kafka_MetadataRequest(rd_kafka_broker_t *rkb, + const rd_list_t *topics, + const char *reason, + rd_bool_t allow_auto_create_topics, + rd_bool_t cgrp_update, + rd_kafka_op_t *rko) { + rd_kafka_buf_t *rkbuf; + int16_t ApiVersion = 0; + size_t of_TopicArrayCnt; + int features; + int topic_cnt = topics ? rd_list_cnt(topics) : 0; + int *full_incr = NULL; + + ApiVersion = rd_kafka_broker_ApiVersion_supported( + rkb, RD_KAFKAP_Metadata, 0, 9, &features); + + rkbuf = rd_kafka_buf_new_flexver_request(rkb, RD_KAFKAP_Metadata, 1, + 4 + (50 * topic_cnt) + 1, + ApiVersion >= 9); + + if (!reason) + reason = ""; + + rkbuf->rkbuf_u.Metadata.reason = rd_strdup(reason); + rkbuf->rkbuf_u.Metadata.cgrp_update = cgrp_update; + + /* TopicArrayCnt */ + of_TopicArrayCnt = rd_kafka_buf_write_arraycnt_pos(rkbuf); + + if (!topics) { + /* v0: keep 0, brokers only not available, + * request all topics */ + /* v1-8: 0 means empty array, brokers only */ + if (ApiVersion >= 9) { + /* v9+: varint encoded empty array (1), brokers only */ + rd_kafka_buf_finalize_arraycnt(rkbuf, of_TopicArrayCnt, + topic_cnt); + } + + rd_rkb_dbg(rkb, METADATA, "METADATA", + "Request metadata for brokers only: %s", reason); + full_incr = + &rkb->rkb_rk->rk_metadata_cache.rkmc_full_brokers_sent; + + } else if (topic_cnt == 0) { + /* v0: keep 0, request all topics */ + if (ApiVersion >= 1 && ApiVersion < 9) { + /* v1-8: update to -1, all topics */ + rd_kafka_buf_update_i32(rkbuf, of_TopicArrayCnt, -1); + } + /* v9+: keep 0, varint encoded null, all topics */ + + rkbuf->rkbuf_u.Metadata.all_topics = 1; + rd_rkb_dbg(rkb, METADATA, "METADATA", + "Request metadata for all topics: " + "%s", + reason); + + if (!rko) + full_incr = &rkb->rkb_rk->rk_metadata_cache + .rkmc_full_topics_sent; + + } else { + /* request cnt topics */ + rd_kafka_buf_finalize_arraycnt(rkbuf, of_TopicArrayCnt, + topic_cnt); + + rd_rkb_dbg(rkb, METADATA, "METADATA", + "Request metadata for %d topic(s): " + "%s", + topic_cnt, reason); + } + + if (full_incr) { + /* Avoid multiple outstanding full requests + * (since they are redundant and side-effect-less). + * Forced requests (app using metadata() API) are passed + * through regardless. */ + + mtx_lock(&rkb->rkb_rk->rk_metadata_cache.rkmc_full_lock); + if (*full_incr > 0 && (!rko || !rko->rko_u.metadata.force)) { + mtx_unlock( + &rkb->rkb_rk->rk_metadata_cache.rkmc_full_lock); + rd_rkb_dbg(rkb, METADATA, "METADATA", + "Skipping metadata request: %s: " + "full request already in-transit", + reason); + rd_kafka_buf_destroy(rkbuf); + return RD_KAFKA_RESP_ERR__PREV_IN_PROGRESS; + } + + (*full_incr)++; + mtx_unlock(&rkb->rkb_rk->rk_metadata_cache.rkmc_full_lock); + rkbuf->rkbuf_u.Metadata.decr = full_incr; + rkbuf->rkbuf_u.Metadata.decr_lock = + &rkb->rkb_rk->rk_metadata_cache.rkmc_full_lock; + } + + + if (topic_cnt > 0) { + char *topic; + int i; + + /* Maintain a copy of the topics list so we can purge + * hints from the metadata cache on error. */ + rkbuf->rkbuf_u.Metadata.topics = + rd_list_copy(topics, rd_list_string_copy, NULL); + + RD_LIST_FOREACH(topic, topics, i) { + rd_kafka_buf_write_str(rkbuf, topic, -1); + /* Tags for previous topic */ + rd_kafka_buf_write_tags(rkbuf); + } + } + + if (ApiVersion >= 4) { + /* AllowAutoTopicCreation */ + rd_kafka_buf_write_bool(rkbuf, allow_auto_create_topics); + + } else if (rkb->rkb_rk->rk_type == RD_KAFKA_CONSUMER && + !rkb->rkb_rk->rk_conf.allow_auto_create_topics && + rd_kafka_conf_is_modified(&rkb->rkb_rk->rk_conf, + "allow.auto.create.topics") && + rd_interval( + &rkb->rkb_rk->rk_suppress.allow_auto_create_topics, + 30 * 60 * 1000 /* every 30 minutes */, 0) >= 0) { + /* Let user know we can't obey allow.auto.create.topics */ + rd_rkb_log(rkb, LOG_WARNING, "AUTOCREATE", + "allow.auto.create.topics=false not supported " + "by broker: requires broker version >= 0.11.0.0: " + "requested topic(s) may be auto created depending " + "on broker auto.create.topics.enable configuration"); + } + + if (ApiVersion >= 8 && ApiVersion < 10) { + /* TODO: implement KIP-430 */ + /* IncludeClusterAuthorizedOperations */ + rd_kafka_buf_write_bool(rkbuf, rd_false); + } + + if (ApiVersion >= 8) { + /* TODO: implement KIP-430 */ + /* IncludeTopicAuthorizedOperations */ + rd_kafka_buf_write_bool(rkbuf, rd_false); + } + + rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0); + + /* Metadata requests are part of the important control plane + * and should go before most other requests (Produce, Fetch, etc). */ + rkbuf->rkbuf_prio = RD_KAFKA_PRIO_HIGH; + + rd_kafka_broker_buf_enq_replyq(rkb, rkbuf, + /* Handle response thru rk_ops, + * but forward parsed result to + * rko's replyq when done. */ + RD_KAFKA_REPLYQ(rkb->rkb_rk->rk_ops, 0), + rd_kafka_handle_Metadata, rko); + + return RD_KAFKA_RESP_ERR_NO_ERROR; +} + + + +/** + * @brief Parses and handles ApiVersion reply. + * + * @param apis will be allocated, populated and sorted + * with broker's supported APIs, or set to NULL. + * @param api_cnt will be set to the number of elements in \p *apis + * + * @returns 0 on success, else an error. + * + * @remark A valid \p apis might be returned even if an error is returned. + */ +rd_kafka_resp_err_t +rd_kafka_handle_ApiVersion(rd_kafka_t *rk, + rd_kafka_broker_t *rkb, + rd_kafka_resp_err_t err, + rd_kafka_buf_t *rkbuf, + rd_kafka_buf_t *request, + struct rd_kafka_ApiVersion **apis, + size_t *api_cnt) { + const int log_decode_errors = LOG_DEBUG; + int32_t ApiArrayCnt; + int16_t ErrorCode; + int i = 0; + + *apis = NULL; + *api_cnt = 0; + + if (err) + goto err; + + rd_kafka_buf_read_i16(rkbuf, &ErrorCode); + err = ErrorCode; + + rd_kafka_buf_read_arraycnt(rkbuf, &ApiArrayCnt, 1000); + if (err && ApiArrayCnt < 1) { + /* Version >=3 returns the ApiVersions array if the error + * code is ERR_UNSUPPORTED_VERSION, previous versions don't */ + goto err; + } + + rd_rkb_dbg(rkb, FEATURE, "APIVERSION", "Broker API support:"); + + *apis = rd_malloc(sizeof(**apis) * ApiArrayCnt); + + for (i = 0; i < ApiArrayCnt; i++) { + struct rd_kafka_ApiVersion *api = &(*apis)[i]; + + rd_kafka_buf_read_i16(rkbuf, &api->ApiKey); + rd_kafka_buf_read_i16(rkbuf, &api->MinVer); + rd_kafka_buf_read_i16(rkbuf, &api->MaxVer); + + rd_rkb_dbg(rkb, FEATURE, "APIVERSION", + " ApiKey %s (%hd) Versions %hd..%hd", + rd_kafka_ApiKey2str(api->ApiKey), api->ApiKey, + api->MinVer, api->MaxVer); + + /* Discard struct tags */ + rd_kafka_buf_skip_tags(rkbuf); + } + + if (request->rkbuf_reqhdr.ApiVersion >= 1) + rd_kafka_buf_read_throttle_time(rkbuf); + + /* Discard end tags */ + rd_kafka_buf_skip_tags(rkbuf); + + *api_cnt = ApiArrayCnt; + qsort(*apis, *api_cnt, sizeof(**apis), rd_kafka_ApiVersion_key_cmp); + + goto done; + +err_parse: + /* If the broker does not support our ApiVersionRequest version it + * will respond with a version 0 response, which will most likely + * fail parsing. Instead of propagating the parse error we + * propagate the original error, unless there isn't one in which case + * we use the parse error. */ + if (!err) + err = rkbuf->rkbuf_err; +err: + /* There are no retryable errors. */ + + if (*apis) + rd_free(*apis); + + *apis = NULL; + *api_cnt = 0; + +done: + return err; +} + + + +/** + * @brief Send ApiVersionRequest (KIP-35) + * + * @param ApiVersion If -1 use the highest supported version, else use the + * specified value. + */ +void rd_kafka_ApiVersionRequest(rd_kafka_broker_t *rkb, + int16_t ApiVersion, + rd_kafka_replyq_t replyq, + rd_kafka_resp_cb_t *resp_cb, + void *opaque) { + rd_kafka_buf_t *rkbuf; + + if (ApiVersion == -1) + ApiVersion = 3; + + rkbuf = rd_kafka_buf_new_flexver_request( + rkb, RD_KAFKAP_ApiVersion, 1, 3, ApiVersion >= 3 /*flexver*/); + + if (ApiVersion >= 3) { + /* KIP-511 adds software name and version through the optional + * protocol fields defined in KIP-482. */ + + /* ClientSoftwareName */ + rd_kafka_buf_write_str(rkbuf, rkb->rkb_rk->rk_conf.sw_name, -1); + + /* ClientSoftwareVersion */ + rd_kafka_buf_write_str(rkbuf, rkb->rkb_rk->rk_conf.sw_version, + -1); + } + + /* Should be sent before any other requests since it is part of + * the initial connection handshake. */ + rkbuf->rkbuf_prio = RD_KAFKA_PRIO_FLASH; + + /* Non-supporting brokers will tear down the connection when they + * receive an unknown API request, so dont retry request on failure. */ + rkbuf->rkbuf_max_retries = RD_KAFKA_REQUEST_NO_RETRIES; + + /* 0.9.0.x brokers will not close the connection on unsupported + * API requests, so we minimize the timeout for the request. + * This is a regression on the broker part. */ + rd_kafka_buf_set_abs_timeout( + rkbuf, rkb->rkb_rk->rk_conf.api_version_request_timeout_ms, 0); + + rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0); + + if (replyq.q) + rd_kafka_broker_buf_enq_replyq(rkb, rkbuf, replyq, resp_cb, + opaque); + else /* in broker thread */ + rd_kafka_broker_buf_enq1(rkb, rkbuf, resp_cb, opaque); +} + + +/** + * Send SaslHandshakeRequest (KIP-43) + */ +void rd_kafka_SaslHandshakeRequest(rd_kafka_broker_t *rkb, + const char *mechanism, + rd_kafka_replyq_t replyq, + rd_kafka_resp_cb_t *resp_cb, + void *opaque) { + rd_kafka_buf_t *rkbuf; + int mechlen = (int)strlen(mechanism); + int16_t ApiVersion; + int features; + + rkbuf = rd_kafka_buf_new_request(rkb, RD_KAFKAP_SaslHandshake, 1, + RD_KAFKAP_STR_SIZE0(mechlen)); + + /* Should be sent before any other requests since it is part of + * the initial connection handshake. */ + rkbuf->rkbuf_prio = RD_KAFKA_PRIO_FLASH; + + rd_kafka_buf_write_str(rkbuf, mechanism, mechlen); + + /* Non-supporting brokers will tear down the conneciton when they + * receive an unknown API request or where the SASL GSSAPI + * token type is not recognized, so dont retry request on failure. */ + rkbuf->rkbuf_max_retries = RD_KAFKA_REQUEST_NO_RETRIES; + + /* 0.9.0.x brokers will not close the connection on unsupported + * API requests, so we minimize the timeout of the request. + * This is a regression on the broker part. */ + if (!rkb->rkb_rk->rk_conf.api_version_request && + rkb->rkb_rk->rk_conf.socket_timeout_ms > 10 * 1000) + rd_kafka_buf_set_abs_timeout(rkbuf, 10 * 1000 /*10s*/, 0); + + /* ApiVersion 1 / RD_KAFKA_FEATURE_SASL_REQ enables + * the SaslAuthenticateRequest */ + ApiVersion = rd_kafka_broker_ApiVersion_supported( + rkb, RD_KAFKAP_SaslHandshake, 0, 1, &features); + + rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0); + + if (replyq.q) + rd_kafka_broker_buf_enq_replyq(rkb, rkbuf, replyq, resp_cb, + opaque); + else /* in broker thread */ + rd_kafka_broker_buf_enq1(rkb, rkbuf, resp_cb, opaque); +} + + +/** + * @brief Parses and handles an SaslAuthenticate reply. + * + * @returns 0 on success, else an error. + * + * @locality broker thread + * @locks none + */ +void rd_kafka_handle_SaslAuthenticate(rd_kafka_t *rk, + rd_kafka_broker_t *rkb, + rd_kafka_resp_err_t err, + rd_kafka_buf_t *rkbuf, + rd_kafka_buf_t *request, + void *opaque) { + const int log_decode_errors = LOG_ERR; + int16_t error_code; + rd_kafkap_str_t error_str; + rd_kafkap_bytes_t auth_data; + char errstr[512]; + + if (err) { + rd_snprintf(errstr, sizeof(errstr), + "SaslAuthenticateRequest failed: %s", + rd_kafka_err2str(err)); + goto err; + } + + rd_kafka_buf_read_i16(rkbuf, &error_code); + rd_kafka_buf_read_str(rkbuf, &error_str); + + if (error_code) { + /* Authentication failed */ + + /* For backwards compatibility translate the + * new broker-side auth error code to our local error code. */ + if (error_code == RD_KAFKA_RESP_ERR_SASL_AUTHENTICATION_FAILED) + err = RD_KAFKA_RESP_ERR__AUTHENTICATION; + else + err = error_code; + + rd_snprintf(errstr, sizeof(errstr), "%.*s", + RD_KAFKAP_STR_PR(&error_str)); + goto err; + } + + rd_kafka_buf_read_bytes(rkbuf, &auth_data); + + /* Pass SASL auth frame to SASL handler */ + if (rd_kafka_sasl_recv(rkb->rkb_transport, auth_data.data, + (size_t)RD_KAFKAP_BYTES_LEN(&auth_data), errstr, + sizeof(errstr)) == -1) { + err = RD_KAFKA_RESP_ERR__AUTHENTICATION; + goto err; + } + + return; + + +err_parse: + err = rkbuf->rkbuf_err; + rd_snprintf(errstr, sizeof(errstr), + "SaslAuthenticateResponse parsing failed: %s", + rd_kafka_err2str(err)); + +err: + rd_kafka_broker_fail(rkb, LOG_ERR, err, "SASL authentication error: %s", + errstr); +} + + +/** + * @brief Send SaslAuthenticateRequest (KIP-152) + */ +void rd_kafka_SaslAuthenticateRequest(rd_kafka_broker_t *rkb, + const void *buf, + size_t size, + rd_kafka_replyq_t replyq, + rd_kafka_resp_cb_t *resp_cb, + void *opaque) { + rd_kafka_buf_t *rkbuf; + + rkbuf = rd_kafka_buf_new_request(rkb, RD_KAFKAP_SaslAuthenticate, 0, 0); + + /* Should be sent before any other requests since it is part of + * the initial connection handshake. */ + rkbuf->rkbuf_prio = RD_KAFKA_PRIO_FLASH; + + /* Broker does not support -1 (Null) for this field */ + rd_kafka_buf_write_bytes(rkbuf, buf ? buf : "", size); + + /* There are no errors that can be retried, instead + * close down the connection and reconnect on failure. */ + rkbuf->rkbuf_max_retries = RD_KAFKA_REQUEST_NO_RETRIES; + + if (replyq.q) + rd_kafka_broker_buf_enq_replyq(rkb, rkbuf, replyq, resp_cb, + opaque); + else /* in broker thread */ + rd_kafka_broker_buf_enq1(rkb, rkbuf, resp_cb, opaque); +} + + + +/** + * @struct Hold temporary result and return values from ProduceResponse + */ +struct rd_kafka_Produce_result { + int64_t offset; /**< Assigned offset of first message */ + int64_t timestamp; /**< (Possibly assigned) offset of first message */ +}; + +/** + * @brief Parses a Produce reply. + * @returns 0 on success or an error code on failure. + * @locality broker thread + */ +static rd_kafka_resp_err_t +rd_kafka_handle_Produce_parse(rd_kafka_broker_t *rkb, + rd_kafka_toppar_t *rktp, + rd_kafka_buf_t *rkbuf, + rd_kafka_buf_t *request, + struct rd_kafka_Produce_result *result) { + int32_t TopicArrayCnt; + int32_t PartitionArrayCnt; + struct { + int32_t Partition; + int16_t ErrorCode; + int64_t Offset; + } hdr; + const int log_decode_errors = LOG_ERR; + int64_t log_start_offset = -1; + + rd_kafka_buf_read_i32(rkbuf, &TopicArrayCnt); + if (TopicArrayCnt != 1) + goto err; + + /* Since we only produce to one single topic+partition in each + * request we assume that the reply only contains one topic+partition + * and that it is the same that we requested. + * If not the broker is buggy. */ + rd_kafka_buf_skip_str(rkbuf); + rd_kafka_buf_read_i32(rkbuf, &PartitionArrayCnt); + + if (PartitionArrayCnt != 1) + goto err; + + rd_kafka_buf_read_i32(rkbuf, &hdr.Partition); + rd_kafka_buf_read_i16(rkbuf, &hdr.ErrorCode); + rd_kafka_buf_read_i64(rkbuf, &hdr.Offset); + + result->offset = hdr.Offset; + + result->timestamp = -1; + if (request->rkbuf_reqhdr.ApiVersion >= 2) + rd_kafka_buf_read_i64(rkbuf, &result->timestamp); + + if (request->rkbuf_reqhdr.ApiVersion >= 5) + rd_kafka_buf_read_i64(rkbuf, &log_start_offset); + + 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); + } + + + return hdr.ErrorCode; + +err_parse: + return rkbuf->rkbuf_err; +err: + return RD_KAFKA_RESP_ERR__BAD_MSG; +} + + +/** + * @struct Hold temporary Produce error state + */ +struct rd_kafka_Produce_err { + rd_kafka_resp_err_t err; /**< Error code */ + int actions; /**< Actions to take */ + int incr_retry; /**< Increase per-message retry cnt */ + rd_kafka_msg_status_t status; /**< Messages persistence status */ + + /* Idempotent Producer */ + int32_t next_ack_seq; /**< Next expected sequence to ack */ + int32_t next_err_seq; /**< Next expected error sequence */ + rd_bool_t update_next_ack; /**< Update next_ack_seq */ + rd_bool_t update_next_err; /**< Update next_err_seq */ + rd_kafka_pid_t rktp_pid; /**< Partition's current PID */ + int32_t last_seq; /**< Last sequence in current batch */ +}; + + +/** + * @brief Error-handling for Idempotent Producer-specific Produce errors. + * + * May update \p errp, \p actionsp and \p incr_retryp. + * + * The resulting \p actionsp are handled by the caller. + * + * @warning May be called on the old leader thread. Lock rktp appropriately! + * + * @locality broker thread (but not necessarily the leader broker) + * @locks none + */ +static void +rd_kafka_handle_idempotent_Produce_error(rd_kafka_broker_t *rkb, + rd_kafka_msgbatch_t *batch, + struct rd_kafka_Produce_err *perr) { + rd_kafka_t *rk = rkb->rkb_rk; + rd_kafka_toppar_t *rktp = batch->rktp; + rd_kafka_msg_t *firstmsg, *lastmsg; + int r; + rd_ts_t now = rd_clock(), state_age; + struct rd_kafka_toppar_err last_err; + + rd_kafka_rdlock(rkb->rkb_rk); + state_age = now - rkb->rkb_rk->rk_eos.ts_idemp_state; + rd_kafka_rdunlock(rkb->rkb_rk); + + firstmsg = rd_kafka_msgq_first(&batch->msgq); + lastmsg = rd_kafka_msgq_last(&batch->msgq); + rd_assert(firstmsg && lastmsg); + + /* Store the last msgid of the batch + * on the first message in case we need to retry + * and thus reconstruct the entire batch. */ + if (firstmsg->rkm_u.producer.last_msgid) { + /* last_msgid already set, make sure it + * actually points to the last message. */ + rd_assert(firstmsg->rkm_u.producer.last_msgid == + lastmsg->rkm_u.producer.msgid); + } else { + firstmsg->rkm_u.producer.last_msgid = + lastmsg->rkm_u.producer.msgid; + } + + if (!rd_kafka_pid_eq(batch->pid, perr->rktp_pid)) { + /* Don't retry if PID changed since we can't + * guarantee correctness across PID sessions. */ + perr->actions = RD_KAFKA_ERR_ACTION_PERMANENT; + perr->status = RD_KAFKA_MSG_STATUS_POSSIBLY_PERSISTED; + + rd_rkb_dbg(rkb, MSG | RD_KAFKA_DBG_EOS, "ERRPID", + "%.*s [%" PRId32 + "] PID mismatch: " + "request %s != partition %s: " + "failing messages with error %s", + RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic), + rktp->rktp_partition, rd_kafka_pid2str(batch->pid), + rd_kafka_pid2str(perr->rktp_pid), + rd_kafka_err2str(perr->err)); + return; + } + + /* + * Special error handling + */ + switch (perr->err) { + case RD_KAFKA_RESP_ERR_OUT_OF_ORDER_SEQUENCE_NUMBER: + /* Compare request's sequence to expected next + * acked sequence. + * + * Example requests in flight: + * R1(base_seq:5) R2(10) R3(15) R4(20) + */ + + /* Acquire the last partition error to help + * troubleshoot this problem. */ + rd_kafka_toppar_lock(rktp); + last_err = rktp->rktp_last_err; + rd_kafka_toppar_unlock(rktp); + + r = batch->first_seq - perr->next_ack_seq; + + if (r == 0) { + /* R1 failed: + * If this was the head-of-line request in-flight it + * means there is a state desynchronization between the + * producer and broker (a bug), in which case + * we'll raise a fatal error since we can no longer + * reason about the state of messages and thus + * not guarantee ordering or once-ness for R1, + * nor give the user a chance to opt out of sending + * R2 to R4 which would be retried automatically. */ + + rd_kafka_idemp_set_fatal_error( + rk, perr->err, + "ProduceRequest for %.*s [%" PRId32 + "] " + "with %d message(s) failed " + "due to sequence desynchronization with " + "broker %" PRId32 " (%s, base seq %" PRId32 + ", " + "idemp state change %" PRId64 + "ms ago, " + "last partition error %s (actions %s, " + "base seq %" PRId32 "..%" PRId32 + ", base msgid %" PRIu64 ", %" PRId64 "ms ago)", + RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic), + rktp->rktp_partition, + rd_kafka_msgq_len(&batch->msgq), rkb->rkb_nodeid, + rd_kafka_pid2str(batch->pid), batch->first_seq, + state_age / 1000, rd_kafka_err2name(last_err.err), + rd_kafka_actions2str(last_err.actions), + last_err.base_seq, last_err.last_seq, + last_err.base_msgid, + last_err.ts ? (now - last_err.ts) / 1000 : -1); + + perr->actions = RD_KAFKA_ERR_ACTION_PERMANENT; + perr->status = RD_KAFKA_MSG_STATUS_POSSIBLY_PERSISTED; + perr->update_next_ack = rd_false; + perr->update_next_err = rd_true; + + } else if (r > 0) { + /* R2 failed: + * With max.in.flight > 1 we can have a situation + * where the first request in-flight (R1) to the broker + * fails, which causes the sub-sequent requests + * that are in-flight to have a non-sequential + * sequence number and thus fail. + * But these sub-sequent requests (R2 to R4) are not at + * the risk of being duplicated so we bump the epoch and + * re-enqueue the messages for later retry + * (without incrementing retries). + */ + rd_rkb_dbg( + rkb, MSG | RD_KAFKA_DBG_EOS, "ERRSEQ", + "ProduceRequest for %.*s [%" PRId32 + "] " + "with %d message(s) failed " + "due to skipped sequence numbers " + "(%s, base seq %" PRId32 + " > " + "next seq %" PRId32 + ") " + "caused by previous failed request " + "(%s, actions %s, " + "base seq %" PRId32 "..%" PRId32 + ", base msgid %" PRIu64 ", %" PRId64 + "ms ago): " + "recovering and retrying", + RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic), + rktp->rktp_partition, + rd_kafka_msgq_len(&batch->msgq), + rd_kafka_pid2str(batch->pid), batch->first_seq, + perr->next_ack_seq, rd_kafka_err2name(last_err.err), + rd_kafka_actions2str(last_err.actions), + last_err.base_seq, last_err.last_seq, + last_err.base_msgid, + last_err.ts ? (now - last_err.ts) / 1000 : -1); + + perr->incr_retry = 0; + perr->actions = RD_KAFKA_ERR_ACTION_RETRY; + perr->status = RD_KAFKA_MSG_STATUS_NOT_PERSISTED; + perr->update_next_ack = rd_false; + perr->update_next_err = rd_true; + + rd_kafka_idemp_drain_epoch_bump( + rk, perr->err, "skipped sequence numbers"); + + } else { + /* Request's sequence is less than next ack, + * this should never happen unless we have + * local bug or the broker did not respond + * to the requests in order. */ + rd_kafka_idemp_set_fatal_error( + rk, perr->err, + "ProduceRequest for %.*s [%" PRId32 + "] " + "with %d message(s) failed " + "with rewound sequence number on " + "broker %" PRId32 + " (%s, " + "base seq %" PRId32 " < next seq %" PRId32 + "): " + "last error %s (actions %s, " + "base seq %" PRId32 "..%" PRId32 + ", base msgid %" PRIu64 ", %" PRId64 "ms ago)", + RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic), + rktp->rktp_partition, + rd_kafka_msgq_len(&batch->msgq), rkb->rkb_nodeid, + rd_kafka_pid2str(batch->pid), batch->first_seq, + perr->next_ack_seq, rd_kafka_err2name(last_err.err), + rd_kafka_actions2str(last_err.actions), + last_err.base_seq, last_err.last_seq, + last_err.base_msgid, + last_err.ts ? (now - last_err.ts) / 1000 : -1); + + perr->actions = RD_KAFKA_ERR_ACTION_PERMANENT; + perr->status = RD_KAFKA_MSG_STATUS_POSSIBLY_PERSISTED; + perr->update_next_ack = rd_false; + perr->update_next_err = rd_false; + } + break; + + case RD_KAFKA_RESP_ERR_DUPLICATE_SEQUENCE_NUMBER: + /* This error indicates that we successfully produced + * this set of messages before but this (supposed) retry failed. + * + * Treat as success, however offset and timestamp + * will be invalid. */ + + /* Future improvement/FIXME: + * But first make sure the first message has actually + * been retried, getting this error for a non-retried message + * indicates a synchronization issue or bug. */ + rd_rkb_dbg(rkb, MSG | RD_KAFKA_DBG_EOS, "DUPSEQ", + "ProduceRequest for %.*s [%" PRId32 + "] " + "with %d message(s) failed " + "due to duplicate sequence number: " + "previous send succeeded but was not acknowledged " + "(%s, base seq %" PRId32 + "): " + "marking the messages successfully delivered", + RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic), + rktp->rktp_partition, + rd_kafka_msgq_len(&batch->msgq), + rd_kafka_pid2str(batch->pid), batch->first_seq); + + /* Void error, delivery succeeded */ + perr->err = RD_KAFKA_RESP_ERR_NO_ERROR; + perr->actions = 0; + perr->status = RD_KAFKA_MSG_STATUS_PERSISTED; + perr->update_next_ack = rd_true; + perr->update_next_err = rd_true; + break; + + case RD_KAFKA_RESP_ERR_UNKNOWN_PRODUCER_ID: + /* The broker/cluster lost track of our PID because + * the last message we produced has now been deleted + * (by DeleteRecords, compaction, or topic retention policy). + * + * If all previous messages are accounted for and this is not + * a retry we can simply bump the epoch and reset the sequence + * number and then retry the message(s) again. + * + * If there are outstanding messages not yet acknowledged + * then there is no safe way to carry on without risking + * duplication or reordering, in which case we fail + * the producer. + * + * In case of the transactional producer and a transaction + * coordinator that supports KIP-360 (>= AK 2.5, checked from + * the txnmgr, not here) we'll raise an abortable error and + * flag that the epoch needs to be bumped on the coordinator. */ + if (rd_kafka_is_transactional(rk)) { + rd_rkb_dbg(rkb, MSG | RD_KAFKA_DBG_EOS, "UNKPID", + "ProduceRequest for %.*s [%" PRId32 + "] " + "with %d message(s) failed " + "due to unknown producer id " + "(%s, base seq %" PRId32 + ", %d retries): " + "failing the current transaction", + RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic), + rktp->rktp_partition, + rd_kafka_msgq_len(&batch->msgq), + rd_kafka_pid2str(batch->pid), + batch->first_seq, + firstmsg->rkm_u.producer.retries); + + /* Drain outstanding requests and bump epoch. */ + rd_kafka_idemp_drain_epoch_bump(rk, perr->err, + "unknown producer id"); + + rd_kafka_txn_set_abortable_error_with_bump( + rk, RD_KAFKA_RESP_ERR_UNKNOWN_PRODUCER_ID, + "ProduceRequest for %.*s [%" PRId32 + "] " + "with %d message(s) failed " + "due to unknown producer id", + RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic), + rktp->rktp_partition, + rd_kafka_msgq_len(&batch->msgq)); + + perr->incr_retry = 0; + perr->actions = RD_KAFKA_ERR_ACTION_PERMANENT; + perr->status = RD_KAFKA_MSG_STATUS_NOT_PERSISTED; + perr->update_next_ack = rd_false; + perr->update_next_err = rd_true; + break; + + } else if (!firstmsg->rkm_u.producer.retries && + perr->next_err_seq == batch->first_seq) { + rd_rkb_dbg(rkb, MSG | RD_KAFKA_DBG_EOS, "UNKPID", + "ProduceRequest for %.*s [%" PRId32 + "] " + "with %d message(s) failed " + "due to unknown producer id " + "(%s, base seq %" PRId32 + ", %d retries): " + "no risk of duplication/reordering: " + "resetting PID and retrying", + RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic), + rktp->rktp_partition, + rd_kafka_msgq_len(&batch->msgq), + rd_kafka_pid2str(batch->pid), + batch->first_seq, + firstmsg->rkm_u.producer.retries); + + /* Drain outstanding requests and bump epoch. */ + rd_kafka_idemp_drain_epoch_bump(rk, perr->err, + "unknown producer id"); + + perr->incr_retry = 0; + perr->actions = RD_KAFKA_ERR_ACTION_RETRY; + perr->status = RD_KAFKA_MSG_STATUS_NOT_PERSISTED; + perr->update_next_ack = rd_false; + perr->update_next_err = rd_true; + break; + } + + rd_kafka_idemp_set_fatal_error( + rk, perr->err, + "ProduceRequest for %.*s [%" PRId32 + "] " + "with %d message(s) failed " + "due to unknown producer id (" + "broker %" PRId32 " %s, base seq %" PRId32 + ", %d retries): " + "unable to retry without risking " + "duplication/reordering", + RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic), + rktp->rktp_partition, rd_kafka_msgq_len(&batch->msgq), + rkb->rkb_nodeid, rd_kafka_pid2str(batch->pid), + batch->first_seq, firstmsg->rkm_u.producer.retries); + + perr->actions = RD_KAFKA_ERR_ACTION_PERMANENT; + perr->status = RD_KAFKA_MSG_STATUS_POSSIBLY_PERSISTED; + perr->update_next_ack = rd_false; + perr->update_next_err = rd_true; + break; + + default: + /* All other errors are handled in the standard + * error Produce handler, which will set + * update_next_ack|err accordingly. */ + break; + } +} + + + +/** + * @brief Error-handling for failed ProduceRequests + * + * @param errp Is the input and output error, it may be changed + * by this function. + * + * @returns 0 if no further processing of the request should be performed, + * such as triggering delivery reports, else 1. + * + * @warning May be called on the old leader thread. Lock rktp appropriately! + * + * @warning \p request may be NULL. + * + * @locality broker thread (but not necessarily the leader broker) + * @locks none + */ +static int rd_kafka_handle_Produce_error(rd_kafka_broker_t *rkb, + const rd_kafka_buf_t *request, + rd_kafka_msgbatch_t *batch, + struct rd_kafka_Produce_err *perr) { + rd_kafka_t *rk = rkb->rkb_rk; + rd_kafka_toppar_t *rktp = batch->rktp; + int is_leader; + + if (unlikely(perr->err == RD_KAFKA_RESP_ERR__DESTROY)) + return 0; /* Terminating */ + + /* When there is a partition leader change any outstanding + * requests to the old broker will be handled by the old + * broker thread when the responses are received/timeout: + * in this case we need to be careful with locking: + * check once if we're the leader (which allows relaxed + * locking), and cache the current rktp's eos state vars. */ + rd_kafka_toppar_lock(rktp); + is_leader = rktp->rktp_broker == rkb; + perr->rktp_pid = rktp->rktp_eos.pid; + perr->next_ack_seq = rktp->rktp_eos.next_ack_seq; + perr->next_err_seq = rktp->rktp_eos.next_err_seq; + rd_kafka_toppar_unlock(rktp); + + /* All failures are initially treated as if the message + * was not persisted, but the status may be changed later + * for specific errors and actions. */ + perr->status = RD_KAFKA_MSG_STATUS_NOT_PERSISTED; + + /* Set actions for known errors (may be overriden later), + * all other errors are considered permanent failures. + * (also see rd_kafka_err_action() for the default actions). */ + perr->actions = rd_kafka_err_action( + rkb, perr->err, request, + + RD_KAFKA_ERR_ACTION_REFRESH | + RD_KAFKA_ERR_ACTION_MSG_POSSIBLY_PERSISTED, + RD_KAFKA_RESP_ERR__TRANSPORT, + + RD_KAFKA_ERR_ACTION_REFRESH | RD_KAFKA_ERR_ACTION_MSG_NOT_PERSISTED, + RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PART, + + RD_KAFKA_ERR_ACTION_PERMANENT | + RD_KAFKA_ERR_ACTION_MSG_NOT_PERSISTED, + RD_KAFKA_RESP_ERR_TOPIC_AUTHORIZATION_FAILED, + + RD_KAFKA_ERR_ACTION_REFRESH | RD_KAFKA_ERR_ACTION_RETRY | + RD_KAFKA_ERR_ACTION_MSG_NOT_PERSISTED, + RD_KAFKA_RESP_ERR_KAFKA_STORAGE_ERROR, + + RD_KAFKA_ERR_ACTION_RETRY | RD_KAFKA_ERR_ACTION_MSG_NOT_PERSISTED, + RD_KAFKA_RESP_ERR_NOT_ENOUGH_REPLICAS, + + RD_KAFKA_ERR_ACTION_RETRY | + RD_KAFKA_ERR_ACTION_MSG_POSSIBLY_PERSISTED, + RD_KAFKA_RESP_ERR_NOT_ENOUGH_REPLICAS_AFTER_APPEND, + + RD_KAFKA_ERR_ACTION_RETRY | RD_KAFKA_ERR_ACTION_MSG_NOT_PERSISTED, + RD_KAFKA_RESP_ERR__TIMED_OUT_QUEUE, + + RD_KAFKA_ERR_ACTION_RETRY | + RD_KAFKA_ERR_ACTION_MSG_POSSIBLY_PERSISTED, + RD_KAFKA_RESP_ERR__TIMED_OUT, + + RD_KAFKA_ERR_ACTION_PERMANENT | + RD_KAFKA_ERR_ACTION_MSG_POSSIBLY_PERSISTED, + RD_KAFKA_RESP_ERR__MSG_TIMED_OUT, + + /* All Idempotent Producer-specific errors are + * initially set as permanent errors, + * special handling may change the actions. */ + RD_KAFKA_ERR_ACTION_PERMANENT | + RD_KAFKA_ERR_ACTION_MSG_POSSIBLY_PERSISTED, + RD_KAFKA_RESP_ERR_OUT_OF_ORDER_SEQUENCE_NUMBER, + + RD_KAFKA_ERR_ACTION_PERMANENT | + RD_KAFKA_ERR_ACTION_MSG_POSSIBLY_PERSISTED, + RD_KAFKA_RESP_ERR_DUPLICATE_SEQUENCE_NUMBER, + + RD_KAFKA_ERR_ACTION_PERMANENT | + RD_KAFKA_ERR_ACTION_MSG_NOT_PERSISTED, + RD_KAFKA_RESP_ERR_UNKNOWN_PRODUCER_ID, + + RD_KAFKA_ERR_ACTION_PERMANENT | + RD_KAFKA_ERR_ACTION_MSG_NOT_PERSISTED, + RD_KAFKA_RESP_ERR_INVALID_PRODUCER_EPOCH, + + /* Message was purged from out-queue due to + * Idempotent Producer Id change */ + RD_KAFKA_ERR_ACTION_RETRY, RD_KAFKA_RESP_ERR__RETRY, + + RD_KAFKA_ERR_ACTION_END); + + rd_rkb_dbg(rkb, MSG, "MSGSET", + "%s [%" PRId32 + "]: MessageSet with %i message(s) " + "(MsgId %" PRIu64 ", BaseSeq %" PRId32 + ") " + "encountered error: %s (actions %s)%s", + rktp->rktp_rkt->rkt_topic->str, rktp->rktp_partition, + rd_kafka_msgq_len(&batch->msgq), batch->first_msgid, + batch->first_seq, rd_kafka_err2str(perr->err), + rd_kafka_actions2str(perr->actions), + is_leader ? "" : " [NOT LEADER]"); + + + /* + * Special handling for Idempotent Producer + * + * Note: Idempotent Producer-specific errors received + * on a non-idempotent producer will be passed through + * directly to the application. + */ + if (rd_kafka_is_idempotent(rk)) + rd_kafka_handle_idempotent_Produce_error(rkb, batch, perr); + + /* Update message persistence status based on action flags. + * None of these are typically set after an idempotent error, + * which sets the status explicitly. */ + if (perr->actions & RD_KAFKA_ERR_ACTION_MSG_POSSIBLY_PERSISTED) + perr->status = RD_KAFKA_MSG_STATUS_POSSIBLY_PERSISTED; + else if (perr->actions & RD_KAFKA_ERR_ACTION_MSG_NOT_PERSISTED) + perr->status = RD_KAFKA_MSG_STATUS_NOT_PERSISTED; + else if (perr->actions & RD_KAFKA_ERR_ACTION_MSG_PERSISTED) + perr->status = RD_KAFKA_MSG_STATUS_PERSISTED; + + /* Save the last error for debugging sub-sequent errors, + * useful for Idempotent Producer throubleshooting. */ + rd_kafka_toppar_lock(rktp); + rktp->rktp_last_err.err = perr->err; + rktp->rktp_last_err.actions = perr->actions; + rktp->rktp_last_err.ts = rd_clock(); + rktp->rktp_last_err.base_seq = batch->first_seq; + rktp->rktp_last_err.last_seq = perr->last_seq; + rktp->rktp_last_err.base_msgid = batch->first_msgid; + rd_kafka_toppar_unlock(rktp); + + /* + * Handle actions + */ + if (perr->actions & + (RD_KAFKA_ERR_ACTION_REFRESH | RD_KAFKA_ERR_ACTION_RETRY)) { + /* Retry (refresh also implies retry) */ + + if (perr->actions & RD_KAFKA_ERR_ACTION_REFRESH) { + /* Request metadata information update. + * These errors imply that we have stale + * information and the request was + * either rejected or not sent - + * we don't need to increment the retry count + * when we perform a retry since: + * - it is a temporary error (hopefully) + * - there is no chance of duplicate delivery + */ + rd_kafka_toppar_leader_unavailable(rktp, "produce", + perr->err); + + /* We can't be certain the request wasn't + * sent in case of transport failure, + * so the ERR__TRANSPORT case will need + * the retry count to be increased, + * In case of certain other errors we want to + * avoid retrying for the duration of the + * message.timeout.ms to speed up error propagation. */ + if (perr->err != RD_KAFKA_RESP_ERR__TRANSPORT && + perr->err != RD_KAFKA_RESP_ERR_KAFKA_STORAGE_ERROR) + perr->incr_retry = 0; + } + + /* If message timed out in queue, not in transit, + * we will retry at a later time but not increment + * the retry count since there is no risk + * of duplicates. */ + if (!rd_kafka_buf_was_sent(request)) + perr->incr_retry = 0; + + if (!perr->incr_retry) { + /* If retries are not to be incremented then + * there is no chance of duplicates on retry, which + * means these messages were not persisted. */ + perr->status = RD_KAFKA_MSG_STATUS_NOT_PERSISTED; + } + + if (rd_kafka_is_idempotent(rk)) { + /* Any currently in-flight requests will + * fail with ERR_OUT_OF_ORDER_SEQUENCE_NUMBER, + * which should not be treated as a fatal error + * since this request and sub-sequent requests + * will be retried and thus return to order. + * Unless the error was a timeout, or similar, + * in which case the request might have made it + * and the messages are considered possibly persisted: + * in this case we allow the next in-flight response + * to be successful, in which case we mark + * this request's messages as succesfully delivered. */ + if (perr->status & + RD_KAFKA_MSG_STATUS_POSSIBLY_PERSISTED) + perr->update_next_ack = rd_true; + else + perr->update_next_ack = rd_false; + perr->update_next_err = rd_true; + + /* Drain outstanding requests so that retries + * are attempted with proper state knowledge and + * without any in-flight requests. */ + rd_kafka_toppar_lock(rktp); + rd_kafka_idemp_drain_toppar(rktp, + "drain before retrying"); + rd_kafka_toppar_unlock(rktp); + } + + /* Since requests are specific to a broker + * we move the retryable messages from the request + * back to the partition queue (prepend) and then + * let the new broker construct a new request. + * While doing this we also make sure the retry count + * for each message is honoured, any messages that + * would exceeded the retry count will not be + * moved but instead fail below. */ + rd_kafka_toppar_retry_msgq(rktp, &batch->msgq, perr->incr_retry, + perr->status); + + if (rd_kafka_msgq_len(&batch->msgq) == 0) { + /* No need do anything more with the request + * here since the request no longer has any + * messages associated with it. */ + return 0; + } + } + + if (perr->actions & RD_KAFKA_ERR_ACTION_PERMANENT && + rd_kafka_is_idempotent(rk)) { + if (rd_kafka_is_transactional(rk) && + perr->err == RD_KAFKA_RESP_ERR_INVALID_PRODUCER_EPOCH) { + /* Producer was fenced by new transactional producer + * with the same transactional.id */ + rd_kafka_txn_set_fatal_error( + rk, RD_DO_LOCK, RD_KAFKA_RESP_ERR__FENCED, + "ProduceRequest for %.*s [%" PRId32 + "] " + "with %d message(s) failed: %s " + "(broker %" PRId32 " %s, base seq %" PRId32 + "): " + "transactional producer fenced by newer " + "producer instance", + RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic), + rktp->rktp_partition, + rd_kafka_msgq_len(&batch->msgq), + rd_kafka_err2str(perr->err), rkb->rkb_nodeid, + rd_kafka_pid2str(batch->pid), batch->first_seq); + + /* Drain outstanding requests and reset PID. */ + rd_kafka_idemp_drain_reset( + rk, "fenced by new transactional producer"); + + } else if (rd_kafka_is_transactional(rk)) { + /* When transactional any permanent produce failure + * would lead to an incomplete transaction, so raise + * an abortable transaction error. */ + rd_kafka_txn_set_abortable_error( + rk, perr->err, + "ProduceRequest for %.*s [%" PRId32 + "] " + "with %d message(s) failed: %s " + "(broker %" PRId32 " %s, base seq %" PRId32 + "): " + "current transaction must be aborted", + RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic), + rktp->rktp_partition, + rd_kafka_msgq_len(&batch->msgq), + rd_kafka_err2str(perr->err), rkb->rkb_nodeid, + rd_kafka_pid2str(batch->pid), batch->first_seq); + + } else if (rk->rk_conf.eos.gapless) { + /* A permanent non-idempotent error will lead to + * gaps in the message series, the next request + * will fail with ...ERR_OUT_OF_ORDER_SEQUENCE_NUMBER. + * To satisfy the gapless guarantee we need to raise + * a fatal error here. */ + rd_kafka_idemp_set_fatal_error( + rk, RD_KAFKA_RESP_ERR__GAPLESS_GUARANTEE, + "ProduceRequest for %.*s [%" PRId32 + "] " + "with %d message(s) failed: " + "%s (broker %" PRId32 " %s, base seq %" PRId32 + "): " + "unable to satisfy gap-less guarantee", + RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic), + rktp->rktp_partition, + rd_kafka_msgq_len(&batch->msgq), + rd_kafka_err2str(perr->err), rkb->rkb_nodeid, + rd_kafka_pid2str(batch->pid), batch->first_seq); + + /* Drain outstanding requests and reset PID. */ + rd_kafka_idemp_drain_reset( + rk, "unable to satisfy gap-less guarantee"); + + } else { + /* If gapless is not set we bump the Epoch and + * renumber the messages to send. */ + + /* Drain outstanding requests and bump the epoch .*/ + rd_kafka_idemp_drain_epoch_bump(rk, perr->err, + "message sequence gap"); + } + + perr->update_next_ack = rd_false; + /* Make sure the next error will not raise a fatal error. */ + perr->update_next_err = rd_true; + } + + if (perr->err == RD_KAFKA_RESP_ERR__TIMED_OUT || + perr->err == RD_KAFKA_RESP_ERR__TIMED_OUT_QUEUE) { + /* Translate request-level timeout error code + * to message-level timeout error code. */ + perr->err = RD_KAFKA_RESP_ERR__MSG_TIMED_OUT; + + } else if (perr->err == RD_KAFKA_RESP_ERR_TOPIC_AUTHORIZATION_FAILED) { + /* If we're no longer authorized to access the topic mark + * it as errored to deny further produce requests. */ + rd_kafka_topic_wrlock(rktp->rktp_rkt); + rd_kafka_topic_set_error(rktp->rktp_rkt, perr->err); + rd_kafka_topic_wrunlock(rktp->rktp_rkt); + } + + return 1; +} + +/** + * @brief Handle ProduceResponse success for idempotent producer + * + * @warning May be called on the old leader thread. Lock rktp appropriately! + * + * @locks none + * @locality broker thread (but not necessarily the leader broker thread) + */ +static void +rd_kafka_handle_idempotent_Produce_success(rd_kafka_broker_t *rkb, + rd_kafka_msgbatch_t *batch, + int32_t next_seq) { + rd_kafka_t *rk = rkb->rkb_rk; + rd_kafka_toppar_t *rktp = batch->rktp; + char fatal_err[512]; + uint64_t first_msgid, last_msgid; + + *fatal_err = '\0'; + + first_msgid = rd_kafka_msgq_first(&batch->msgq)->rkm_u.producer.msgid; + last_msgid = rd_kafka_msgq_last(&batch->msgq)->rkm_u.producer.msgid; + + rd_kafka_toppar_lock(rktp); + + /* If the last acked msgid is higher than + * the next message to (re)transmit in the message queue + * it means a previous series of R1,R2 ProduceRequests + * had R1 fail with uncertain persistence status, + * such as timeout or transport error, but R2 succeeded, + * which means the messages in R1 were in fact persisted. + * In this case trigger delivery reports for all messages + * in queue until we hit a non-acked message msgid. */ + if (unlikely(rktp->rktp_eos.acked_msgid < first_msgid - 1)) { + rd_kafka_dr_implicit_ack(rkb, rktp, last_msgid); + + } else if (unlikely(batch->first_seq != rktp->rktp_eos.next_ack_seq && + batch->first_seq == rktp->rktp_eos.next_err_seq)) { + /* Response ordering is typically not a concern + * (but will not happen with current broker versions), + * unless we're expecting an error to be returned at + * this sequence rather than a success ack, in which + * case raise a fatal error. */ + + /* Can't call set_fatal_error() while + * holding the toppar lock, so construct + * the error string here and call + * set_fatal_error() below after + * toppar lock has been released. */ + rd_snprintf(fatal_err, sizeof(fatal_err), + "ProduceRequest for %.*s [%" PRId32 + "] " + "with %d message(s) " + "succeeded when expecting failure " + "(broker %" PRId32 + " %s, " + "base seq %" PRId32 + ", " + "next ack seq %" PRId32 + ", " + "next err seq %" PRId32 + ": " + "unable to retry without risking " + "duplication/reordering", + RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic), + rktp->rktp_partition, + rd_kafka_msgq_len(&batch->msgq), rkb->rkb_nodeid, + rd_kafka_pid2str(batch->pid), batch->first_seq, + rktp->rktp_eos.next_ack_seq, + rktp->rktp_eos.next_err_seq); + + rktp->rktp_eos.next_err_seq = next_seq; + } + + if (likely(!*fatal_err)) { + /* Advance next expected err and/or ack sequence */ + + /* Only step err seq if it hasn't diverged. */ + if (rktp->rktp_eos.next_err_seq == rktp->rktp_eos.next_ack_seq) + rktp->rktp_eos.next_err_seq = next_seq; + + rktp->rktp_eos.next_ack_seq = next_seq; + } + + /* Store the last acked message sequence, + * since retries within the broker cache window (5 requests) + * will succeed for older messages we must only update the + * acked msgid if it is higher than the last acked. */ + if (last_msgid > rktp->rktp_eos.acked_msgid) + rktp->rktp_eos.acked_msgid = last_msgid; + + rd_kafka_toppar_unlock(rktp); + + /* Must call set_fatal_error() after releasing + * the toppar lock. */ + if (unlikely(*fatal_err)) + rd_kafka_idemp_set_fatal_error( + rk, RD_KAFKA_RESP_ERR__INCONSISTENT, "%s", fatal_err); +} + + +/** + * @brief Handle ProduceRequest result for a message batch. + * + * @warning \p request may be NULL. + * + * @localiy broker thread (but not necessarily the toppar's handler thread) + * @locks none + */ +static void rd_kafka_msgbatch_handle_Produce_result( + rd_kafka_broker_t *rkb, + rd_kafka_msgbatch_t *batch, + rd_kafka_resp_err_t err, + const struct rd_kafka_Produce_result *presult, + const rd_kafka_buf_t *request) { + + rd_kafka_t *rk = rkb->rkb_rk; + rd_kafka_toppar_t *rktp = batch->rktp; + rd_kafka_msg_status_t status = RD_KAFKA_MSG_STATUS_POSSIBLY_PERSISTED; + rd_bool_t last_inflight; + int32_t next_seq; + + /* Decrease partition's messages in-flight counter */ + rd_assert(rd_atomic32_get(&rktp->rktp_msgs_inflight) >= + rd_kafka_msgq_len(&batch->msgq)); + last_inflight = !rd_atomic32_sub(&rktp->rktp_msgs_inflight, + rd_kafka_msgq_len(&batch->msgq)); + + /* Next expected sequence (and handle wrap) */ + next_seq = rd_kafka_seq_wrap(batch->first_seq + + rd_kafka_msgq_len(&batch->msgq)); + + if (likely(!err)) { + rd_rkb_dbg(rkb, MSG, "MSGSET", + "%s [%" PRId32 + "]: MessageSet with %i message(s) " + "(MsgId %" PRIu64 ", BaseSeq %" PRId32 ") delivered", + rktp->rktp_rkt->rkt_topic->str, rktp->rktp_partition, + rd_kafka_msgq_len(&batch->msgq), batch->first_msgid, + batch->first_seq); + + if (rktp->rktp_rkt->rkt_conf.required_acks != 0) + status = RD_KAFKA_MSG_STATUS_PERSISTED; + + if (rd_kafka_is_idempotent(rk)) + rd_kafka_handle_idempotent_Produce_success(rkb, batch, + next_seq); + } else { + /* Error handling */ + struct rd_kafka_Produce_err perr = { + .err = err, + .incr_retry = 1, + .status = status, + .update_next_ack = rd_true, + .update_next_err = rd_true, + .last_seq = (batch->first_seq + + rd_kafka_msgq_len(&batch->msgq) - 1)}; + + rd_kafka_handle_Produce_error(rkb, request, batch, &perr); + + /* Update next expected acked and/or err sequence. */ + if (perr.update_next_ack || perr.update_next_err) { + rd_kafka_toppar_lock(rktp); + if (perr.update_next_ack) + rktp->rktp_eos.next_ack_seq = next_seq; + if (perr.update_next_err) + rktp->rktp_eos.next_err_seq = next_seq; + rd_kafka_toppar_unlock(rktp); + } + + err = perr.err; + status = perr.status; + } + + + /* Messages to retry will have been removed from the request's queue */ + if (likely(rd_kafka_msgq_len(&batch->msgq) > 0)) { + /* Set offset, timestamp and status for each message. */ + rd_kafka_msgq_set_metadata(&batch->msgq, rkb->rkb_nodeid, + presult->offset, presult->timestamp, + status); + + /* Enqueue messages for delivery report. */ + rd_kafka_dr_msgq(rktp->rktp_rkt, &batch->msgq, err); + } + + if (rd_kafka_is_idempotent(rk) && last_inflight) + rd_kafka_idemp_inflight_toppar_sub(rk, rktp); +} + + +/** + * @brief Handle ProduceResponse + * + * @param reply is NULL when `acks=0` and on various local errors. + * + * @remark ProduceRequests are never retried, retriable errors are + * instead handled by re-enqueuing the request's messages back + * on the partition queue to have a new ProduceRequest constructed + * eventually. + * + * @warning May be called on the old leader thread. Lock rktp appropriately! + * + * @locality broker thread (but not necessarily the leader broker thread) + */ +static void rd_kafka_handle_Produce(rd_kafka_t *rk, + rd_kafka_broker_t *rkb, + rd_kafka_resp_err_t err, + rd_kafka_buf_t *reply, + rd_kafka_buf_t *request, + void *opaque) { + rd_kafka_msgbatch_t *batch = &request->rkbuf_batch; + rd_kafka_toppar_t *rktp = batch->rktp; + struct rd_kafka_Produce_result result = { + .offset = RD_KAFKA_OFFSET_INVALID, .timestamp = -1}; + + /* Unit test interface: inject errors */ + if (unlikely(rk->rk_conf.ut.handle_ProduceResponse != NULL)) { + err = rk->rk_conf.ut.handle_ProduceResponse( + rkb->rkb_rk, rkb->rkb_nodeid, batch->first_msgid, err); + } + + /* Parse Produce reply (unless the request errored) */ + if (!err && reply) + err = rd_kafka_handle_Produce_parse(rkb, rktp, reply, request, + &result); + + rd_kafka_msgbatch_handle_Produce_result(rkb, batch, err, &result, + request); +} + + +/** + * @brief Send ProduceRequest for messages in toppar queue. + * + * @returns the number of messages included, or 0 on error / no messages. + * + * @locality broker thread + */ +int rd_kafka_ProduceRequest(rd_kafka_broker_t *rkb, + rd_kafka_toppar_t *rktp, + const rd_kafka_pid_t pid, + uint64_t epoch_base_msgid) { + rd_kafka_buf_t *rkbuf; + rd_kafka_topic_t *rkt = rktp->rktp_rkt; + size_t MessageSetSize = 0; + int cnt; + rd_ts_t now; + int64_t first_msg_timeout; + int tmout; + + /** + * Create ProduceRequest with as many messages from the toppar + * transmit queue as possible. + */ + rkbuf = rd_kafka_msgset_create_ProduceRequest( + rkb, rktp, &rktp->rktp_xmit_msgq, pid, epoch_base_msgid, + &MessageSetSize); + if (unlikely(!rkbuf)) + return 0; + + cnt = rd_kafka_msgq_len(&rkbuf->rkbuf_batch.msgq); + rd_dassert(cnt > 0); + + rd_avg_add(&rktp->rktp_rkt->rkt_avg_batchcnt, (int64_t)cnt); + rd_avg_add(&rktp->rktp_rkt->rkt_avg_batchsize, (int64_t)MessageSetSize); + + if (!rkt->rkt_conf.required_acks) + rkbuf->rkbuf_flags |= RD_KAFKA_OP_F_NO_RESPONSE; + + /* Use timeout from first message in batch */ + now = rd_clock(); + first_msg_timeout = + (rd_kafka_msgq_first(&rkbuf->rkbuf_batch.msgq)->rkm_ts_timeout - + now) / + 1000; + + if (unlikely(first_msg_timeout <= 0)) { + /* Message has already timed out, allow 100 ms + * to produce anyway */ + tmout = 100; + } else { + tmout = (int)RD_MIN(INT_MAX, first_msg_timeout); + } + + /* Set absolute timeout (including retries), the + * effective timeout for this specific request will be + * capped by socket.timeout.ms */ + rd_kafka_buf_set_abs_timeout(rkbuf, tmout, now); + + rd_kafka_broker_buf_enq_replyq(rkb, rkbuf, RD_KAFKA_NO_REPLYQ, + rd_kafka_handle_Produce, NULL); + + return cnt; +} + + +/** + * @brief Construct and send CreateTopicsRequest to \p rkb + * with the topics (NewTopic_t*) in \p new_topics, using + * \p options. + * + * The response (unparsed) will be enqueued on \p replyq + * for handling by \p resp_cb (with \p opaque passed). + * + * @returns RD_KAFKA_RESP_ERR_NO_ERROR if the request was enqueued for + * transmission, otherwise an error code and errstr will be + * updated with a human readable error string. + */ +rd_kafka_resp_err_t +rd_kafka_CreateTopicsRequest(rd_kafka_broker_t *rkb, + const rd_list_t *new_topics /*(NewTopic_t*)*/, + rd_kafka_AdminOptions_t *options, + char *errstr, + size_t errstr_size, + rd_kafka_replyq_t replyq, + rd_kafka_resp_cb_t *resp_cb, + void *opaque) { + rd_kafka_buf_t *rkbuf; + int16_t ApiVersion = 0; + int features; + int i = 0; + rd_kafka_NewTopic_t *newt; + int op_timeout; + + if (rd_list_cnt(new_topics) == 0) { + rd_snprintf(errstr, errstr_size, "No topics to create"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__INVALID_ARG; + } + + ApiVersion = rd_kafka_broker_ApiVersion_supported( + rkb, RD_KAFKAP_CreateTopics, 0, 4, &features); + if (ApiVersion == -1) { + rd_snprintf(errstr, errstr_size, + "Topic Admin API (KIP-4) not supported " + "by broker, requires broker version >= 0.10.2.0"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE; + } + + if (rd_kafka_confval_get_int(&options->validate_only) && + ApiVersion < 1) { + rd_snprintf(errstr, errstr_size, + "CreateTopics.validate_only=true not " + "supported by broker"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE; + } + + rkbuf = rd_kafka_buf_new_request(rkb, RD_KAFKAP_CreateTopics, 1, + 4 + (rd_list_cnt(new_topics) * 200) + + 4 + 1); + + /* #topics */ + rd_kafka_buf_write_i32(rkbuf, rd_list_cnt(new_topics)); + + while ((newt = rd_list_elem(new_topics, i++))) { + int partition; + int ei = 0; + const rd_kafka_ConfigEntry_t *entry; + + if (ApiVersion < 4) { + if (newt->num_partitions == -1) { + rd_snprintf(errstr, errstr_size, + "Default partition count (KIP-464) " + "not supported by broker, " + "requires broker version <= 2.4.0"); + rd_kafka_replyq_destroy(&replyq); + rd_kafka_buf_destroy(rkbuf); + return RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE; + } + + if (newt->replication_factor == -1 && + rd_list_empty(&newt->replicas)) { + rd_snprintf(errstr, errstr_size, + "Default replication factor " + "(KIP-464) " + "not supported by broker, " + "requires broker version <= 2.4.0"); + rd_kafka_replyq_destroy(&replyq); + rd_kafka_buf_destroy(rkbuf); + return RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE; + } + } + + /* topic */ + rd_kafka_buf_write_str(rkbuf, newt->topic, -1); + + if (rd_list_cnt(&newt->replicas)) { + /* num_partitions and replication_factor must be + * set to -1 if a replica assignment is sent. */ + /* num_partitions */ + rd_kafka_buf_write_i32(rkbuf, -1); + /* replication_factor */ + rd_kafka_buf_write_i16(rkbuf, -1); + } else { + /* num_partitions */ + rd_kafka_buf_write_i32(rkbuf, newt->num_partitions); + /* replication_factor */ + rd_kafka_buf_write_i16( + rkbuf, (int16_t)newt->replication_factor); + } + + /* #replica_assignment */ + rd_kafka_buf_write_i32(rkbuf, rd_list_cnt(&newt->replicas)); + + /* Replicas per partition, see rdkafka_admin.[ch] + * for how these are constructed. */ + for (partition = 0; partition < rd_list_cnt(&newt->replicas); + partition++) { + const rd_list_t *replicas; + int ri = 0; + + replicas = rd_list_elem(&newt->replicas, partition); + if (!replicas) + continue; + + /* partition */ + rd_kafka_buf_write_i32(rkbuf, partition); + /* #replicas */ + rd_kafka_buf_write_i32(rkbuf, rd_list_cnt(replicas)); + + for (ri = 0; ri < rd_list_cnt(replicas); ri++) { + /* replica */ + rd_kafka_buf_write_i32( + rkbuf, rd_list_get_int32(replicas, ri)); + } + } + + /* #config_entries */ + rd_kafka_buf_write_i32(rkbuf, rd_list_cnt(&newt->config)); + + RD_LIST_FOREACH(entry, &newt->config, ei) { + /* config_name */ + rd_kafka_buf_write_str(rkbuf, entry->kv->name, -1); + /* config_value (nullable) */ + rd_kafka_buf_write_str(rkbuf, entry->kv->value, -1); + } + } + + /* timeout */ + op_timeout = rd_kafka_confval_get_int(&options->operation_timeout); + rd_kafka_buf_write_i32(rkbuf, op_timeout); + + if (op_timeout > rkb->rkb_rk->rk_conf.socket_timeout_ms) + rd_kafka_buf_set_abs_timeout(rkbuf, op_timeout + 1000, 0); + + if (ApiVersion >= 1) { + /* validate_only */ + rd_kafka_buf_write_i8( + rkbuf, rd_kafka_confval_get_int(&options->validate_only)); + } + + rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0); + + rd_kafka_broker_buf_enq_replyq(rkb, rkbuf, replyq, resp_cb, opaque); + + return RD_KAFKA_RESP_ERR_NO_ERROR; +} + + +/** + * @brief Construct and send DeleteTopicsRequest to \p rkb + * with the topics (DeleteTopic_t *) in \p del_topics, using + * \p options. + * + * The response (unparsed) will be enqueued on \p replyq + * for handling by \p resp_cb (with \p opaque passed). + * + * @returns RD_KAFKA_RESP_ERR_NO_ERROR if the request was enqueued for + * transmission, otherwise an error code and errstr will be + * updated with a human readable error string. + */ +rd_kafka_resp_err_t +rd_kafka_DeleteTopicsRequest(rd_kafka_broker_t *rkb, + const rd_list_t *del_topics /*(DeleteTopic_t*)*/, + rd_kafka_AdminOptions_t *options, + char *errstr, + size_t errstr_size, + rd_kafka_replyq_t replyq, + rd_kafka_resp_cb_t *resp_cb, + void *opaque) { + rd_kafka_buf_t *rkbuf; + int16_t ApiVersion = 0; + int features; + int i = 0; + rd_kafka_DeleteTopic_t *delt; + int op_timeout; + + if (rd_list_cnt(del_topics) == 0) { + rd_snprintf(errstr, errstr_size, "No topics to delete"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__INVALID_ARG; + } + + ApiVersion = rd_kafka_broker_ApiVersion_supported( + rkb, RD_KAFKAP_DeleteTopics, 0, 1, &features); + if (ApiVersion == -1) { + rd_snprintf(errstr, errstr_size, + "Topic Admin API (KIP-4) not supported " + "by broker, requires broker version >= 0.10.2.0"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE; + } + + rkbuf = + rd_kafka_buf_new_request(rkb, RD_KAFKAP_DeleteTopics, 1, + /* FIXME */ + 4 + (rd_list_cnt(del_topics) * 100) + 4); + + /* #topics */ + rd_kafka_buf_write_i32(rkbuf, rd_list_cnt(del_topics)); + + while ((delt = rd_list_elem(del_topics, i++))) + rd_kafka_buf_write_str(rkbuf, delt->topic, -1); + + /* timeout */ + op_timeout = rd_kafka_confval_get_int(&options->operation_timeout); + rd_kafka_buf_write_i32(rkbuf, op_timeout); + + if (op_timeout > rkb->rkb_rk->rk_conf.socket_timeout_ms) + rd_kafka_buf_set_abs_timeout(rkbuf, op_timeout + 1000, 0); + + rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0); + + rd_kafka_broker_buf_enq_replyq(rkb, rkbuf, replyq, resp_cb, opaque); + + return RD_KAFKA_RESP_ERR_NO_ERROR; +} + + +/** + * @brief Construct and send DeleteRecordsRequest to \p rkb + * with the offsets to delete (rd_kafka_topic_partition_list_t *) in + * \p offsets_list, using \p options. + * + * The response (unparsed) will be enqueued on \p replyq + * for handling by \p resp_cb (with \p opaque passed). + * + * @remark The rd_kafka_topic_partition_list_t in \p offsets_list must already + * be sorted. + * + * @returns RD_KAFKA_RESP_ERR_NO_ERROR if the request was enqueued for + * transmission, otherwise an error code and errstr will be + * updated with a human readable error string. + */ +rd_kafka_resp_err_t +rd_kafka_DeleteRecordsRequest(rd_kafka_broker_t *rkb, + /*(rd_kafka_topic_partition_list_t*)*/ + const rd_list_t *offsets_list, + rd_kafka_AdminOptions_t *options, + char *errstr, + size_t errstr_size, + rd_kafka_replyq_t replyq, + rd_kafka_resp_cb_t *resp_cb, + void *opaque) { + rd_kafka_buf_t *rkbuf; + int16_t ApiVersion = 0; + int features; + const rd_kafka_topic_partition_list_t *partitions; + int op_timeout; + + partitions = rd_list_elem(offsets_list, 0); + + ApiVersion = rd_kafka_broker_ApiVersion_supported( + rkb, RD_KAFKAP_DeleteRecords, 0, 1, &features); + if (ApiVersion == -1) { + rd_snprintf(errstr, errstr_size, + "DeleteRecords Admin API (KIP-107) not supported " + "by broker, requires broker version >= 0.11.0"); + return RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE; + } + + rkbuf = rd_kafka_buf_new_request(rkb, RD_KAFKAP_DeleteRecords, 1, + 4 + (partitions->cnt * 100) + 4); + + const rd_kafka_topic_partition_field_t fields[] = { + RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION, + RD_KAFKA_TOPIC_PARTITION_FIELD_OFFSET, + RD_KAFKA_TOPIC_PARTITION_FIELD_END}; + rd_kafka_buf_write_topic_partitions( + rkbuf, partitions, rd_false /*don't skip invalid offsets*/, + rd_false /*any offset*/, fields); + + /* timeout */ + op_timeout = rd_kafka_confval_get_int(&options->operation_timeout); + rd_kafka_buf_write_i32(rkbuf, op_timeout); + + if (op_timeout > rkb->rkb_rk->rk_conf.socket_timeout_ms) + rd_kafka_buf_set_abs_timeout(rkbuf, op_timeout + 1000, 0); + + rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0); + + rd_kafka_broker_buf_enq_replyq(rkb, rkbuf, replyq, resp_cb, opaque); + + return RD_KAFKA_RESP_ERR_NO_ERROR; +} + + +/** + * @brief Construct and send CreatePartitionsRequest to \p rkb + * with the topics (NewPartitions_t*) in \p new_parts, using + * \p options. + * + * The response (unparsed) will be enqueued on \p replyq + * for handling by \p resp_cb (with \p opaque passed). + * + * @returns RD_KAFKA_RESP_ERR_NO_ERROR if the request was enqueued for + * transmission, otherwise an error code and errstr will be + * updated with a human readable error string. + */ +rd_kafka_resp_err_t +rd_kafka_CreatePartitionsRequest(rd_kafka_broker_t *rkb, + /*(NewPartitions_t*)*/ + const rd_list_t *new_parts, + rd_kafka_AdminOptions_t *options, + char *errstr, + size_t errstr_size, + rd_kafka_replyq_t replyq, + rd_kafka_resp_cb_t *resp_cb, + void *opaque) { + rd_kafka_buf_t *rkbuf; + int16_t ApiVersion = 0; + int i = 0; + rd_kafka_NewPartitions_t *newp; + int op_timeout; + + if (rd_list_cnt(new_parts) == 0) { + rd_snprintf(errstr, errstr_size, "No partitions to create"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__INVALID_ARG; + } + + ApiVersion = rd_kafka_broker_ApiVersion_supported( + rkb, RD_KAFKAP_CreatePartitions, 0, 0, NULL); + if (ApiVersion == -1) { + rd_snprintf(errstr, errstr_size, + "CreatePartitions (KIP-195) not supported " + "by broker, requires broker version >= 1.0.0"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE; + } + + rkbuf = rd_kafka_buf_new_request(rkb, RD_KAFKAP_CreatePartitions, 1, + 4 + (rd_list_cnt(new_parts) * 200) + + 4 + 1); + + /* #topics */ + rd_kafka_buf_write_i32(rkbuf, rd_list_cnt(new_parts)); + + while ((newp = rd_list_elem(new_parts, i++))) { + /* topic */ + rd_kafka_buf_write_str(rkbuf, newp->topic, -1); + + /* New partition count */ + rd_kafka_buf_write_i32(rkbuf, (int32_t)newp->total_cnt); + + /* #replica_assignment */ + if (rd_list_empty(&newp->replicas)) { + rd_kafka_buf_write_i32(rkbuf, -1); + } else { + const rd_list_t *replicas; + int pi = -1; + + rd_kafka_buf_write_i32(rkbuf, + rd_list_cnt(&newp->replicas)); + + while ( + (replicas = rd_list_elem(&newp->replicas, ++pi))) { + int ri = 0; + + /* replica count */ + rd_kafka_buf_write_i32(rkbuf, + rd_list_cnt(replicas)); + + /* replica */ + for (ri = 0; ri < rd_list_cnt(replicas); ri++) { + rd_kafka_buf_write_i32( + rkbuf, + rd_list_get_int32(replicas, ri)); + } + } + } + } + + /* timeout */ + op_timeout = rd_kafka_confval_get_int(&options->operation_timeout); + rd_kafka_buf_write_i32(rkbuf, op_timeout); + + if (op_timeout > rkb->rkb_rk->rk_conf.socket_timeout_ms) + rd_kafka_buf_set_abs_timeout(rkbuf, op_timeout + 1000, 0); + + /* validate_only */ + rd_kafka_buf_write_i8( + rkbuf, rd_kafka_confval_get_int(&options->validate_only)); + + rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0); + + rd_kafka_broker_buf_enq_replyq(rkb, rkbuf, replyq, resp_cb, opaque); + + return RD_KAFKA_RESP_ERR_NO_ERROR; +} + + +/** + * @brief Construct and send AlterConfigsRequest to \p rkb + * with the configs (ConfigResource_t*) in \p configs, using + * \p options. + * + * The response (unparsed) will be enqueued on \p replyq + * for handling by \p resp_cb (with \p opaque passed). + * + * @returns RD_KAFKA_RESP_ERR_NO_ERROR if the request was enqueued for + * transmission, otherwise an error code and errstr will be + * updated with a human readable error string. + */ +rd_kafka_resp_err_t +rd_kafka_AlterConfigsRequest(rd_kafka_broker_t *rkb, + const rd_list_t *configs /*(ConfigResource_t*)*/, + rd_kafka_AdminOptions_t *options, + char *errstr, + size_t errstr_size, + rd_kafka_replyq_t replyq, + rd_kafka_resp_cb_t *resp_cb, + void *opaque) { + rd_kafka_buf_t *rkbuf; + int16_t ApiVersion = 0; + int i; + const rd_kafka_ConfigResource_t *config; + int op_timeout; + + if (rd_list_cnt(configs) == 0) { + rd_snprintf(errstr, errstr_size, + "No config resources specified"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__INVALID_ARG; + } + + ApiVersion = rd_kafka_broker_ApiVersion_supported( + rkb, RD_KAFKAP_AlterConfigs, 0, 1, NULL); + if (ApiVersion == -1) { + rd_snprintf(errstr, errstr_size, + "AlterConfigs (KIP-133) not supported " + "by broker, requires broker version >= 0.11.0"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE; + } + + /* Incremental requires IncrementalAlterConfigs */ + if (rd_kafka_confval_get_int(&options->incremental)) { + rd_snprintf(errstr, errstr_size, + "AlterConfigs.incremental=true (KIP-248) " + "not supported by broker, " + "replaced by IncrementalAlterConfigs"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE; + } + + rkbuf = rd_kafka_buf_new_request(rkb, RD_KAFKAP_AlterConfigs, 1, + rd_list_cnt(configs) * 200); + + /* #resources */ + rd_kafka_buf_write_i32(rkbuf, rd_list_cnt(configs)); + + RD_LIST_FOREACH(config, configs, i) { + const rd_kafka_ConfigEntry_t *entry; + int ei; + + /* resource_type */ + rd_kafka_buf_write_i8(rkbuf, config->restype); + + /* resource_name */ + rd_kafka_buf_write_str(rkbuf, config->name, -1); + + /* #config */ + rd_kafka_buf_write_i32(rkbuf, rd_list_cnt(&config->config)); + + RD_LIST_FOREACH(entry, &config->config, ei) { + /* config_name */ + rd_kafka_buf_write_str(rkbuf, entry->kv->name, -1); + /* config_value (nullable) */ + rd_kafka_buf_write_str(rkbuf, entry->kv->value, -1); + + if (entry->a.operation != RD_KAFKA_ALTER_OP_SET) { + rd_snprintf(errstr, errstr_size, + "IncrementalAlterConfigs required " + "for add/delete config " + "entries: only set supported " + "by this operation"); + rd_kafka_buf_destroy(rkbuf); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE; + } + } + } + + /* timeout */ + op_timeout = rd_kafka_confval_get_int(&options->operation_timeout); + if (op_timeout > rkb->rkb_rk->rk_conf.socket_timeout_ms) + rd_kafka_buf_set_abs_timeout(rkbuf, op_timeout + 1000, 0); + + /* validate_only */ + rd_kafka_buf_write_i8( + rkbuf, rd_kafka_confval_get_int(&options->validate_only)); + + rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0); + + rd_kafka_broker_buf_enq_replyq(rkb, rkbuf, replyq, resp_cb, opaque); + + return RD_KAFKA_RESP_ERR_NO_ERROR; +} + + +/** + * @brief Construct and send DescribeConfigsRequest to \p rkb + * with the configs (ConfigResource_t*) in \p configs, using + * \p options. + * + * The response (unparsed) will be enqueued on \p replyq + * for handling by \p resp_cb (with \p opaque passed). + * + * @returns RD_KAFKA_RESP_ERR_NO_ERROR if the request was enqueued for + * transmission, otherwise an error code and errstr will be + * updated with a human readable error string. + */ +rd_kafka_resp_err_t rd_kafka_DescribeConfigsRequest( + rd_kafka_broker_t *rkb, + const rd_list_t *configs /*(ConfigResource_t*)*/, + rd_kafka_AdminOptions_t *options, + char *errstr, + size_t errstr_size, + rd_kafka_replyq_t replyq, + rd_kafka_resp_cb_t *resp_cb, + void *opaque) { + rd_kafka_buf_t *rkbuf; + int16_t ApiVersion = 0; + int i; + const rd_kafka_ConfigResource_t *config; + int op_timeout; + + if (rd_list_cnt(configs) == 0) { + rd_snprintf(errstr, errstr_size, + "No config resources specified"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__INVALID_ARG; + } + + ApiVersion = rd_kafka_broker_ApiVersion_supported( + rkb, RD_KAFKAP_DescribeConfigs, 0, 1, NULL); + if (ApiVersion == -1) { + rd_snprintf(errstr, errstr_size, + "DescribeConfigs (KIP-133) not supported " + "by broker, requires broker version >= 0.11.0"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE; + } + + rkbuf = rd_kafka_buf_new_request(rkb, RD_KAFKAP_DescribeConfigs, 1, + rd_list_cnt(configs) * 200); + + /* #resources */ + rd_kafka_buf_write_i32(rkbuf, rd_list_cnt(configs)); + + RD_LIST_FOREACH(config, configs, i) { + const rd_kafka_ConfigEntry_t *entry; + int ei; + + /* resource_type */ + rd_kafka_buf_write_i8(rkbuf, config->restype); + + /* resource_name */ + rd_kafka_buf_write_str(rkbuf, config->name, -1); + + /* #config */ + if (rd_list_empty(&config->config)) { + /* Get all configs */ + rd_kafka_buf_write_i32(rkbuf, -1); + } else { + /* Get requested configs only */ + rd_kafka_buf_write_i32(rkbuf, + rd_list_cnt(&config->config)); + } + + RD_LIST_FOREACH(entry, &config->config, ei) { + /* config_name */ + rd_kafka_buf_write_str(rkbuf, entry->kv->name, -1); + } + } + + + if (ApiVersion == 1) { + /* include_synonyms */ + rd_kafka_buf_write_i8(rkbuf, 1); + } + + /* timeout */ + op_timeout = rd_kafka_confval_get_int(&options->operation_timeout); + if (op_timeout > rkb->rkb_rk->rk_conf.socket_timeout_ms) + rd_kafka_buf_set_abs_timeout(rkbuf, op_timeout + 1000, 0); + + rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0); + + rd_kafka_broker_buf_enq_replyq(rkb, rkbuf, replyq, resp_cb, opaque); + + return RD_KAFKA_RESP_ERR_NO_ERROR; +} + + +/** + * @brief Construct and send DeleteGroupsRequest to \p rkb + * with the groups (DeleteGroup_t *) in \p del_groups, using + * \p options. + * + * The response (unparsed) will be enqueued on \p replyq + * for handling by \p resp_cb (with \p opaque passed). + * + * @returns RD_KAFKA_RESP_ERR_NO_ERROR if the request was enqueued for + * transmission, otherwise an error code and errstr will be + * updated with a human readable error string. + */ +rd_kafka_resp_err_t +rd_kafka_DeleteGroupsRequest(rd_kafka_broker_t *rkb, + const rd_list_t *del_groups /*(DeleteGroup_t*)*/, + rd_kafka_AdminOptions_t *options, + char *errstr, + size_t errstr_size, + rd_kafka_replyq_t replyq, + rd_kafka_resp_cb_t *resp_cb, + void *opaque) { + rd_kafka_buf_t *rkbuf; + int16_t ApiVersion = 0; + int features; + int i = 0; + rd_kafka_DeleteGroup_t *delt; + + ApiVersion = rd_kafka_broker_ApiVersion_supported( + rkb, RD_KAFKAP_DeleteGroups, 0, 1, &features); + if (ApiVersion == -1) { + rd_snprintf(errstr, errstr_size, + "DeleteGroups Admin API (KIP-229) not supported " + "by broker, requires broker version >= 1.1.0"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE; + } + + rkbuf = + rd_kafka_buf_new_request(rkb, RD_KAFKAP_DeleteGroups, 1, + 4 + (rd_list_cnt(del_groups) * 100) + 4); + + /* #groups */ + rd_kafka_buf_write_i32(rkbuf, rd_list_cnt(del_groups)); + + while ((delt = rd_list_elem(del_groups, i++))) + rd_kafka_buf_write_str(rkbuf, delt->group, -1); + + rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0); + + rd_kafka_broker_buf_enq_replyq(rkb, rkbuf, replyq, resp_cb, opaque); + + return RD_KAFKA_RESP_ERR_NO_ERROR; +} + +/** + * @brief Returns the request size needed to send a specific AclBinding + * specified in \p acl, using the ApiVersion provided in + * \p ApiVersion. + * + * @returns and int16_t with the request size in bytes. + */ +static RD_INLINE size_t +rd_kafka_AclBinding_request_size(const rd_kafka_AclBinding_t *acl, + int ApiVersion) { + return 1 + 2 + (acl->name ? strlen(acl->name) : 0) + 2 + + (acl->principal ? strlen(acl->principal) : 0) + 2 + + (acl->host ? strlen(acl->host) : 0) + 1 + 1 + + (ApiVersion > 0 ? 1 : 0); +} + +/** + * @brief Construct and send CreateAclsRequest to \p rkb + * with the acls (AclBinding_t*) in \p new_acls, using + * \p options. + * + * The response (unparsed) will be enqueued on \p replyq + * for handling by \p resp_cb (with \p opaque passed). + * + * @returns RD_KAFKA_RESP_ERR_NO_ERROR if the request was enqueued for + * transmission, otherwise an error code and errstr will be + * updated with a human readable error string. + */ +rd_kafka_resp_err_t +rd_kafka_CreateAclsRequest(rd_kafka_broker_t *rkb, + const rd_list_t *new_acls /*(AclBinding_t*)*/, + rd_kafka_AdminOptions_t *options, + char *errstr, + size_t errstr_size, + rd_kafka_replyq_t replyq, + rd_kafka_resp_cb_t *resp_cb, + void *opaque) { + rd_kafka_buf_t *rkbuf; + int16_t ApiVersion; + int i; + size_t len; + int op_timeout; + rd_kafka_AclBinding_t *new_acl; + + if (rd_list_cnt(new_acls) == 0) { + rd_snprintf(errstr, errstr_size, "No acls to create"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__INVALID_ARG; + } + + ApiVersion = rd_kafka_broker_ApiVersion_supported( + rkb, RD_KAFKAP_CreateAcls, 0, 1, NULL); + if (ApiVersion == -1) { + rd_snprintf(errstr, errstr_size, + "ACLs Admin API (KIP-140) not supported " + "by broker, requires broker version >= 0.11.0.0"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE; + } + + if (ApiVersion == 0) { + RD_LIST_FOREACH(new_acl, new_acls, i) { + if (new_acl->resource_pattern_type != + RD_KAFKA_RESOURCE_PATTERN_LITERAL) { + rd_snprintf(errstr, errstr_size, + "Broker only supports LITERAL " + "resource pattern types"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE; + } + } + } else { + RD_LIST_FOREACH(new_acl, new_acls, i) { + if (new_acl->resource_pattern_type != + RD_KAFKA_RESOURCE_PATTERN_LITERAL && + new_acl->resource_pattern_type != + RD_KAFKA_RESOURCE_PATTERN_PREFIXED) { + rd_snprintf(errstr, errstr_size, + "Only LITERAL and PREFIXED " + "resource patterns are supported " + "when creating ACLs"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE; + } + } + } + + len = 4; + RD_LIST_FOREACH(new_acl, new_acls, i) { + len += rd_kafka_AclBinding_request_size(new_acl, ApiVersion); + } + + rkbuf = rd_kafka_buf_new_request(rkb, RD_KAFKAP_CreateAcls, 1, len); + + /* #acls */ + rd_kafka_buf_write_i32(rkbuf, rd_list_cnt(new_acls)); + + RD_LIST_FOREACH(new_acl, new_acls, i) { + rd_kafka_buf_write_i8(rkbuf, new_acl->restype); + + rd_kafka_buf_write_str(rkbuf, new_acl->name, -1); + + if (ApiVersion >= 1) { + rd_kafka_buf_write_i8(rkbuf, + new_acl->resource_pattern_type); + } + + rd_kafka_buf_write_str(rkbuf, new_acl->principal, -1); + + rd_kafka_buf_write_str(rkbuf, new_acl->host, -1); + + rd_kafka_buf_write_i8(rkbuf, new_acl->operation); + + rd_kafka_buf_write_i8(rkbuf, new_acl->permission_type); + } + + /* timeout */ + op_timeout = rd_kafka_confval_get_int(&options->operation_timeout); + if (op_timeout > rkb->rkb_rk->rk_conf.socket_timeout_ms) + rd_kafka_buf_set_abs_timeout(rkbuf, op_timeout + 1000, 0); + + rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0); + + rd_kafka_broker_buf_enq_replyq(rkb, rkbuf, replyq, resp_cb, opaque); + + return RD_KAFKA_RESP_ERR_NO_ERROR; +} + +/** + * @brief Construct and send DescribeAclsRequest to \p rkb + * with the acls (AclBinding_t*) in \p acls, using + * \p options. + * + * The response (unparsed) will be enqueued on \p replyq + * for handling by \p resp_cb (with \p opaque passed). + * + * @returns RD_KAFKA_RESP_ERR_NO_ERROR if the request was enqueued for + * transmission, otherwise an error code and errstr will be + * updated with a human readable error string. + */ +rd_kafka_resp_err_t rd_kafka_DescribeAclsRequest( + rd_kafka_broker_t *rkb, + const rd_list_t *acls /*(rd_kafka_AclBindingFilter_t*)*/, + rd_kafka_AdminOptions_t *options, + char *errstr, + size_t errstr_size, + rd_kafka_replyq_t replyq, + rd_kafka_resp_cb_t *resp_cb, + void *opaque) { + rd_kafka_buf_t *rkbuf; + int16_t ApiVersion = 0; + const rd_kafka_AclBindingFilter_t *acl; + int op_timeout; + + if (rd_list_cnt(acls) == 0) { + rd_snprintf(errstr, errstr_size, + "No acl binding filters specified"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__INVALID_ARG; + } + if (rd_list_cnt(acls) > 1) { + rd_snprintf(errstr, errstr_size, + "Too many acl binding filters specified"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__INVALID_ARG; + } + + acl = rd_list_elem(acls, 0); + + ApiVersion = rd_kafka_broker_ApiVersion_supported( + rkb, RD_KAFKAP_DescribeAcls, 0, 1, NULL); + if (ApiVersion == -1) { + rd_snprintf(errstr, errstr_size, + "ACLs Admin API (KIP-140) not supported " + "by broker, requires broker version >= 0.11.0.0"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE; + } + + if (ApiVersion == 0) { + if (acl->resource_pattern_type != + RD_KAFKA_RESOURCE_PATTERN_LITERAL && + acl->resource_pattern_type != + RD_KAFKA_RESOURCE_PATTERN_ANY) { + rd_snprintf(errstr, errstr_size, + "Broker only supports LITERAL and ANY " + "resource pattern types"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE; + } + } else { + if (acl->resource_pattern_type == + RD_KAFKA_RESOURCE_PATTERN_UNKNOWN) { + rd_snprintf(errstr, errstr_size, + "Filter contains UNKNOWN elements"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE; + } + } + + rkbuf = rd_kafka_buf_new_request( + rkb, RD_KAFKAP_DescribeAcls, 1, + rd_kafka_AclBinding_request_size(acl, ApiVersion)); + + /* resource_type */ + rd_kafka_buf_write_i8(rkbuf, acl->restype); + + /* resource_name filter */ + rd_kafka_buf_write_str(rkbuf, acl->name, -1); + + if (ApiVersion > 0) { + /* resource_pattern_type (rd_kafka_ResourcePatternType_t) */ + rd_kafka_buf_write_i8(rkbuf, acl->resource_pattern_type); + } + + /* principal filter */ + rd_kafka_buf_write_str(rkbuf, acl->principal, -1); + + /* host filter */ + rd_kafka_buf_write_str(rkbuf, acl->host, -1); + + /* operation (rd_kafka_AclOperation_t) */ + rd_kafka_buf_write_i8(rkbuf, acl->operation); + + /* permission type (rd_kafka_AclPermissionType_t) */ + rd_kafka_buf_write_i8(rkbuf, acl->permission_type); + + /* timeout */ + op_timeout = rd_kafka_confval_get_int(&options->operation_timeout); + if (op_timeout > rkb->rkb_rk->rk_conf.socket_timeout_ms) + rd_kafka_buf_set_abs_timeout(rkbuf, op_timeout + 1000, 0); + + rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0); + + rd_kafka_broker_buf_enq_replyq(rkb, rkbuf, replyq, resp_cb, opaque); + + return RD_KAFKA_RESP_ERR_NO_ERROR; +} + +/** + * @brief Construct and send DeleteAclsRequest to \p rkb + * with the acl filters (AclBindingFilter_t*) in \p del_acls, using + * \p options. + * + * The response (unparsed) will be enqueued on \p replyq + * for handling by \p resp_cb (with \p opaque passed). + * + * @returns RD_KAFKA_RESP_ERR_NO_ERROR if the request was enqueued for + * transmission, otherwise an error code and errstr will be + * updated with a human readable error string. + */ +rd_kafka_resp_err_t +rd_kafka_DeleteAclsRequest(rd_kafka_broker_t *rkb, + const rd_list_t *del_acls /*(AclBindingFilter_t*)*/, + rd_kafka_AdminOptions_t *options, + char *errstr, + size_t errstr_size, + rd_kafka_replyq_t replyq, + rd_kafka_resp_cb_t *resp_cb, + void *opaque) { + rd_kafka_buf_t *rkbuf; + int16_t ApiVersion = 0; + const rd_kafka_AclBindingFilter_t *acl; + int op_timeout; + int i; + size_t len; + + if (rd_list_cnt(del_acls) == 0) { + rd_snprintf(errstr, errstr_size, + "No acl binding filters specified"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__INVALID_ARG; + } + + ApiVersion = rd_kafka_broker_ApiVersion_supported( + rkb, RD_KAFKAP_DeleteAcls, 0, 1, NULL); + if (ApiVersion == -1) { + rd_snprintf(errstr, errstr_size, + "ACLs Admin API (KIP-140) not supported " + "by broker, requires broker version >= 0.11.0.0"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE; + } + + len = 4; + + RD_LIST_FOREACH(acl, del_acls, i) { + if (ApiVersion == 0) { + if (acl->resource_pattern_type != + RD_KAFKA_RESOURCE_PATTERN_LITERAL && + acl->resource_pattern_type != + RD_KAFKA_RESOURCE_PATTERN_ANY) { + rd_snprintf(errstr, errstr_size, + "Broker only supports LITERAL " + "and ANY resource pattern types"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE; + } + } else { + if (acl->resource_pattern_type == + RD_KAFKA_RESOURCE_PATTERN_UNKNOWN) { + rd_snprintf(errstr, errstr_size, + "Filter contains UNKNOWN elements"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE; + } + } + + len += rd_kafka_AclBinding_request_size(acl, ApiVersion); + } + + rkbuf = rd_kafka_buf_new_request(rkb, RD_KAFKAP_DeleteAcls, 1, len); + + /* #acls */ + rd_kafka_buf_write_i32(rkbuf, rd_list_cnt(del_acls)); + + RD_LIST_FOREACH(acl, del_acls, i) { + /* resource_type */ + rd_kafka_buf_write_i8(rkbuf, acl->restype); + + /* resource_name filter */ + rd_kafka_buf_write_str(rkbuf, acl->name, -1); + + if (ApiVersion > 0) { + /* resource_pattern_type + * (rd_kafka_ResourcePatternType_t) */ + rd_kafka_buf_write_i8(rkbuf, + acl->resource_pattern_type); + } + + /* principal filter */ + rd_kafka_buf_write_str(rkbuf, acl->principal, -1); + + /* host filter */ + rd_kafka_buf_write_str(rkbuf, acl->host, -1); + + /* operation (rd_kafka_AclOperation_t) */ + rd_kafka_buf_write_i8(rkbuf, acl->operation); + + /* permission type (rd_kafka_AclPermissionType_t) */ + rd_kafka_buf_write_i8(rkbuf, acl->permission_type); + } + + /* timeout */ + op_timeout = rd_kafka_confval_get_int(&options->operation_timeout); + if (op_timeout > rkb->rkb_rk->rk_conf.socket_timeout_ms) + rd_kafka_buf_set_abs_timeout(rkbuf, op_timeout + 1000, 0); + + rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0); + + rd_kafka_broker_buf_enq_replyq(rkb, rkbuf, replyq, resp_cb, opaque); + + return RD_KAFKA_RESP_ERR_NO_ERROR; +} + +/** + * @brief Parses and handles an InitProducerId reply. + * + * @locality rdkafka main thread + * @locks none + */ +void rd_kafka_handle_InitProducerId(rd_kafka_t *rk, + rd_kafka_broker_t *rkb, + rd_kafka_resp_err_t err, + rd_kafka_buf_t *rkbuf, + rd_kafka_buf_t *request, + void *opaque) { + const int log_decode_errors = LOG_ERR; + int16_t error_code; + rd_kafka_pid_t pid; + + if (err) + goto err; + + rd_kafka_buf_read_throttle_time(rkbuf); + + rd_kafka_buf_read_i16(rkbuf, &error_code); + if ((err = error_code)) + goto err; + + rd_kafka_buf_read_i64(rkbuf, &pid.id); + rd_kafka_buf_read_i16(rkbuf, &pid.epoch); + + rd_kafka_idemp_pid_update(rkb, pid); + + return; + +err_parse: + err = rkbuf->rkbuf_err; +err: + if (err == RD_KAFKA_RESP_ERR__DESTROY) + return; + + /* Retries are performed by idempotence state handler */ + rd_kafka_idemp_request_pid_failed(rkb, err); +} + +/** + * @brief Construct and send InitProducerIdRequest to \p rkb. + * + * @param transactional_id may be NULL. + * @param transaction_timeout_ms may be set to -1. + * @param current_pid the current PID to reset, requires KIP-360. If not NULL + * and KIP-360 is not supported by the broker this function + * will return RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE. + * + * The response (unparsed) will be handled by \p resp_cb served + * by queue \p replyq. + * + * @returns RD_KAFKA_RESP_ERR_NO_ERROR if the request was enqueued for + * transmission, otherwise an error code and errstr will be + * updated with a human readable error string. + */ +rd_kafka_resp_err_t +rd_kafka_InitProducerIdRequest(rd_kafka_broker_t *rkb, + const char *transactional_id, + int transaction_timeout_ms, + const rd_kafka_pid_t *current_pid, + char *errstr, + size_t errstr_size, + rd_kafka_replyq_t replyq, + rd_kafka_resp_cb_t *resp_cb, + void *opaque) { + rd_kafka_buf_t *rkbuf; + int16_t ApiVersion; + + if (current_pid) { + ApiVersion = rd_kafka_broker_ApiVersion_supported( + rkb, RD_KAFKAP_InitProducerId, 3, 4, NULL); + if (ApiVersion == -1) { + rd_snprintf(errstr, errstr_size, + "InitProducerId (KIP-360) not supported by " + "broker, requires broker version >= 2.5.0: " + "unable to recover from previous " + "transactional error"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE; + } + } else { + ApiVersion = rd_kafka_broker_ApiVersion_supported( + rkb, RD_KAFKAP_InitProducerId, 0, 4, NULL); + + if (ApiVersion == -1) { + rd_snprintf(errstr, errstr_size, + "InitProducerId (KIP-98) not supported by " + "broker, requires broker " + "version >= 0.11.0"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE; + } + } + + rkbuf = rd_kafka_buf_new_flexver_request( + rkb, RD_KAFKAP_InitProducerId, 1, + 2 + (transactional_id ? strlen(transactional_id) : 0) + 4 + 8 + 4, + ApiVersion >= 2 /*flexver*/); + + /* transactional_id */ + rd_kafka_buf_write_str(rkbuf, transactional_id, -1); + + /* transaction_timeout_ms */ + rd_kafka_buf_write_i32(rkbuf, transaction_timeout_ms); + + if (ApiVersion >= 3) { + /* Current PID */ + rd_kafka_buf_write_i64(rkbuf, + current_pid ? current_pid->id : -1); + /* Current Epoch */ + rd_kafka_buf_write_i16(rkbuf, + current_pid ? current_pid->epoch : -1); + } + + rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0); + + /* Let the idempotence state handler perform retries */ + rkbuf->rkbuf_max_retries = RD_KAFKA_REQUEST_NO_RETRIES; + + rd_kafka_broker_buf_enq_replyq(rkb, rkbuf, replyq, resp_cb, opaque); + + return RD_KAFKA_RESP_ERR_NO_ERROR; +} + + +/** + * @brief Construct and send AddPartitionsToTxnRequest to \p rkb. + * + * The response (unparsed) will be handled by \p resp_cb served + * by queue \p replyq. + * + * @param rktps MUST be sorted by topic name. + * + * + * @returns RD_KAFKA_RESP_ERR_NO_ERROR if the request was enqueued for + * transmission, otherwise an error code. + */ +rd_kafka_resp_err_t +rd_kafka_AddPartitionsToTxnRequest(rd_kafka_broker_t *rkb, + const char *transactional_id, + rd_kafka_pid_t pid, + const rd_kafka_toppar_tqhead_t *rktps, + char *errstr, + size_t errstr_size, + rd_kafka_replyq_t replyq, + rd_kafka_resp_cb_t *resp_cb, + void *opaque) { + rd_kafka_buf_t *rkbuf; + int16_t ApiVersion = 0; + rd_kafka_toppar_t *rktp; + rd_kafka_topic_t *last_rkt = NULL; + size_t of_TopicCnt; + ssize_t of_PartCnt = -1; + int TopicCnt = 0, PartCnt = 0; + + ApiVersion = rd_kafka_broker_ApiVersion_supported( + rkb, RD_KAFKAP_AddPartitionsToTxn, 0, 0, NULL); + if (ApiVersion == -1) { + rd_snprintf(errstr, errstr_size, + "AddPartitionsToTxnRequest (KIP-98) not supported " + "by broker, requires broker version >= 0.11.0"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE; + } + + rkbuf = + rd_kafka_buf_new_request(rkb, RD_KAFKAP_AddPartitionsToTxn, 1, 500); + + /* transactional_id */ + rd_kafka_buf_write_str(rkbuf, transactional_id, -1); + + /* PID */ + rd_kafka_buf_write_i64(rkbuf, pid.id); + rd_kafka_buf_write_i16(rkbuf, pid.epoch); + + /* Topics/partitions array (count updated later) */ + of_TopicCnt = rd_kafka_buf_write_i32(rkbuf, 0); + + TAILQ_FOREACH(rktp, rktps, rktp_txnlink) { + if (last_rkt != rktp->rktp_rkt) { + + if (last_rkt) { + /* Update last topic's partition count field */ + rd_kafka_buf_update_i32(rkbuf, of_PartCnt, + PartCnt); + of_PartCnt = -1; + } + + /* Topic name */ + rd_kafka_buf_write_kstr(rkbuf, + rktp->rktp_rkt->rkt_topic); + /* Partition count, updated later */ + of_PartCnt = rd_kafka_buf_write_i32(rkbuf, 0); + + PartCnt = 0; + TopicCnt++; + last_rkt = rktp->rktp_rkt; + } + + /* Partition id */ + rd_kafka_buf_write_i32(rkbuf, rktp->rktp_partition); + PartCnt++; + } + + /* Update last partition and topic count fields */ + if (of_PartCnt != -1) + rd_kafka_buf_update_i32(rkbuf, (size_t)of_PartCnt, PartCnt); + rd_kafka_buf_update_i32(rkbuf, of_TopicCnt, TopicCnt); + + rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0); + + /* Let the handler perform retries so that it can pick + * up more added partitions. */ + rkbuf->rkbuf_max_retries = RD_KAFKA_REQUEST_NO_RETRIES; + + rd_kafka_broker_buf_enq_replyq(rkb, rkbuf, replyq, resp_cb, opaque); + + return RD_KAFKA_RESP_ERR_NO_ERROR; +} + + +/** + * @brief Construct and send AddOffsetsToTxnRequest to \p rkb. + * + * The response (unparsed) will be handled by \p resp_cb served + * by queue \p replyq. + * + * @returns RD_KAFKA_RESP_ERR_NO_ERROR if the request was enqueued for + * transmission, otherwise an error code. + */ +rd_kafka_resp_err_t +rd_kafka_AddOffsetsToTxnRequest(rd_kafka_broker_t *rkb, + const char *transactional_id, + rd_kafka_pid_t pid, + const char *group_id, + char *errstr, + size_t errstr_size, + rd_kafka_replyq_t replyq, + rd_kafka_resp_cb_t *resp_cb, + void *opaque) { + rd_kafka_buf_t *rkbuf; + int16_t ApiVersion = 0; + + ApiVersion = rd_kafka_broker_ApiVersion_supported( + rkb, RD_KAFKAP_AddOffsetsToTxn, 0, 0, NULL); + if (ApiVersion == -1) { + rd_snprintf(errstr, errstr_size, + "AddOffsetsToTxnRequest (KIP-98) not supported " + "by broker, requires broker version >= 0.11.0"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE; + } + + rkbuf = + rd_kafka_buf_new_request(rkb, RD_KAFKAP_AddOffsetsToTxn, 1, 100); + + /* transactional_id */ + rd_kafka_buf_write_str(rkbuf, transactional_id, -1); + + /* PID */ + rd_kafka_buf_write_i64(rkbuf, pid.id); + rd_kafka_buf_write_i16(rkbuf, pid.epoch); + + /* Group Id */ + rd_kafka_buf_write_str(rkbuf, group_id, -1); + + rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0); + + rkbuf->rkbuf_max_retries = RD_KAFKA_REQUEST_MAX_RETRIES; + + rd_kafka_broker_buf_enq_replyq(rkb, rkbuf, replyq, resp_cb, opaque); + + return RD_KAFKA_RESP_ERR_NO_ERROR; +} + + + +/** + * @brief Construct and send EndTxnRequest to \p rkb. + * + * The response (unparsed) will be handled by \p resp_cb served + * by queue \p replyq. + * + * @returns RD_KAFKA_RESP_ERR_NO_ERROR if the request was enqueued for + * transmission, otherwise an error code. + */ +rd_kafka_resp_err_t rd_kafka_EndTxnRequest(rd_kafka_broker_t *rkb, + const char *transactional_id, + rd_kafka_pid_t pid, + rd_bool_t committed, + char *errstr, + size_t errstr_size, + rd_kafka_replyq_t replyq, + rd_kafka_resp_cb_t *resp_cb, + void *opaque) { + rd_kafka_buf_t *rkbuf; + int16_t ApiVersion = 0; + + ApiVersion = rd_kafka_broker_ApiVersion_supported(rkb, RD_KAFKAP_EndTxn, + 0, 1, NULL); + if (ApiVersion == -1) { + rd_snprintf(errstr, errstr_size, + "EndTxnRequest (KIP-98) not supported " + "by broker, requires broker version >= 0.11.0"); + rd_kafka_replyq_destroy(&replyq); + return RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE; + } + + rkbuf = rd_kafka_buf_new_request(rkb, RD_KAFKAP_EndTxn, 1, 500); + + /* transactional_id */ + rd_kafka_buf_write_str(rkbuf, transactional_id, -1); + + /* PID */ + rd_kafka_buf_write_i64(rkbuf, pid.id); + rd_kafka_buf_write_i16(rkbuf, pid.epoch); + + /* Committed */ + rd_kafka_buf_write_bool(rkbuf, committed); + rkbuf->rkbuf_u.EndTxn.commit = committed; + + rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0); + + rkbuf->rkbuf_max_retries = RD_KAFKA_REQUEST_MAX_RETRIES; + + rd_kafka_broker_buf_enq_replyq(rkb, rkbuf, replyq, resp_cb, opaque); + + return RD_KAFKA_RESP_ERR_NO_ERROR; +} + + + +/** + * @name Unit tests + * @{ + * + * + * + * + */ + +/** + * @brief Create \p cnt messages, starting at \p msgid, and add them + * to \p rkmq. + * + * @returns the number of messages added. + */ +static int ut_create_msgs(rd_kafka_msgq_t *rkmq, uint64_t msgid, int cnt) { + int i; + + for (i = 0; i < cnt; i++) { + rd_kafka_msg_t *rkm; + + rkm = ut_rd_kafka_msg_new(0); + rkm->rkm_u.producer.msgid = msgid++; + rkm->rkm_ts_enq = rd_clock(); + rkm->rkm_ts_timeout = rkm->rkm_ts_enq + (900 * 1000 * 1000); + + rd_kafka_msgq_enq(rkmq, rkm); + } + + return cnt; +} + +/** + * @brief Idempotent Producer request/response unit tests + * + * The current test verifies proper handling of the following case: + * Batch 0 succeeds + * Batch 1 fails with temporary error + * Batch 2,3 fails with out of order sequence + * Retry Batch 1-3 should succeed. + */ +static int unittest_idempotent_producer(void) { + rd_kafka_t *rk; + rd_kafka_conf_t *conf; + rd_kafka_broker_t *rkb; +#define _BATCH_CNT 4 +#define _MSGS_PER_BATCH 3 + const int msgcnt = _BATCH_CNT * _MSGS_PER_BATCH; + int remaining_batches; + uint64_t msgid = 1; + rd_kafka_toppar_t *rktp; + rd_kafka_pid_t pid = {.id = 1000, .epoch = 0}; + struct rd_kafka_Produce_result result = {.offset = 1, + .timestamp = 1000}; + rd_kafka_queue_t *rkqu; + rd_kafka_event_t *rkev; + rd_kafka_buf_t *request[_BATCH_CNT]; + int rcnt = 0; + int retry_msg_cnt = 0; + int drcnt = 0; + rd_kafka_msgq_t rkmq = RD_KAFKA_MSGQ_INITIALIZER(rkmq); + const char *tmp; + int i, r; + + RD_UT_SAY("Verifying idempotent producer error handling"); + + conf = rd_kafka_conf_new(); + rd_kafka_conf_set(conf, "batch.num.messages", "3", NULL, 0); + rd_kafka_conf_set(conf, "retry.backoff.ms", "1", NULL, 0); + if ((tmp = rd_getenv("TEST_DEBUG", NULL))) + rd_kafka_conf_set(conf, "debug", tmp, NULL, 0); + if (rd_kafka_conf_set(conf, "enable.idempotence", "true", NULL, 0) != + RD_KAFKA_CONF_OK) + RD_UT_FAIL("Failed to enable idempotence"); + rd_kafka_conf_set_events(conf, RD_KAFKA_EVENT_DR); + + rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf, NULL, 0); + RD_UT_ASSERT(rk, "failed to create producer"); + + rkqu = rd_kafka_queue_get_main(rk); + + /* We need a broker handle, use a logical broker to avoid + * any connection attempts. */ + rkb = rd_kafka_broker_add_logical(rk, "unittest"); + + /* Have the broker support everything so msgset_writer selects + * the most up-to-date output features. */ + rd_kafka_broker_lock(rkb); + rkb->rkb_features = RD_KAFKA_FEATURE_UNITTEST | RD_KAFKA_FEATURE_ALL; + rd_kafka_broker_unlock(rkb); + + /* Get toppar */ + rktp = rd_kafka_toppar_get2(rk, "uttopic", 0, rd_false, rd_true); + RD_UT_ASSERT(rktp, "failed to get toppar"); + + /* Set the topic as exists so messages are enqueued on + * the desired rktp away (otherwise UA partition) */ + rd_ut_kafka_topic_set_topic_exists(rktp->rktp_rkt, 1, -1); + + /* Produce messages */ + ut_create_msgs(&rkmq, 1, msgcnt); + + /* Set the pid */ + rd_kafka_idemp_set_state(rk, RD_KAFKA_IDEMP_STATE_WAIT_PID); + rd_kafka_idemp_pid_update(rkb, pid); + pid = rd_kafka_idemp_get_pid(rk); + RD_UT_ASSERT(rd_kafka_pid_valid(pid), "PID is invalid"); + rd_kafka_toppar_pid_change(rktp, pid, msgid); + + remaining_batches = _BATCH_CNT; + + /* Create a ProduceRequest for each batch */ + for (rcnt = 0; rcnt < remaining_batches; rcnt++) { + size_t msize; + request[rcnt] = rd_kafka_msgset_create_ProduceRequest( + rkb, rktp, &rkmq, rd_kafka_idemp_get_pid(rk), 0, &msize); + RD_UT_ASSERT(request[rcnt], "request #%d failed", rcnt); + } + + RD_UT_ASSERT(rd_kafka_msgq_len(&rkmq) == 0, + "expected input message queue to be empty, " + "but still has %d message(s)", + rd_kafka_msgq_len(&rkmq)); + + /* + * Mock handling of each request + */ + + /* Batch 0: accepted */ + i = 0; + r = rd_kafka_msgq_len(&request[i]->rkbuf_batch.msgq); + RD_UT_ASSERT(r == _MSGS_PER_BATCH, "."); + rd_kafka_msgbatch_handle_Produce_result(rkb, &request[i]->rkbuf_batch, + RD_KAFKA_RESP_ERR_NO_ERROR, + &result, request[i]); + result.offset += r; + RD_UT_ASSERT(rd_kafka_msgq_len(&rktp->rktp_msgq) == 0, + "batch %d: expected no messages in rktp_msgq, not %d", i, + rd_kafka_msgq_len(&rktp->rktp_msgq)); + rd_kafka_buf_destroy(request[i]); + remaining_batches--; + + /* Batch 1: fail, triggering retry (re-enq on rktp_msgq) */ + i = 1; + r = rd_kafka_msgq_len(&request[i]->rkbuf_batch.msgq); + RD_UT_ASSERT(r == _MSGS_PER_BATCH, "."); + rd_kafka_msgbatch_handle_Produce_result( + rkb, &request[i]->rkbuf_batch, + RD_KAFKA_RESP_ERR_NOT_LEADER_FOR_PARTITION, &result, request[i]); + retry_msg_cnt += r; + RD_UT_ASSERT(rd_kafka_msgq_len(&rktp->rktp_msgq) == retry_msg_cnt, + "batch %d: expected %d messages in rktp_msgq, not %d", i, + retry_msg_cnt, rd_kafka_msgq_len(&rktp->rktp_msgq)); + rd_kafka_buf_destroy(request[i]); + + /* Batch 2: OUT_OF_ORDER, triggering retry .. */ + i = 2; + r = rd_kafka_msgq_len(&request[i]->rkbuf_batch.msgq); + RD_UT_ASSERT(r == _MSGS_PER_BATCH, "."); + rd_kafka_msgbatch_handle_Produce_result( + rkb, &request[i]->rkbuf_batch, + RD_KAFKA_RESP_ERR_OUT_OF_ORDER_SEQUENCE_NUMBER, &result, + request[i]); + retry_msg_cnt += r; + RD_UT_ASSERT(rd_kafka_msgq_len(&rktp->rktp_msgq) == retry_msg_cnt, + "batch %d: expected %d messages in rktp_xmit_msgq, not %d", + i, retry_msg_cnt, rd_kafka_msgq_len(&rktp->rktp_msgq)); + rd_kafka_buf_destroy(request[i]); + + /* Batch 3: OUT_OF_ORDER, triggering retry .. */ + i = 3; + r = rd_kafka_msgq_len(&request[i]->rkbuf_batch.msgq); + rd_kafka_msgbatch_handle_Produce_result( + rkb, &request[i]->rkbuf_batch, + RD_KAFKA_RESP_ERR_OUT_OF_ORDER_SEQUENCE_NUMBER, &result, + request[i]); + retry_msg_cnt += r; + RD_UT_ASSERT(rd_kafka_msgq_len(&rktp->rktp_msgq) == retry_msg_cnt, + "batch %d: expected %d messages in rktp_xmit_msgq, not %d", + i, retry_msg_cnt, rd_kafka_msgq_len(&rktp->rktp_msgq)); + rd_kafka_buf_destroy(request[i]); + + + /* Retried messages will have been moved to rktp_msgq, + * move them back to our local queue. */ + rd_kafka_toppar_lock(rktp); + rd_kafka_msgq_move(&rkmq, &rktp->rktp_msgq); + rd_kafka_toppar_unlock(rktp); + + RD_UT_ASSERT(rd_kafka_msgq_len(&rkmq) == retry_msg_cnt, + "Expected %d messages in retry queue, not %d", + retry_msg_cnt, rd_kafka_msgq_len(&rkmq)); + + /* Sleep a short while to make sure the retry backoff expires. */ + rd_usleep(5 * 1000, NULL); /* 5ms */ + + /* + * Create requests for remaining batches. + */ + for (rcnt = 0; rcnt < remaining_batches; rcnt++) { + size_t msize; + request[rcnt] = rd_kafka_msgset_create_ProduceRequest( + rkb, rktp, &rkmq, rd_kafka_idemp_get_pid(rk), 0, &msize); + RD_UT_ASSERT(request[rcnt], + "Failed to create retry #%d (%d msgs in queue)", + rcnt, rd_kafka_msgq_len(&rkmq)); + } + + /* + * Mock handling of each request, they will now succeed. + */ + for (i = 0; i < rcnt; i++) { + r = rd_kafka_msgq_len(&request[i]->rkbuf_batch.msgq); + rd_kafka_msgbatch_handle_Produce_result( + rkb, &request[i]->rkbuf_batch, RD_KAFKA_RESP_ERR_NO_ERROR, + &result, request[i]); + result.offset += r; + rd_kafka_buf_destroy(request[i]); + } + + retry_msg_cnt = 0; + RD_UT_ASSERT(rd_kafka_msgq_len(&rktp->rktp_msgq) == retry_msg_cnt, + "batch %d: expected %d messages in rktp_xmit_msgq, not %d", + i, retry_msg_cnt, rd_kafka_msgq_len(&rktp->rktp_msgq)); + + /* + * Wait for delivery reports, they should all be successful. + */ + while ((rkev = rd_kafka_queue_poll(rkqu, 1000))) { + const rd_kafka_message_t *rkmessage; + + RD_UT_SAY("Got %s event with %d message(s)", + rd_kafka_event_name(rkev), + (int)rd_kafka_event_message_count(rkev)); + + while ((rkmessage = rd_kafka_event_message_next(rkev))) { + RD_UT_SAY(" DR for message: %s: (persistence=%d)", + rd_kafka_err2str(rkmessage->err), + rd_kafka_message_status(rkmessage)); + if (rkmessage->err) + RD_UT_WARN(" ^ Should not have failed"); + else + drcnt++; + } + rd_kafka_event_destroy(rkev); + } + + /* Should be no more messages in queues */ + r = rd_kafka_outq_len(rk); + RD_UT_ASSERT(r == 0, "expected outq to return 0, not %d", r); + + /* Verify the expected number of good delivery reports were seen */ + RD_UT_ASSERT(drcnt == msgcnt, "expected %d DRs, not %d", msgcnt, drcnt); + + rd_kafka_queue_destroy(rkqu); + rd_kafka_toppar_destroy(rktp); + rd_kafka_broker_destroy(rkb); + rd_kafka_destroy(rk); + + RD_UT_PASS(); + return 0; +} + +/** + * @brief Request/response unit tests + */ +int unittest_request(void) { + int fails = 0; + + fails += unittest_idempotent_producer(); + + return fails; +} + +/**@}*/ |