You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ph...@apache.org on 2018/06/06 14:15:00 UTC
[32/51] [partial] nifi-minifi-cpp git commit: MINIFICPP-512 - upgrade
to librdkafka 0.11.4
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/src/rdkafka_broker.c
----------------------------------------------------------------------
diff --git a/thirdparty/librdkafka-0.11.1/src/rdkafka_broker.c b/thirdparty/librdkafka-0.11.1/src/rdkafka_broker.c
deleted file mode 100644
index 742e0fd..0000000
--- a/thirdparty/librdkafka-0.11.1/src/rdkafka_broker.c
+++ /dev/null
@@ -1,3797 +0,0 @@
-/*
- * 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.
- */
-
-
-
-#ifndef _MSC_VER
-#define _GNU_SOURCE
-/*
- * AIX defines this and the value needs to be set correctly. For Solaris,
- * src/rd.h defines _POSIX_SOURCE to be 200809L, which corresponds to XPG7,
- * which itself is not compatible with _XOPEN_SOURCE on that platform.
- */
-#if !defined(_AIX) && !defined(__sun)
-#define _XOPEN_SOURCE
-#endif
-#include <signal.h>
-#endif
-
-#include <stdio.h>
-#include <stdarg.h>
-#include <string.h>
-#include <ctype.h>
-
-#include "rd.h"
-#include "rdkafka_int.h"
-#include "rdkafka_msg.h"
-#include "rdkafka_msgset.h"
-#include "rdkafka_topic.h"
-#include "rdkafka_partition.h"
-#include "rdkafka_broker.h"
-#include "rdkafka_offset.h"
-#include "rdkafka_transport.h"
-#include "rdkafka_proto.h"
-#include "rdkafka_buf.h"
-#include "rdkafka_request.h"
-#include "rdkafka_sasl.h"
-#include "rdkafka_interceptor.h"
-#include "rdtime.h"
-#include "rdcrc32.h"
-#include "rdrand.h"
-#include "rdkafka_lz4.h"
-#if WITH_SSL
-#include <openssl/err.h>
-#endif
-#include "rdendian.h"
-
-
-const char *rd_kafka_broker_state_names[] = {
- "INIT",
- "DOWN",
- "CONNECT",
- "AUTH",
- "UP",
- "UPDATE",
- "APIVERSION_QUERY",
- "AUTH_HANDSHAKE"
-};
-
-const char *rd_kafka_secproto_names[] = {
- [RD_KAFKA_PROTO_PLAINTEXT] = "plaintext",
- [RD_KAFKA_PROTO_SSL] = "ssl",
- [RD_KAFKA_PROTO_SASL_PLAINTEXT] = "sasl_plaintext",
- [RD_KAFKA_PROTO_SASL_SSL] = "sasl_ssl",
- NULL
-};
-
-
-
-
-
-
-
-#define rd_kafka_broker_terminating(rkb) \
- (rd_refcnt_get(&(rkb)->rkb_refcnt) <= 1)
-
-
-/**
- * Construct broker nodename.
- */
-static void rd_kafka_mk_nodename (char *dest, size_t dsize,
- const char *name, uint16_t port) {
- rd_snprintf(dest, dsize, "%s:%hu", name, port);
-}
-
-/**
- * Construct descriptive broker name
- */
-static void rd_kafka_mk_brokername (char *dest, size_t dsize,
- rd_kafka_secproto_t proto,
- const char *nodename, int32_t nodeid,
- rd_kafka_confsource_t source) {
-
- /* Prepend protocol name to brokername, unless it is a
- * standard plaintext broker in which case we omit the protocol part. */
- if (proto != RD_KAFKA_PROTO_PLAINTEXT) {
- int r = rd_snprintf(dest, dsize, "%s://",
- rd_kafka_secproto_names[proto]);
- if (r >= (int)dsize) /* Skip proto name if it wont fit.. */
- r = 0;
-
- dest += r;
- dsize -= r;
- }
-
- if (nodeid == RD_KAFKA_NODEID_UA)
- rd_snprintf(dest, dsize, "%s/%s",
- nodename,
- source == RD_KAFKA_INTERNAL ?
- "internal":"bootstrap");
- else
- rd_snprintf(dest, dsize, "%s/%"PRId32, nodename, nodeid);
-}
-
-
-/**
- * @brief Enable protocol feature(s) for the current broker.
- *
- * Locality: broker thread
- */
-static void rd_kafka_broker_feature_enable (rd_kafka_broker_t *rkb,
- int features) {
- if (features & rkb->rkb_features)
- return;
-
- rkb->rkb_features |= features;
- rd_rkb_dbg(rkb, BROKER | RD_KAFKA_DBG_PROTOCOL | RD_KAFKA_DBG_FEATURE,
- "FEATURE",
- "Updated enabled protocol features +%s to %s",
- rd_kafka_features2str(features),
- rd_kafka_features2str(rkb->rkb_features));
-}
-
-
-/**
- * @brief Disable protocol feature(s) for the current broker.
- *
- * Locality: broker thread
- */
-static void rd_kafka_broker_feature_disable (rd_kafka_broker_t *rkb,
- int features) {
- if (!(features & rkb->rkb_features))
- return;
-
- rkb->rkb_features &= ~features;
- rd_rkb_dbg(rkb, BROKER | RD_KAFKA_DBG_PROTOCOL | RD_KAFKA_DBG_FEATURE,
- "FEATURE",
- "Updated enabled protocol features -%s to %s",
- rd_kafka_features2str(features),
- rd_kafka_features2str(rkb->rkb_features));
-}
-
-
-/**
- * @brief Set protocol feature(s) for the current broker.
- *
- * @remark This replaces the previous feature set.
- *
- * @locality broker thread
- * @locks rd_kafka_broker_lock()
- */
-static void rd_kafka_broker_features_set (rd_kafka_broker_t *rkb, int features) {
- if (rkb->rkb_features == features)
- return;
-
- rkb->rkb_features = features;
- rd_rkb_dbg(rkb, BROKER, "FEATURE",
- "Updated enabled protocol features to %s",
- rd_kafka_features2str(rkb->rkb_features));
-}
-
-
-/**
- * @brief Check and return supported ApiVersion for \p ApiKey.
- *
- * @returns the highest supported ApiVersion in the specified range (inclusive)
- * or -1 if the ApiKey is not supported or no matching ApiVersion.
- * The current feature set is also returned in \p featuresp
- * @locks none
- * @locality any
- */
-int16_t rd_kafka_broker_ApiVersion_supported (rd_kafka_broker_t *rkb,
- int16_t ApiKey,
- int16_t minver, int16_t maxver,
- int *featuresp) {
- struct rd_kafka_ApiVersion skel = { .ApiKey = ApiKey };
- struct rd_kafka_ApiVersion ret = RD_ZERO_INIT, *retp;
-
- rd_kafka_broker_lock(rkb);
- retp = bsearch(&skel, rkb->rkb_ApiVersions, rkb->rkb_ApiVersions_cnt,
- sizeof(*rkb->rkb_ApiVersions),
- rd_kafka_ApiVersion_key_cmp);
- if (retp)
- ret = *retp;
- if (featuresp)
- *featuresp = rkb->rkb_features;
- rd_kafka_broker_unlock(rkb);
-
- if (!retp)
- return -1;
-
- if (ret.MaxVer < maxver) {
- if (ret.MaxVer < minver)
- return -1;
- else
- return ret.MaxVer;
- } else if (ret.MinVer > maxver)
- return -1;
- else
- return maxver;
-}
-
-
-/**
- * Locks: rd_kafka_broker_lock() MUST be held.
- * Locality: broker thread
- */
-void rd_kafka_broker_set_state (rd_kafka_broker_t *rkb, int state) {
- if ((int)rkb->rkb_state == state)
- return;
-
- rd_kafka_dbg(rkb->rkb_rk, BROKER, "STATE",
- "%s: Broker changed state %s -> %s",
- rkb->rkb_name,
- rd_kafka_broker_state_names[rkb->rkb_state],
- rd_kafka_broker_state_names[state]);
-
- if (rkb->rkb_source == RD_KAFKA_INTERNAL) {
- /* no-op */
- } else if (state == RD_KAFKA_BROKER_STATE_DOWN &&
- !rkb->rkb_down_reported &&
- rkb->rkb_state != RD_KAFKA_BROKER_STATE_APIVERSION_QUERY) {
- /* Propagate ALL_BROKERS_DOWN event if all brokers are
- * now down, unless we're terminating.
- * Dont do this if we're querying for ApiVersion since it
- * is bound to fail once on older brokers. */
- if (rd_atomic32_add(&rkb->rkb_rk->rk_broker_down_cnt, 1) ==
- rd_atomic32_get(&rkb->rkb_rk->rk_broker_cnt) &&
- !rd_atomic32_get(&rkb->rkb_rk->rk_terminate))
- rd_kafka_op_err(rkb->rkb_rk,
- RD_KAFKA_RESP_ERR__ALL_BROKERS_DOWN,
- "%i/%i brokers are down",
- rd_atomic32_get(&rkb->rkb_rk->
- rk_broker_down_cnt),
- rd_atomic32_get(&rkb->rkb_rk->
- rk_broker_cnt));
- rkb->rkb_down_reported = 1;
-
- } else if (rkb->rkb_state >= RD_KAFKA_BROKER_STATE_UP &&
- rkb->rkb_down_reported) {
- rd_atomic32_sub(&rkb->rkb_rk->rk_broker_down_cnt, 1);
- rkb->rkb_down_reported = 0;
- }
-
- rkb->rkb_state = state;
- rkb->rkb_ts_state = rd_clock();
-
- rd_kafka_brokers_broadcast_state_change(rkb->rkb_rk);
-}
-
-
-/**
- * @brief Locks broker, acquires the states, unlocks, and returns
- * the state.
- * @locks !broker_lock
- * @locality any
- */
-int rd_kafka_broker_get_state (rd_kafka_broker_t *rkb) {
- int state;
- rd_kafka_broker_lock(rkb);
- state = rkb->rkb_state;
- rd_kafka_broker_unlock(rkb);
- return state;
-}
-
-
-/**
- * Failure propagation to application.
- * Will tear down connection to broker and trigger a reconnect.
- *
- * If 'fmt' is NULL nothing will be logged or propagated to the application.
- *
- * \p level is the log level, <=LOG_INFO will be logged while =LOG_DEBUG will
- * be debug-logged.
- *
- * Locality: Broker thread
- */
-void rd_kafka_broker_fail (rd_kafka_broker_t *rkb,
- int level, rd_kafka_resp_err_t err,
- const char *fmt, ...) {
- va_list ap;
- int errno_save = errno;
- rd_kafka_bufq_t tmpq_waitresp, tmpq;
- int old_state;
-
- rd_kafka_assert(rkb->rkb_rk, thrd_is_current(rkb->rkb_thread));
-
- rd_kafka_dbg(rkb->rkb_rk, BROKER | RD_KAFKA_DBG_PROTOCOL, "BROKERFAIL",
- "%s: failed: err: %s: (errno: %s)",
- rkb->rkb_name, rd_kafka_err2str(err),
- rd_strerror(errno_save));
-
- rkb->rkb_err.err = errno_save;
-
- if (rkb->rkb_transport) {
- rd_kafka_transport_close(rkb->rkb_transport);
- rkb->rkb_transport = NULL;
- }
-
- rkb->rkb_req_timeouts = 0;
-
- if (rkb->rkb_recv_buf) {
- rd_kafka_buf_destroy(rkb->rkb_recv_buf);
- rkb->rkb_recv_buf = NULL;
- }
-
- rd_kafka_broker_lock(rkb);
-
- /* The caller may omit the format if it thinks this is a recurring
- * failure, in which case the following things are omitted:
- * - log message
- * - application OP_ERR
- * - metadata request
- *
- * Dont log anything if this was the termination signal, or if the
- * socket disconnected while trying ApiVersionRequest.
- */
- if (fmt &&
- !(errno_save == EINTR &&
- rd_atomic32_get(&rkb->rkb_rk->rk_terminate)) &&
- !(err == RD_KAFKA_RESP_ERR__TRANSPORT &&
- rkb->rkb_state == RD_KAFKA_BROKER_STATE_APIVERSION_QUERY)) {
- int of;
-
- /* Insert broker name in log message if it fits. */
- of = rd_snprintf(rkb->rkb_err.msg, sizeof(rkb->rkb_err.msg),
- "%s: ", rkb->rkb_name);
- if (of >= (int)sizeof(rkb->rkb_err.msg))
- of = 0;
- va_start(ap, fmt);
- rd_vsnprintf(rkb->rkb_err.msg+of,
- sizeof(rkb->rkb_err.msg)-of, fmt, ap);
- va_end(ap);
-
- if (level >= LOG_DEBUG)
- rd_kafka_dbg(rkb->rkb_rk, BROKER, "FAIL",
- "%s", rkb->rkb_err.msg);
- else {
- /* Don't log if an error callback is registered */
- if (!rkb->rkb_rk->rk_conf.error_cb)
- rd_kafka_log(rkb->rkb_rk, level, "FAIL",
- "%s", rkb->rkb_err.msg);
- /* Send ERR op back to application for processing. */
- rd_kafka_op_err(rkb->rkb_rk, err,
- "%s", rkb->rkb_err.msg);
- }
- }
-
- /* If we're currently asking for ApiVersion and the connection
- * went down it probably means the broker does not support that request
- * and tore down the connection. In this case we disable that feature flag. */
- if (rkb->rkb_state == RD_KAFKA_BROKER_STATE_APIVERSION_QUERY)
- rd_kafka_broker_feature_disable(rkb, RD_KAFKA_FEATURE_APIVERSION);
-
- /* Set broker state */
- old_state = rkb->rkb_state;
- rd_kafka_broker_set_state(rkb, RD_KAFKA_BROKER_STATE_DOWN);
-
- /* Unlock broker since a requeue will try to lock it. */
- rd_kafka_broker_unlock(rkb);
-
- /*
- * Purge all buffers
- * (put bufs on a temporary queue since bufs may be requeued,
- * make sure outstanding requests are re-enqueued before
- * bufs on outbufs queue.)
- */
- rd_kafka_bufq_init(&tmpq_waitresp);
- rd_kafka_bufq_init(&tmpq);
- rd_kafka_bufq_concat(&tmpq_waitresp, &rkb->rkb_waitresps);
- rd_kafka_bufq_concat(&tmpq, &rkb->rkb_outbufs);
- rd_atomic32_init(&rkb->rkb_blocking_request_cnt, 0);
-
- /* Purge the buffers (might get re-enqueued in case of retries) */
- rd_kafka_bufq_purge(rkb, &tmpq_waitresp, err);
-
- /* Put the outbufs back on queue */
- rd_kafka_bufq_concat(&rkb->rkb_outbufs, &tmpq);
-
- /* Update bufq for connection reset:
- * - Purge connection-setup requests from outbufs since they will be
- * reissued on the next connect.
- * - Reset any partially sent buffer's offset.
- */
- rd_kafka_bufq_connection_reset(rkb, &rkb->rkb_outbufs);
-
- /* Extra debugging for tracking termination-hang issues:
- * show what is keeping this broker from decommissioning. */
- if (rd_kafka_terminating(rkb->rkb_rk) &&
- !rd_kafka_broker_terminating(rkb)) {
- rd_rkb_dbg(rkb, BROKER | RD_KAFKA_DBG_PROTOCOL, "BRKTERM",
- "terminating: broker still has %d refcnt(s), "
- "%"PRId32" buffer(s), %d partition(s)",
- rd_refcnt_get(&rkb->rkb_refcnt),
- rd_kafka_bufq_cnt(&rkb->rkb_outbufs),
- rkb->rkb_toppar_cnt);
- rd_kafka_bufq_dump(rkb, "BRKOUTBUFS", &rkb->rkb_outbufs);
-#if ENABLE_SHAREDPTR_DEBUG
- if (rd_refcnt_get(&rkb->rkb_refcnt) > 1) {
- rd_rkb_dbg(rkb, BROKER, "BRKTERM",
- "Dumping shared pointers: "
- "this broker is %p", rkb);
- rd_shared_ptrs_dump();
- }
-#endif
- }
-
-
- /* Query for topic leaders to quickly pick up on failover. */
- if (fmt && err != RD_KAFKA_RESP_ERR__DESTROY &&
- old_state >= RD_KAFKA_BROKER_STATE_UP)
- rd_kafka_metadata_refresh_known_topics(rkb->rkb_rk, NULL,
- 1/*force*/,
- "broker down");
-}
-
-
-
-
-
-/**
- * Scan bufq for buffer timeouts, trigger buffer callback on timeout.
- *
- * If \p partial_cntp is non-NULL any partially sent buffers will increase
- * the provided counter by 1.
- *
- * @returns the number of timed out buffers.
- *
- * @locality broker thread
- */
-static int rd_kafka_broker_bufq_timeout_scan (rd_kafka_broker_t *rkb,
- int is_waitresp_q,
- rd_kafka_bufq_t *rkbq,
- int *partial_cntp,
- rd_kafka_resp_err_t err,
- rd_ts_t now) {
- rd_kafka_buf_t *rkbuf, *tmp;
- int cnt = 0;
-
- TAILQ_FOREACH_SAFE(rkbuf, &rkbq->rkbq_bufs, rkbuf_link, tmp) {
-
- if (likely(now && rkbuf->rkbuf_ts_timeout > now))
- continue;
-
- if (partial_cntp && rd_slice_offset(&rkbuf->rkbuf_reader) > 0)
- (*partial_cntp)++;
-
- /* Convert rkbuf_ts_sent to elapsed time since request */
- if (rkbuf->rkbuf_ts_sent)
- rkbuf->rkbuf_ts_sent = now - rkbuf->rkbuf_ts_sent;
- else
- rkbuf->rkbuf_ts_sent = now - rkbuf->rkbuf_ts_enq;
-
- rd_kafka_bufq_deq(rkbq, rkbuf);
-
- if (is_waitresp_q && rkbuf->rkbuf_flags & RD_KAFKA_OP_F_BLOCKING
- && rd_atomic32_sub(&rkb->rkb_blocking_request_cnt, 1) == 0)
- rd_kafka_brokers_broadcast_state_change(rkb->rkb_rk);
-
- rd_kafka_buf_callback(rkb->rkb_rk, rkb, err, NULL, rkbuf);
- cnt++;
- }
-
- return cnt;
-}
-
-
-/**
- * Scan the wait-response and outbuf queues for message timeouts.
- *
- * Locality: Broker thread
- */
-static void rd_kafka_broker_timeout_scan (rd_kafka_broker_t *rkb, rd_ts_t now) {
- int req_cnt, retry_cnt, q_cnt;
-
- rd_kafka_assert(rkb->rkb_rk, thrd_is_current(rkb->rkb_thread));
-
- /* Outstanding requests waiting for response */
- req_cnt = rd_kafka_broker_bufq_timeout_scan(
- rkb, 1, &rkb->rkb_waitresps, NULL,
- RD_KAFKA_RESP_ERR__TIMED_OUT, now);
- /* Requests in retry queue */
- retry_cnt = rd_kafka_broker_bufq_timeout_scan(
- rkb, 0, &rkb->rkb_retrybufs, NULL,
- RD_KAFKA_RESP_ERR__TIMED_OUT, now);
- /* Requests in local queue not sent yet. */
- q_cnt = rd_kafka_broker_bufq_timeout_scan(
- rkb, 0, &rkb->rkb_outbufs, &req_cnt,
- RD_KAFKA_RESP_ERR__TIMED_OUT, now);
-
- if (req_cnt + retry_cnt + q_cnt > 0) {
- rd_rkb_dbg(rkb, MSG|RD_KAFKA_DBG_BROKER,
- "REQTMOUT", "Timed out %i+%i+%i requests",
- req_cnt, retry_cnt, q_cnt);
-
- /* Fail the broker if socket.max.fails is configured and
- * now exceeded. */
- rkb->rkb_req_timeouts += req_cnt + q_cnt;
- rd_atomic64_add(&rkb->rkb_c.req_timeouts, req_cnt + q_cnt);
-
- /* If this was an in-flight request that timed out, or
- * the other queues has reached the socket.max.fails threshold,
- * we need to take down the connection. */
- if ((req_cnt > 0 ||
- (rkb->rkb_rk->rk_conf.socket_max_fails &&
- rkb->rkb_req_timeouts >=
- rkb->rkb_rk->rk_conf.socket_max_fails)) &&
- rkb->rkb_state >= RD_KAFKA_BROKER_STATE_UP) {
- char rttinfo[32];
- /* Print average RTT (if avail) to help diagnose. */
- rd_avg_calc(&rkb->rkb_avg_rtt, now);
- if (rkb->rkb_avg_rtt.ra_v.avg)
- rd_snprintf(rttinfo, sizeof(rttinfo),
- " (average rtt %.3fms)",
- (float)(rkb->rkb_avg_rtt.ra_v.avg/
- 1000.0f));
- else
- rttinfo[0] = 0;
- errno = ETIMEDOUT;
- rd_kafka_broker_fail(rkb, LOG_ERR,
- RD_KAFKA_RESP_ERR__MSG_TIMED_OUT,
- "%i request(s) timed out: "
- "disconnect%s",
- rkb->rkb_req_timeouts, rttinfo);
- }
- }
-}
-
-
-
-static ssize_t
-rd_kafka_broker_send (rd_kafka_broker_t *rkb, rd_slice_t *slice) {
- ssize_t r;
- char errstr[128];
-
- rd_kafka_assert(rkb->rkb_rk, rkb->rkb_state >= RD_KAFKA_BROKER_STATE_UP);
- rd_kafka_assert(rkb->rkb_rk, rkb->rkb_transport);
-
- r = rd_kafka_transport_send(rkb->rkb_transport, slice,
- errstr, sizeof(errstr));
-
- if (r == -1) {
- rd_kafka_broker_fail(rkb, LOG_ERR, RD_KAFKA_RESP_ERR__TRANSPORT,
- "Send failed: %s", errstr);
- rd_atomic64_add(&rkb->rkb_c.tx_err, 1);
- return -1;
- }
-
- rd_atomic64_add(&rkb->rkb_c.tx_bytes, r);
- rd_atomic64_add(&rkb->rkb_c.tx, 1);
- return r;
-}
-
-
-
-
-static int rd_kafka_broker_resolve (rd_kafka_broker_t *rkb) {
- const char *errstr;
-
- if (rkb->rkb_rsal &&
- rkb->rkb_t_rsal_last + rkb->rkb_rk->rk_conf.broker_addr_ttl <
- time(NULL)) { // FIXME: rd_clock()
- /* Address list has expired. */
- rd_sockaddr_list_destroy(rkb->rkb_rsal);
- rkb->rkb_rsal = NULL;
- }
-
- if (!rkb->rkb_rsal) {
- /* Resolve */
-
- rkb->rkb_rsal = rd_getaddrinfo(rkb->rkb_nodename,
- RD_KAFKA_PORT_STR,
- AI_ADDRCONFIG,
- rkb->rkb_rk->rk_conf.
- broker_addr_family,
- SOCK_STREAM,
- IPPROTO_TCP, &errstr);
-
- if (!rkb->rkb_rsal) {
- rd_kafka_broker_fail(rkb, LOG_ERR,
- RD_KAFKA_RESP_ERR__RESOLVE,
- /* Avoid duplicate log messages */
- rkb->rkb_err.err == errno ?
- NULL :
- "Failed to resolve '%s': %s",
- rkb->rkb_nodename, errstr);
- return -1;
- }
- }
-
- return 0;
-}
-
-
-static void rd_kafka_broker_buf_enq0 (rd_kafka_broker_t *rkb,
- rd_kafka_buf_t *rkbuf, int at_head) {
- rd_kafka_assert(rkb->rkb_rk, thrd_is_current(rkb->rkb_thread));
-
- rkbuf->rkbuf_ts_enq = rd_clock();
-
- /* Set timeout if not already set */
- if (!rkbuf->rkbuf_ts_timeout)
- rkbuf->rkbuf_ts_timeout = rkbuf->rkbuf_ts_enq +
- rkb->rkb_rk->rk_conf.socket_timeout_ms * 1000;
-
- if (unlikely(at_head)) {
- /* Insert message at head of queue */
- rd_kafka_buf_t *prev, *after = NULL;
-
- /* Put us behind any flash messages and partially sent buffers.
- * We need to check if buf corrid is set rather than
- * rkbuf_of since SSL_write may return 0 and expect the
- * exact same arguments the next call. */
- TAILQ_FOREACH(prev, &rkb->rkb_outbufs.rkbq_bufs, rkbuf_link) {
- if (!(prev->rkbuf_flags & RD_KAFKA_OP_F_FLASH) &&
- prev->rkbuf_corrid == 0)
- break;
- after = prev;
- }
-
- if (after)
- TAILQ_INSERT_AFTER(&rkb->rkb_outbufs.rkbq_bufs,
- after, rkbuf, rkbuf_link);
- else
- TAILQ_INSERT_HEAD(&rkb->rkb_outbufs.rkbq_bufs,
- rkbuf, rkbuf_link);
- } else {
- /* Insert message at tail of queue */
- TAILQ_INSERT_TAIL(&rkb->rkb_outbufs.rkbq_bufs,
- rkbuf, rkbuf_link);
- }
-
- (void)rd_atomic32_add(&rkb->rkb_outbufs.rkbq_cnt, 1);
- (void)rd_atomic32_add(&rkb->rkb_outbufs.rkbq_msg_cnt,
- rd_atomic32_get(&rkbuf->rkbuf_msgq.rkmq_msg_cnt));
-}
-
-
-/**
- * Finalize a stuffed rkbuf for sending to broker.
- */
-static void rd_kafka_buf_finalize (rd_kafka_t *rk, rd_kafka_buf_t *rkbuf) {
- size_t totsize;
-
- /* Calculate total request buffer length. */
- totsize = rd_buf_len(&rkbuf->rkbuf_buf) - 4;
- rd_assert(totsize <= (size_t)rk->rk_conf.max_msg_size);
-
- /* Set up a buffer reader for sending the buffer. */
- rd_slice_init_full(&rkbuf->rkbuf_reader, &rkbuf->rkbuf_buf);
-
- /**
- * Update request header fields
- */
- /* Total reuqest length */
- rd_kafka_buf_update_i32(rkbuf, 0, (int32_t)totsize);
-
- /* ApiVersion */
- rd_kafka_buf_update_i16(rkbuf, 4+2, rkbuf->rkbuf_reqhdr.ApiVersion);
-}
-
-
-void rd_kafka_broker_buf_enq1 (rd_kafka_broker_t *rkb,
- rd_kafka_buf_t *rkbuf,
- rd_kafka_resp_cb_t *resp_cb,
- void *opaque) {
-
-
- rkbuf->rkbuf_cb = resp_cb;
- rkbuf->rkbuf_opaque = opaque;
-
- rd_kafka_buf_finalize(rkb->rkb_rk, rkbuf);
-
- rd_kafka_broker_buf_enq0(rkb, rkbuf,
- (rkbuf->rkbuf_flags & RD_KAFKA_OP_F_FLASH)?
- 1/*head*/: 0/*tail*/);
-}
-
-
-/**
- * Enqueue buffer on broker's xmit queue, but fail buffer immediately
- * if broker is not up.
- *
- * Locality: broker thread
- */
-static int rd_kafka_broker_buf_enq2 (rd_kafka_broker_t *rkb,
- rd_kafka_buf_t *rkbuf) {
- if (unlikely(rkb->rkb_source == RD_KAFKA_INTERNAL)) {
- /* Fail request immediately if this is the internal broker. */
- rd_kafka_buf_callback(rkb->rkb_rk, rkb,
- RD_KAFKA_RESP_ERR__TRANSPORT,
- NULL, rkbuf);
- return -1;
- }
-
- rd_kafka_broker_buf_enq0(rkb, rkbuf,
- (rkbuf->rkbuf_flags & RD_KAFKA_OP_F_FLASH)?
- 1/*head*/: 0/*tail*/);
-
- return 0;
-}
-
-
-
-/**
- * Enqueue buffer for tranmission.
- * Responses are enqueued on 'replyq' (RD_KAFKA_OP_RECV_BUF)
- *
- * Locality: any thread
- */
-void rd_kafka_broker_buf_enq_replyq (rd_kafka_broker_t *rkb,
- rd_kafka_buf_t *rkbuf,
- rd_kafka_replyq_t replyq,
- rd_kafka_resp_cb_t *resp_cb,
- void *opaque) {
-
- assert(rkbuf->rkbuf_rkb == rkb);
- if (resp_cb) {
- rkbuf->rkbuf_replyq = replyq;
- rkbuf->rkbuf_cb = resp_cb;
- rkbuf->rkbuf_opaque = opaque;
- } else {
- rd_dassert(!replyq.q);
- }
-
- rd_kafka_buf_finalize(rkb->rkb_rk, rkbuf);
-
-
- if (thrd_is_current(rkb->rkb_thread)) {
- rd_kafka_broker_buf_enq2(rkb, rkbuf);
-
- } else {
- rd_kafka_op_t *rko = rd_kafka_op_new(RD_KAFKA_OP_XMIT_BUF);
- rko->rko_u.xbuf.rkbuf = rkbuf;
- rd_kafka_q_enq(rkb->rkb_ops, rko);
- }
-}
-
-
-
-
-/**
- * @returns the current broker state change version.
- * Pass this value to fugure rd_kafka_brokers_wait_state_change() calls
- * to avoid the race condition where a state-change happens between
- * an initial call to some API that fails and the sub-sequent
- * .._wait_state_change() call.
- */
-int rd_kafka_brokers_get_state_version (rd_kafka_t *rk) {
- int version;
- mtx_lock(&rk->rk_broker_state_change_lock);
- version = rk->rk_broker_state_change_version;
- mtx_unlock(&rk->rk_broker_state_change_lock);
- return version;
-}
-
-/**
- * @brief Wait at most \p timeout_ms for any state change for any broker.
- * \p stored_version is the value previously returned by
- * rd_kafka_brokers_get_state_version() prior to another API call
- * that failed due to invalid state.
- *
- * Triggers:
- * - broker state changes
- * - broker transitioning from blocking to non-blocking
- * - partition leader changes
- * - group state changes
- *
- * @remark There is no guarantee that a state change actually took place.
- *
- * @returns 1 if a state change was signaled (maybe), else 0 (timeout)
- *
- * @locality any thread
- */
-int rd_kafka_brokers_wait_state_change (rd_kafka_t *rk, int stored_version,
- int timeout_ms) {
- int r;
- mtx_lock(&rk->rk_broker_state_change_lock);
- if (stored_version != rk->rk_broker_state_change_version)
- r = 1;
- else
- r = cnd_timedwait_ms(&rk->rk_broker_state_change_cnd,
- &rk->rk_broker_state_change_lock,
- timeout_ms) == thrd_success;
- mtx_unlock(&rk->rk_broker_state_change_lock);
- return r;
-}
-
-
-/**
- * @brief Broadcast broker state change to listeners, if any.
- *
- * @locality any thread
- */
-void rd_kafka_brokers_broadcast_state_change (rd_kafka_t *rk) {
- rd_kafka_dbg(rk, GENERIC, "BROADCAST",
- "Broadcasting state change");
- mtx_lock(&rk->rk_broker_state_change_lock);
- rk->rk_broker_state_change_version++;
- cnd_broadcast(&rk->rk_broker_state_change_cnd);
- mtx_unlock(&rk->rk_broker_state_change_lock);
-}
-
-
-/**
- * Returns a random broker (with refcnt increased) in state 'state'.
- * Uses Reservoir sampling.
- *
- * 'filter' is an optional callback used to filter out undesired brokers.
- * The filter function should return 1 to filter out a broker, or 0 to keep it
- * in the list of eligible brokers to return.
- * rd_kafka_broker_lock() is held during the filter callback.
- *
- * Locks: rd_kafka_rdlock(rk) MUST be held.
- * Locality: any thread
- */
-rd_kafka_broker_t *rd_kafka_broker_any (rd_kafka_t *rk, int state,
- int (*filter) (rd_kafka_broker_t *rkb,
- void *opaque),
- void *opaque) {
- rd_kafka_broker_t *rkb, *good = NULL;
- int cnt = 0;
-
- TAILQ_FOREACH(rkb, &rk->rk_brokers, rkb_link) {
- rd_kafka_broker_lock(rkb);
- if ((int)rkb->rkb_state == state &&
- (!filter || !filter(rkb, opaque))) {
- if (cnt < 1 || rd_jitter(0, cnt) < 1) {
- if (good)
- rd_kafka_broker_destroy(good);
- rd_kafka_broker_keep(rkb);
- good = rkb;
- }
- cnt += 1;
- }
- rd_kafka_broker_unlock(rkb);
- }
-
- return good;
-}
-
-
-/**
- * @brief Spend at most \p timeout_ms to acquire a usable (Up && non-blocking)
- * broker.
- *
- * @returns A probably usable broker with increased refcount, or NULL on timeout
- * @locks rd_kafka_*lock() if !do_lock
- * @locality any
- */
-rd_kafka_broker_t *rd_kafka_broker_any_usable (rd_kafka_t *rk,
- int timeout_ms,
- int do_lock) {
- const rd_ts_t ts_end = rd_timeout_init(timeout_ms);
-
- while (1) {
- rd_kafka_broker_t *rkb;
- int remains;
- int version = rd_kafka_brokers_get_state_version(rk);
-
- /* Try non-blocking (e.g., non-fetching) brokers first. */
- if (do_lock)
- rd_kafka_rdlock(rk);
- rkb = rd_kafka_broker_any(rk, RD_KAFKA_BROKER_STATE_UP,
- rd_kafka_broker_filter_non_blocking,
- NULL);
- if (!rkb)
- rkb = rd_kafka_broker_any(rk, RD_KAFKA_BROKER_STATE_UP,
- NULL, NULL);
- if (do_lock)
- rd_kafka_rdunlock(rk);
-
- if (rkb)
- return rkb;
-
- remains = rd_timeout_remains(ts_end);
- if (rd_timeout_expired(remains))
- return NULL;
-
- rd_kafka_brokers_wait_state_change(rk, version, remains);
- }
-
- return NULL;
-}
-
-
-
-/**
- * Returns a broker in state `state`, preferring the one with
- * matching `broker_id`.
- * Uses Reservoir sampling.
- *
- * Locks: rd_kafka_rdlock(rk) MUST be held.
- * Locality: any thread
- */
-rd_kafka_broker_t *rd_kafka_broker_prefer (rd_kafka_t *rk, int32_t broker_id,
- int state) {
- rd_kafka_broker_t *rkb, *good = NULL;
- int cnt = 0;
-
- TAILQ_FOREACH(rkb, &rk->rk_brokers, rkb_link) {
- rd_kafka_broker_lock(rkb);
- if ((int)rkb->rkb_state == state) {
- if (broker_id != -1 && rkb->rkb_nodeid == broker_id) {
- if (good)
- rd_kafka_broker_destroy(good);
- rd_kafka_broker_keep(rkb);
- good = rkb;
- rd_kafka_broker_unlock(rkb);
- break;
- }
- if (cnt < 1 || rd_jitter(0, cnt) < 1) {
- if (good)
- rd_kafka_broker_destroy(good);
- rd_kafka_broker_keep(rkb);
- good = rkb;
- }
- cnt += 1;
- }
- rd_kafka_broker_unlock(rkb);
- }
-
- return good;
-}
-
-
-
-
-
-
-/**
- * Find a waitresp (rkbuf awaiting response) by the correlation id.
- */
-static rd_kafka_buf_t *rd_kafka_waitresp_find (rd_kafka_broker_t *rkb,
- int32_t corrid) {
- rd_kafka_buf_t *rkbuf;
- rd_ts_t now = rd_clock();
-
- rd_kafka_assert(rkb->rkb_rk, thrd_is_current(rkb->rkb_thread));
-
- TAILQ_FOREACH(rkbuf, &rkb->rkb_waitresps.rkbq_bufs, rkbuf_link)
- if (rkbuf->rkbuf_corrid == corrid) {
- /* Convert ts_sent to RTT */
- rkbuf->rkbuf_ts_sent = now - rkbuf->rkbuf_ts_sent;
- rd_avg_add(&rkb->rkb_avg_rtt, rkbuf->rkbuf_ts_sent);
-
- if (rkbuf->rkbuf_flags & RD_KAFKA_OP_F_BLOCKING &&
- rd_atomic32_sub(&rkb->rkb_blocking_request_cnt,
- 1) == 1)
- rd_kafka_brokers_broadcast_state_change(
- rkb->rkb_rk);
-
- rd_kafka_bufq_deq(&rkb->rkb_waitresps, rkbuf);
- return rkbuf;
- }
- return NULL;
-}
-
-
-
-
-/**
- * Map a response message to a request.
- */
-static int rd_kafka_req_response (rd_kafka_broker_t *rkb,
- rd_kafka_buf_t *rkbuf) {
- rd_kafka_buf_t *req;
-
- rd_kafka_assert(rkb->rkb_rk, thrd_is_current(rkb->rkb_thread));
-
-
- /* Find corresponding request message by correlation id */
- if (unlikely(!(req =
- rd_kafka_waitresp_find(rkb,
- rkbuf->rkbuf_reshdr.CorrId)))) {
- /* unknown response. probably due to request timeout */
- rd_atomic64_add(&rkb->rkb_c.rx_corrid_err, 1);
- rd_rkb_dbg(rkb, BROKER, "RESPONSE",
- "Response for unknown CorrId %"PRId32" (timed out?)",
- rkbuf->rkbuf_reshdr.CorrId);
- rd_kafka_buf_destroy(rkbuf);
- return -1;
- }
-
- rd_rkb_dbg(rkb, PROTOCOL, "RECV",
- "Received %sResponse (v%hd, %"PRIusz" bytes, CorrId %"PRId32
- ", rtt %.2fms)",
- rd_kafka_ApiKey2str(req->rkbuf_reqhdr.ApiKey),
- req->rkbuf_reqhdr.ApiVersion,
- rkbuf->rkbuf_totlen, rkbuf->rkbuf_reshdr.CorrId,
- (float)req->rkbuf_ts_sent / 1000.0f);
-
- /* Set up response reader slice starting past the response header */
- rd_slice_init(&rkbuf->rkbuf_reader, &rkbuf->rkbuf_buf,
- RD_KAFKAP_RESHDR_SIZE,
- rd_buf_len(&rkbuf->rkbuf_buf) - RD_KAFKAP_RESHDR_SIZE);
-
- if (!rkbuf->rkbuf_rkb) {
- rkbuf->rkbuf_rkb = rkb;
- rd_kafka_broker_keep(rkbuf->rkbuf_rkb);
- } else
- rd_assert(rkbuf->rkbuf_rkb == rkb);
-
- /* Call callback. */
- rd_kafka_buf_callback(rkb->rkb_rk, rkb, 0, rkbuf, req);
-
- return 0;
-}
-
-
-
-
-int rd_kafka_recv (rd_kafka_broker_t *rkb) {
- rd_kafka_buf_t *rkbuf;
- ssize_t r;
- /* errstr is not set by buf_read errors, so default it here. */
- char errstr[512] = "Protocol parse failure";
- rd_kafka_resp_err_t err = RD_KAFKA_RESP_ERR_NO_ERROR;
- const int log_decode_errors = LOG_ERR;
-
-
- /* It is impossible to estimate the correct size of the response
- * so we split the read up in two parts: first we read the protocol
- * length and correlation id (i.e., the Response header), and then
- * when we know the full length of the response we allocate a new
- * buffer and call receive again.
- * All this in an async fashion (e.g., partial reads).
- */
- if (!(rkbuf = rkb->rkb_recv_buf)) {
- /* No receive in progress: create new buffer */
-
- rkbuf = rd_kafka_buf_new(2, RD_KAFKAP_RESHDR_SIZE);
-
- rkb->rkb_recv_buf = rkbuf;
-
- /* Set up buffer reader for the response header. */
- rd_buf_write_ensure(&rkbuf->rkbuf_buf,
- RD_KAFKAP_RESHDR_SIZE,
- RD_KAFKAP_RESHDR_SIZE);
- }
-
- rd_dassert(rd_buf_write_remains(&rkbuf->rkbuf_buf) > 0);
-
- r = rd_kafka_transport_recv(rkb->rkb_transport, &rkbuf->rkbuf_buf,
- errstr, sizeof(errstr));
- if (unlikely(r <= 0)) {
- if (r == 0)
- return 0; /* EAGAIN */
- err = RD_KAFKA_RESP_ERR__TRANSPORT;
- rd_atomic64_add(&rkb->rkb_c.rx_err, 1);
- goto err;
- }
-
- if (rkbuf->rkbuf_totlen == 0) {
- /* Packet length not known yet. */
-
- if (unlikely(rd_buf_write_pos(&rkbuf->rkbuf_buf) <
- RD_KAFKAP_RESHDR_SIZE)) {
- /* Need response header for packet length and corrid.
- * Wait for more data. */
- return 0;
- }
-
- rd_assert(!rkbuf->rkbuf_rkb);
- rkbuf->rkbuf_rkb = rkb; /* Protocol parsing code needs
- * the rkb for logging, but we dont
- * want to keep a reference to the
- * broker this early since that extra
- * refcount will mess with the broker's
- * refcount-based termination code. */
-
- /* Initialize reader */
- rd_slice_init(&rkbuf->rkbuf_reader, &rkbuf->rkbuf_buf, 0,
- RD_KAFKAP_RESHDR_SIZE);
-
- /* Read protocol header */
- rd_kafka_buf_read_i32(rkbuf, &rkbuf->rkbuf_reshdr.Size);
- rd_kafka_buf_read_i32(rkbuf, &rkbuf->rkbuf_reshdr.CorrId);
-
- rkbuf->rkbuf_rkb = NULL; /* Reset */
-
- rkbuf->rkbuf_totlen = rkbuf->rkbuf_reshdr.Size;
-
- /* Make sure message size is within tolerable limits. */
- if (rkbuf->rkbuf_totlen < 4/*CorrId*/ ||
- rkbuf->rkbuf_totlen >
- (size_t)rkb->rkb_rk->rk_conf.recv_max_msg_size) {
- rd_snprintf(errstr, sizeof(errstr),
- "Invalid response size %"PRId32" (0..%i): "
- "increase receive.message.max.bytes",
- rkbuf->rkbuf_reshdr.Size,
- rkb->rkb_rk->rk_conf.recv_max_msg_size);
- err = RD_KAFKA_RESP_ERR__BAD_MSG;
- rd_atomic64_add(&rkb->rkb_c.rx_err, 1);
- goto err;
- }
-
- rkbuf->rkbuf_totlen -= 4; /*CorrId*/
-
- if (rkbuf->rkbuf_totlen > 0) {
- /* Allocate another buffer that fits all data (short of
- * the common response header). We want all
- * data to be in contigious memory. */
-
- rd_buf_write_ensure_contig(&rkbuf->rkbuf_buf,
- rkbuf->rkbuf_totlen);
- }
- }
-
- if (rd_buf_write_pos(&rkbuf->rkbuf_buf) - RD_KAFKAP_RESHDR_SIZE ==
- rkbuf->rkbuf_totlen) {
- /* Message is complete, pass it on to the original requester. */
- rkb->rkb_recv_buf = NULL;
- rd_atomic64_add(&rkb->rkb_c.rx, 1);
- rd_atomic64_add(&rkb->rkb_c.rx_bytes,
- rd_buf_write_pos(&rkbuf->rkbuf_buf));
- rd_kafka_req_response(rkb, rkbuf);
- }
-
- return 1;
-
- err_parse:
- err = rkbuf->rkbuf_err;
- err:
- rd_kafka_broker_fail(rkb,
- !rkb->rkb_rk->rk_conf.log_connection_close &&
- !strcmp(errstr, "Disconnected") ?
- LOG_DEBUG : LOG_ERR, err,
- "Receive failed: %s", errstr);
- return -1;
-}
-
-
-/**
- * Linux version of socket_cb providing racefree CLOEXEC.
- */
-int rd_kafka_socket_cb_linux (int domain, int type, int protocol,
- void *opaque) {
-#ifdef SOCK_CLOEXEC
- return socket(domain, type | SOCK_CLOEXEC, protocol);
-#else
- return rd_kafka_socket_cb_generic(domain, type, protocol, opaque);
-#endif
-}
-
-/**
- * Fallback version of socket_cb NOT providing racefree CLOEXEC,
- * but setting CLOEXEC after socket creation (if FD_CLOEXEC is defined).
- */
-int rd_kafka_socket_cb_generic (int domain, int type, int protocol,
- void *opaque) {
- int s;
- int on = 1;
- s = (int)socket(domain, type, protocol);
- if (s == -1)
- return -1;
-#ifdef FD_CLOEXEC
- fcntl(s, F_SETFD, FD_CLOEXEC, &on);
-#endif
- return s;
-}
-
-
-/**
- * Initiate asynchronous connection attempt to the next address
- * in the broker's address list.
- * While the connect is asynchronous and its IO served in the CONNECT state,
- * the initial name resolve is blocking.
- *
- * Returns -1 on error, else 0.
- */
-static int rd_kafka_broker_connect (rd_kafka_broker_t *rkb) {
- const rd_sockaddr_inx_t *sinx;
- char errstr[512];
-
- rd_rkb_dbg(rkb, BROKER, "CONNECT",
- "broker in state %s connecting",
- rd_kafka_broker_state_names[rkb->rkb_state]);
-
- if (rd_kafka_broker_resolve(rkb) == -1)
- return -1;
-
- sinx = rd_sockaddr_list_next(rkb->rkb_rsal);
-
- rd_kafka_assert(rkb->rkb_rk, !rkb->rkb_transport);
-
- if (!(rkb->rkb_transport = rd_kafka_transport_connect(rkb, sinx,
- errstr, sizeof(errstr)))) {
- /* Avoid duplicate log messages */
- if (rkb->rkb_err.err == errno)
- rd_kafka_broker_fail(rkb, LOG_DEBUG,
- RD_KAFKA_RESP_ERR__FAIL, NULL);
- else
- rd_kafka_broker_fail(rkb, LOG_ERR,
- RD_KAFKA_RESP_ERR__TRANSPORT,
- "%s", errstr);
- return -1;
- }
-
- rd_kafka_broker_lock(rkb);
- rd_kafka_broker_set_state(rkb, RD_KAFKA_BROKER_STATE_CONNECT);
- rd_kafka_broker_unlock(rkb);
-
- return 0;
-}
-
-
-/**
- * @brief Call when connection is ready to transition to fully functional
- * UP state.
- *
- * @locality Broker thread
- */
-void rd_kafka_broker_connect_up (rd_kafka_broker_t *rkb) {
-
- rkb->rkb_max_inflight = rkb->rkb_rk->rk_conf.max_inflight;
- rkb->rkb_err.err = 0;
-
- rd_kafka_broker_lock(rkb);
- rd_kafka_broker_set_state(rkb, RD_KAFKA_BROKER_STATE_UP);
- rd_kafka_broker_unlock(rkb);
-
- /* Request metadata (async):
- * try locally known topics first and if there are none try
- * getting just the broker list. */
- if (rd_kafka_metadata_refresh_known_topics(NULL, rkb, 0/*dont force*/,
- "connected") ==
- RD_KAFKA_RESP_ERR__UNKNOWN_TOPIC)
- rd_kafka_metadata_refresh_brokers(NULL, rkb, "connected");
-}
-
-
-
-static void rd_kafka_broker_connect_auth (rd_kafka_broker_t *rkb);
-
-
-/**
- * @brief Parses and handles SaslMechanism response, transitions
- * the broker state.
- *
- */
-static void
-rd_kafka_broker_handle_SaslHandshake (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;
- int32_t MechCnt;
- int16_t ErrorCode;
- int i = 0;
- char *mechs = "(n/a)";
- size_t msz, mof = 0;
-
- if (err == RD_KAFKA_RESP_ERR__DESTROY)
- return;
-
- if (err)
- goto err;
-
- rd_kafka_buf_read_i16(rkbuf, &ErrorCode);
- rd_kafka_buf_read_i32(rkbuf, &MechCnt);
-
- /* Build a CSV string of supported mechanisms. */
- msz = RD_MIN(511, MechCnt * 32);
- mechs = rd_alloca(msz);
- *mechs = '\0';
-
- for (i = 0 ; i < MechCnt ; i++) {
- rd_kafkap_str_t mech;
- rd_kafka_buf_read_str(rkbuf, &mech);
-
- mof += rd_snprintf(mechs+mof, msz-mof, "%s%.*s",
- i ? ",":"", RD_KAFKAP_STR_PR(&mech));
-
- if (mof >= msz)
- break;
- }
-
- rd_rkb_dbg(rkb,
- PROTOCOL | RD_KAFKA_DBG_SECURITY | RD_KAFKA_DBG_BROKER,
- "SASLMECHS", "Broker supported SASL mechanisms: %s",
- mechs);
-
- if (ErrorCode) {
- err = ErrorCode;
- goto err;
- }
-
- /* Circle back to connect_auth() to start proper AUTH state. */
- rd_kafka_broker_connect_auth(rkb);
- return;
-
- err_parse:
- err = rkbuf->rkbuf_err;
- err:
- rd_kafka_broker_fail(rkb, LOG_ERR,
- RD_KAFKA_RESP_ERR__AUTHENTICATION,
- "SASL %s mechanism handshake failed: %s: "
- "broker's supported mechanisms: %s",
- rkb->rkb_rk->rk_conf.sasl.mechanisms,
- rd_kafka_err2str(err), mechs);
-}
-
-
-/**
- * @brief Transition state to:
- * - AUTH_HANDSHAKE (if SASL is configured and handshakes supported)
- * - AUTH (if SASL is configured but no handshake is required or
- * not supported, or has already taken place.)
- * - UP (if SASL is not configured)
- */
-static void rd_kafka_broker_connect_auth (rd_kafka_broker_t *rkb) {
-
- if ((rkb->rkb_proto == RD_KAFKA_PROTO_SASL_PLAINTEXT ||
- rkb->rkb_proto == RD_KAFKA_PROTO_SASL_SSL)) {
-
- rd_rkb_dbg(rkb, SECURITY | RD_KAFKA_DBG_BROKER, "AUTH",
- "Auth in state %s (handshake %ssupported)",
- rd_kafka_broker_state_names[rkb->rkb_state],
- (rkb->rkb_features&RD_KAFKA_FEATURE_SASL_HANDSHAKE)
- ? "" : "not ");
-
- /* Broker >= 0.10.0: send request to select mechanism */
- if (rkb->rkb_state != RD_KAFKA_BROKER_STATE_AUTH_HANDSHAKE &&
- (rkb->rkb_features & RD_KAFKA_FEATURE_SASL_HANDSHAKE)) {
-
- rd_kafka_broker_lock(rkb);
- rd_kafka_broker_set_state(
- rkb, RD_KAFKA_BROKER_STATE_AUTH_HANDSHAKE);
- rd_kafka_broker_unlock(rkb);
-
- rd_kafka_SaslHandshakeRequest(
- rkb, rkb->rkb_rk->rk_conf.sasl.mechanisms,
- RD_KAFKA_NO_REPLYQ,
- rd_kafka_broker_handle_SaslHandshake,
- NULL, 1 /* flash */);
-
- } else {
- /* Either Handshake succeeded (protocol selected)
- * or Handshakes were not supported.
- * In both cases continue with authentication. */
- char sasl_errstr[512];
-
- rd_kafka_broker_lock(rkb);
- rd_kafka_broker_set_state(rkb,
- RD_KAFKA_BROKER_STATE_AUTH);
- rd_kafka_broker_unlock(rkb);
-
- if (rd_kafka_sasl_client_new(
- rkb->rkb_transport, sasl_errstr,
- sizeof(sasl_errstr)) == -1) {
- errno = EINVAL;
- rd_kafka_broker_fail(
- rkb, LOG_ERR,
- RD_KAFKA_RESP_ERR__AUTHENTICATION,
- "Failed to initialize "
- "SASL authentication: %s",
- sasl_errstr);
- return;
- }
-
- /* Enter non-Kafka-protocol-framed SASL communication
- * state handled in rdkafka_sasl.c */
- rd_kafka_broker_lock(rkb);
- rd_kafka_broker_set_state(rkb,
- RD_KAFKA_BROKER_STATE_AUTH);
- rd_kafka_broker_unlock(rkb);
- }
-
- return;
- }
-
- /* No authentication required. */
- rd_kafka_broker_connect_up(rkb);
-}
-
-
-/**
- * @brief Specify API versions to use for this connection.
- *
- * @param apis is an allocated list of supported partitions.
- * If NULL the default set will be used based on the
- * \p broker.version.fallback property.
- * @param api_cnt number of elements in \p apis
- *
- * @remark \p rkb takes ownership of \p apis.
- *
- * @locality Broker thread
- * @locks none
- */
-static void rd_kafka_broker_set_api_versions (rd_kafka_broker_t *rkb,
- struct rd_kafka_ApiVersion *apis,
- size_t api_cnt) {
-
- rd_kafka_broker_lock(rkb);
-
- if (rkb->rkb_ApiVersions)
- rd_free(rkb->rkb_ApiVersions);
-
-
- if (!apis) {
- rd_rkb_dbg(rkb, PROTOCOL | RD_KAFKA_DBG_BROKER, "APIVERSION",
- "Using (configuration fallback) %s protocol features",
- rkb->rkb_rk->rk_conf.broker_version_fallback);
-
-
- rd_kafka_get_legacy_ApiVersions(rkb->rkb_rk->rk_conf.
- broker_version_fallback,
- &apis, &api_cnt,
- rkb->rkb_rk->rk_conf.
- broker_version_fallback);
-
- /* Make a copy to store on broker. */
- rd_kafka_ApiVersions_copy(apis, api_cnt, &apis, &api_cnt);
- }
-
- rkb->rkb_ApiVersions = apis;
- rkb->rkb_ApiVersions_cnt = api_cnt;
-
- /* Update feature set based on supported broker APIs. */
- rd_kafka_broker_features_set(rkb,
- rd_kafka_features_check(rkb, apis, api_cnt));
-
- rd_kafka_broker_unlock(rkb);
-}
-
-
-/**
- * Handler for ApiVersion response.
- */
-static void
-rd_kafka_broker_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, void *opaque) {
- struct rd_kafka_ApiVersion *apis;
- size_t api_cnt;
-
- if (err == RD_KAFKA_RESP_ERR__DESTROY)
- return;
-
- err = rd_kafka_handle_ApiVersion(rk, rkb, err, rkbuf, request,
- &apis, &api_cnt);
-
- if (err) {
- rd_kafka_broker_fail(rkb, LOG_DEBUG,
- RD_KAFKA_RESP_ERR__NOT_IMPLEMENTED,
- "ApiVersionRequest failed: %s: "
- "probably due to old broker version",
- rd_kafka_err2str(err));
- return;
- }
-
- rd_kafka_broker_set_api_versions(rkb, apis, api_cnt);
-
- rd_kafka_broker_connect_auth(rkb);
-}
-
-
-/**
- * Call when asynchronous connection attempt completes, either succesfully
- * (if errstr is NULL) or fails.
- *
- * Locality: broker thread
- */
-void rd_kafka_broker_connect_done (rd_kafka_broker_t *rkb, const char *errstr) {
-
- if (errstr) {
- /* Connect failed */
- rd_kafka_broker_fail(rkb,
- errno != 0 && rkb->rkb_err.err == errno ?
- LOG_DEBUG : LOG_ERR,
- RD_KAFKA_RESP_ERR__TRANSPORT,
- "%s", errstr);
- return;
- }
-
- /* Connect succeeded */
- rkb->rkb_connid++;
- rd_rkb_dbg(rkb, BROKER | RD_KAFKA_DBG_PROTOCOL,
- "CONNECTED", "Connected (#%d)", rkb->rkb_connid);
- rkb->rkb_err.err = 0;
- rkb->rkb_max_inflight = 1; /* Hold back other requests until
- * ApiVersion, SaslHandshake, etc
- * are done. */
-
- rd_kafka_transport_poll_set(rkb->rkb_transport, POLLIN);
-
- if (rkb->rkb_rk->rk_conf.api_version_request &&
- rd_interval_immediate(&rkb->rkb_ApiVersion_fail_intvl, 0, 0) > 0) {
- /* Use ApiVersion to query broker for supported API versions. */
- rd_kafka_broker_feature_enable(rkb, RD_KAFKA_FEATURE_APIVERSION);
- }
-
-
- if (rkb->rkb_features & RD_KAFKA_FEATURE_APIVERSION) {
- /* Query broker for supported API versions.
- * This may fail with a disconnect on non-supporting brokers
- * so hold off any other requests until we get a response,
- * and if the connection is torn down we disable this feature. */
- rd_kafka_broker_lock(rkb);
- rd_kafka_broker_set_state(rkb,RD_KAFKA_BROKER_STATE_APIVERSION_QUERY);
- rd_kafka_broker_unlock(rkb);
-
- rd_kafka_ApiVersionRequest(
- rkb, RD_KAFKA_NO_REPLYQ,
- rd_kafka_broker_handle_ApiVersion, NULL,
- 1 /*Flash message: prepend to transmit queue*/);
- } else {
-
- /* Use configured broker.version.fallback to
- * figure out API versions */
- rd_kafka_broker_set_api_versions(rkb, NULL, 0);
-
- /* Authenticate if necessary */
- rd_kafka_broker_connect_auth(rkb);
- }
-
-}
-
-
-
-/**
- * @brief Checks if the given API request+version is supported by the broker.
- * @returns 1 if supported, else 0.
- * @locality broker thread
- * @locks none
- */
-static RD_INLINE int
-rd_kafka_broker_request_supported (rd_kafka_broker_t *rkb,
- rd_kafka_buf_t *rkbuf) {
- struct rd_kafka_ApiVersion skel = {
- .ApiKey = rkbuf->rkbuf_reqhdr.ApiKey
- };
- struct rd_kafka_ApiVersion *ret;
-
- if (unlikely(rkbuf->rkbuf_reqhdr.ApiKey == RD_KAFKAP_ApiVersion))
- return 1; /* ApiVersion requests are used to detect
- * the supported API versions, so should always
- * be allowed through. */
-
- /* First try feature flags, if any, which may cover a larger
- * set of APIs. */
- if (rkbuf->rkbuf_features)
- return (rkb->rkb_features & rkbuf->rkbuf_features) ==
- rkbuf->rkbuf_features;
-
- /* Then try the ApiVersion map. */
- ret = bsearch(&skel, rkb->rkb_ApiVersions, rkb->rkb_ApiVersions_cnt,
- sizeof(*rkb->rkb_ApiVersions),
- rd_kafka_ApiVersion_key_cmp);
- if (!ret)
- return 0;
-
- return ret->MinVer <= rkbuf->rkbuf_reqhdr.ApiVersion &&
- rkbuf->rkbuf_reqhdr.ApiVersion <= ret->MaxVer;
-}
-
-
-/**
- * Send queued messages to broker
- *
- * Locality: io thread
- */
-int rd_kafka_send (rd_kafka_broker_t *rkb) {
- rd_kafka_buf_t *rkbuf;
- unsigned int cnt = 0;
-
- rd_kafka_assert(rkb->rkb_rk, thrd_is_current(rkb->rkb_thread));
-
- while (rkb->rkb_state >= RD_KAFKA_BROKER_STATE_UP &&
- rd_kafka_bufq_cnt(&rkb->rkb_waitresps) < rkb->rkb_max_inflight &&
- (rkbuf = TAILQ_FIRST(&rkb->rkb_outbufs.rkbq_bufs))) {
- ssize_t r;
- size_t pre_of = rd_slice_offset(&rkbuf->rkbuf_reader);
-
- /* Check for broker support */
- if (unlikely(!rd_kafka_broker_request_supported(rkb, rkbuf))) {
- rd_kafka_bufq_deq(&rkb->rkb_outbufs, rkbuf);
- rd_rkb_dbg(rkb, BROKER | RD_KAFKA_DBG_PROTOCOL,
- "UNSUPPORTED",
- "Failing %sResponse "
- "(v%hd, %"PRIusz" bytes, CorrId %"PRId32"): "
- "request not supported by broker "
- "(missing api.version.request or "
- "incorrect broker.version.fallback config?)",
- rd_kafka_ApiKey2str(rkbuf->rkbuf_reqhdr.
- ApiKey),
- rkbuf->rkbuf_reqhdr.ApiVersion,
- rkbuf->rkbuf_totlen,
- rkbuf->rkbuf_reshdr.CorrId);
- rd_kafka_buf_callback(
- rkb->rkb_rk, rkb,
- RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE,
- NULL, rkbuf);
- continue;
- }
-
- /* Set CorrId header field, unless this is the latter part
- * of a partial send in which case the corrid has already
- * been set.
- * Due to how SSL_write() will accept a buffer but still
- * return 0 in some cases we can't rely on the buffer offset
- * but need to use corrid to check this. SSL_write() expects
- * us to send the same buffer again when 0 is returned.
- */
- if (rkbuf->rkbuf_corrid == 0 ||
- rkbuf->rkbuf_connid != rkb->rkb_connid) {
- rd_assert(rd_slice_offset(&rkbuf->rkbuf_reader) == 0);
- rkbuf->rkbuf_corrid = ++rkb->rkb_corrid;
- rd_kafka_buf_update_i32(rkbuf, 4+2+2,
- rkbuf->rkbuf_corrid);
- rkbuf->rkbuf_connid = rkb->rkb_connid;
- } else if (pre_of > RD_KAFKAP_REQHDR_SIZE) {
- rd_kafka_assert(NULL,
- rkbuf->rkbuf_connid == rkb->rkb_connid);
- }
-
- if (0) {
- rd_rkb_dbg(rkb, PROTOCOL, "SEND",
- "Send %s corrid %"PRId32" at "
- "offset %"PRIusz"/%"PRIusz,
- rd_kafka_ApiKey2str(rkbuf->rkbuf_reqhdr.
- ApiKey),
- rkbuf->rkbuf_corrid,
- pre_of, rd_slice_size(&rkbuf->rkbuf_reader));
- }
-
- if ((r = rd_kafka_broker_send(rkb, &rkbuf->rkbuf_reader)) == -1)
- return -1;
-
- /* Partial send? Continue next time. */
- if (rd_slice_remains(&rkbuf->rkbuf_reader) > 0) {
- rd_rkb_dbg(rkb, PROTOCOL, "SEND",
- "Sent partial %sRequest "
- "(v%hd, "
- "%"PRIdsz"+%"PRIdsz"/%"PRIusz" bytes, "
- "CorrId %"PRId32")",
- rd_kafka_ApiKey2str(rkbuf->rkbuf_reqhdr.
- ApiKey),
- rkbuf->rkbuf_reqhdr.ApiVersion,
- (ssize_t)pre_of, r,
- rd_slice_size(&rkbuf->rkbuf_reader),
- rkbuf->rkbuf_corrid);
- return 0;
- }
-
- rd_rkb_dbg(rkb, PROTOCOL, "SEND",
- "Sent %sRequest (v%hd, %"PRIusz" bytes @ %"PRIusz", "
- "CorrId %"PRId32")",
- rd_kafka_ApiKey2str(rkbuf->rkbuf_reqhdr.ApiKey),
- rkbuf->rkbuf_reqhdr.ApiVersion,
- rd_slice_size(&rkbuf->rkbuf_reader),
- pre_of, rkbuf->rkbuf_corrid);
-
- /* Entire buffer sent, unlink from outbuf */
- rd_kafka_bufq_deq(&rkb->rkb_outbufs, rkbuf);
-
- /* Store time for RTT calculation */
- rkbuf->rkbuf_ts_sent = rd_clock();
-
- if (rkbuf->rkbuf_flags & RD_KAFKA_OP_F_BLOCKING &&
- rd_atomic32_add(&rkb->rkb_blocking_request_cnt, 1) == 1)
- rd_kafka_brokers_broadcast_state_change(rkb->rkb_rk);
-
- /* Put buffer on response wait list unless we are not
- * expecting a response (required_acks=0). */
- if (!(rkbuf->rkbuf_flags & RD_KAFKA_OP_F_NO_RESPONSE))
- rd_kafka_bufq_enq(&rkb->rkb_waitresps, rkbuf);
- else { /* Call buffer callback for delivery report. */
- rd_kafka_buf_callback(rkb->rkb_rk, rkb, 0, NULL, rkbuf);
- }
-
- cnt++;
- }
-
- return cnt;
-}
-
-
-/**
- * Add 'rkbuf' to broker 'rkb's retry queue.
- */
-void rd_kafka_broker_buf_retry (rd_kafka_broker_t *rkb, rd_kafka_buf_t *rkbuf) {
-
- /* Restore original replyq since replyq.q will have been NULLed
- * by buf_callback()/replyq_enq(). */
- if (!rkbuf->rkbuf_replyq.q && rkbuf->rkbuf_orig_replyq.q) {
- rkbuf->rkbuf_replyq = rkbuf->rkbuf_orig_replyq;
- rd_kafka_replyq_clear(&rkbuf->rkbuf_orig_replyq);
- }
-
- /* If called from another thread than rkb's broker thread
- * enqueue the buffer on the broker's op queue. */
- if (!thrd_is_current(rkb->rkb_thread)) {
- rd_kafka_op_t *rko = rd_kafka_op_new(RD_KAFKA_OP_XMIT_RETRY);
- rko->rko_u.xbuf.rkbuf = rkbuf;
- rd_kafka_q_enq(rkb->rkb_ops, rko);
- return;
- }
-
- rd_rkb_dbg(rkb, PROTOCOL, "RETRY",
- "Retrying %sRequest (v%hd, %"PRIusz" bytes, retry %d/%d)",
- rd_kafka_ApiKey2str(rkbuf->rkbuf_reqhdr.ApiKey),
- rkbuf->rkbuf_reqhdr.ApiVersion,
- rd_slice_size(&rkbuf->rkbuf_reader),
- rkbuf->rkbuf_retries, rkb->rkb_rk->rk_conf.max_retries);
-
- rd_atomic64_add(&rkb->rkb_c.tx_retries, 1);
-
- rkbuf->rkbuf_ts_retry = rd_clock() +
- (rkb->rkb_rk->rk_conf.retry_backoff_ms * 1000);
- /* Reset send offset */
- rd_slice_seek(&rkbuf->rkbuf_reader, 0);
- rkbuf->rkbuf_corrid = 0;
-
- rd_kafka_bufq_enq(&rkb->rkb_retrybufs, rkbuf);
-}
-
-
-/**
- * Move buffers that have expired their retry backoff time from the
- * retry queue to the outbuf.
- */
-static void rd_kafka_broker_retry_bufs_move (rd_kafka_broker_t *rkb) {
- rd_ts_t now = rd_clock();
- rd_kafka_buf_t *rkbuf;
-
- while ((rkbuf = TAILQ_FIRST(&rkb->rkb_retrybufs.rkbq_bufs))) {
- if (rkbuf->rkbuf_ts_retry > now)
- break;
-
- rd_kafka_bufq_deq(&rkb->rkb_retrybufs, rkbuf);
-
- rd_kafka_broker_buf_enq0(rkb, rkbuf, 0/*tail*/);
- }
-}
-
-
-/**
- * Propagate delivery report for entire message queue.
- */
-void rd_kafka_dr_msgq (rd_kafka_itopic_t *rkt,
- rd_kafka_msgq_t *rkmq, rd_kafka_resp_err_t err) {
- rd_kafka_t *rk = rkt->rkt_rk;
-
- if (unlikely(rd_kafka_msgq_len(rkmq) == 0))
- return;
-
- /* Call on_acknowledgement() interceptors */
- rd_kafka_interceptors_on_acknowledgement_queue(rk, rkmq);
-
- if ((rk->rk_conf.enabled_events & RD_KAFKA_EVENT_DR) &&
- (!rk->rk_conf.dr_err_only || err)) {
- /* Pass all messages to application thread in one op. */
- rd_kafka_op_t *rko;
-
- rko = rd_kafka_op_new(RD_KAFKA_OP_DR);
- rko->rko_err = err;
- rko->rko_u.dr.s_rkt = rd_kafka_topic_keep(rkt);
- rd_kafka_msgq_init(&rko->rko_u.dr.msgq);
-
- /* Move all messages to op's msgq */
- rd_kafka_msgq_move(&rko->rko_u.dr.msgq, rkmq);
-
- rd_kafka_q_enq(rk->rk_rep, rko);
-
- } else {
- /* No delivery report callback. */
-
- /* Destroy the messages right away. */
- rd_kafka_msgq_purge(rk, rkmq);
- }
-}
-
-
-
-
-
-
-
-
-
-
-
-/**
- * @brief Map and assign existing partitions to this broker using
- * the leader-id.
- *
- * @locks none
- * @locality any
- */
-static void rd_kafka_broker_map_partitions (rd_kafka_broker_t *rkb) {
- rd_kafka_t *rk = rkb->rkb_rk;
- rd_kafka_itopic_t *rkt;
- int cnt = 0;
-
- if (rkb->rkb_nodeid == -1)
- return;
-
- rd_kafka_rdlock(rk);
- TAILQ_FOREACH(rkt, &rk->rk_topics, rkt_link) {
- int i;
-
- rd_kafka_topic_wrlock(rkt);
- for (i = 0 ; i < rkt->rkt_partition_cnt ; i++) {
- shptr_rd_kafka_toppar_t *s_rktp = rkt->rkt_p[i];
- rd_kafka_toppar_t *rktp = rd_kafka_toppar_s2i(s_rktp);
-
- /* Only map unassigned partitions matching this broker*/
- rd_kafka_toppar_lock(rktp);
- if (rktp->rktp_leader_id == rkb->rkb_nodeid &&
- !(rktp->rktp_leader && rktp->rktp_next_leader)) {
- rd_kafka_toppar_leader_update(
- rktp, rktp->rktp_leader_id, rkb);
- cnt++;
- }
- rd_kafka_toppar_unlock(rktp);
- }
- rd_kafka_topic_wrunlock(rkt);
- }
- rd_kafka_rdunlock(rk);
-
- rd_rkb_dbg(rkb, TOPIC|RD_KAFKA_DBG_BROKER, "LEADER",
- "Mapped %d partition(s) to broker", cnt);
-}
-
-
-/**
- * @brief Broker id comparator
- */
-static int rd_kafka_broker_cmp_by_id (const void *_a, const void *_b) {
- const rd_kafka_broker_t *a = _a, *b = _b;
- return a->rkb_nodeid - b->rkb_nodeid;
-}
-
-
-
-/**
- * @brief Serve a broker op (an op posted by another thread to be handled by
- * this broker's thread).
- *
- * @returns 0 if calling op loop should break out, else 1 to continue.
- * @locality broker thread
- * @locks none
- */
-static int rd_kafka_broker_op_serve (rd_kafka_broker_t *rkb,
- rd_kafka_op_t *rko) {
- shptr_rd_kafka_toppar_t *s_rktp;
- rd_kafka_toppar_t *rktp;
- int ret = 1;
-
- rd_kafka_assert(rkb->rkb_rk, thrd_is_current(rkb->rkb_thread));
-
- switch (rko->rko_type)
- {
- case RD_KAFKA_OP_NODE_UPDATE:
- {
- enum {
- _UPD_NAME = 0x1,
- _UPD_ID = 0x2
- } updated = 0;
- char brokername[RD_KAFKA_NODENAME_SIZE];
-
- /* Need kafka_wrlock for updating rk_broker_by_id */
- rd_kafka_wrlock(rkb->rkb_rk);
- rd_kafka_broker_lock(rkb);
-
- if (strcmp(rkb->rkb_nodename,
- rko->rko_u.node.nodename)) {
- rd_rkb_dbg(rkb, BROKER, "UPDATE",
- "Nodename changed from %s to %s",
- rkb->rkb_nodename,
- rko->rko_u.node.nodename);
- strncpy(rkb->rkb_nodename,
- rko->rko_u.node.nodename,
- sizeof(rkb->rkb_nodename)-1);
- updated |= _UPD_NAME;
- }
-
- if (rko->rko_u.node.nodeid != -1 &&
- rko->rko_u.node.nodeid != rkb->rkb_nodeid) {
- int32_t old_nodeid = rkb->rkb_nodeid;
- rd_rkb_dbg(rkb, BROKER, "UPDATE",
- "NodeId changed from %"PRId32" to %"PRId32,
- rkb->rkb_nodeid,
- rko->rko_u.node.nodeid);
-
- rkb->rkb_nodeid = rko->rko_u.node.nodeid;
-
- /* Update broker_by_id sorted list */
- if (old_nodeid == -1)
- rd_list_add(&rkb->rkb_rk->rk_broker_by_id, rkb);
- rd_list_sort(&rkb->rkb_rk->rk_broker_by_id,
- rd_kafka_broker_cmp_by_id);
-
- updated |= _UPD_ID;
- }
-
- rd_kafka_mk_brokername(brokername, sizeof(brokername),
- rkb->rkb_proto,
- rkb->rkb_nodename, rkb->rkb_nodeid,
- RD_KAFKA_LEARNED);
- if (strcmp(rkb->rkb_name, brokername)) {
- /* Udate the name copy used for logging. */
- mtx_lock(&rkb->rkb_logname_lock);
- rd_free(rkb->rkb_logname);
- rkb->rkb_logname = rd_strdup(brokername);
- mtx_unlock(&rkb->rkb_logname_lock);
-
- rd_rkb_dbg(rkb, BROKER, "UPDATE",
- "Name changed from %s to %s",
- rkb->rkb_name, brokername);
- strncpy(rkb->rkb_name, brokername,
- sizeof(rkb->rkb_name)-1);
- }
- rd_kafka_broker_unlock(rkb);
- rd_kafka_wrunlock(rkb->rkb_rk);
-
- if (updated & _UPD_NAME)
- rd_kafka_broker_fail(rkb, LOG_NOTICE,
- RD_KAFKA_RESP_ERR__NODE_UPDATE,
- "Broker hostname updated");
- else if (updated & _UPD_ID) {
- /* Map existing partitions to this broker. */
- rd_kafka_broker_map_partitions(rkb);
-
- /* If broker is currently in state up we need
- * to trigger a state change so it exits its
- * state&type based .._serve() loop. */
- rd_kafka_broker_lock(rkb);
- if (rkb->rkb_state == RD_KAFKA_BROKER_STATE_UP)
- rd_kafka_broker_set_state(
- rkb, RD_KAFKA_BROKER_STATE_UPDATE);
- rd_kafka_broker_unlock(rkb);
- }
- break;
- }
-
- case RD_KAFKA_OP_XMIT_BUF:
- rd_kafka_broker_buf_enq2(rkb, rko->rko_u.xbuf.rkbuf);
- rko->rko_u.xbuf.rkbuf = NULL; /* buffer now owned by broker */
- if (rko->rko_replyq.q) {
- /* Op will be reused for forwarding response. */
- rko = NULL;
- }
- break;
-
- case RD_KAFKA_OP_XMIT_RETRY:
- rd_kafka_broker_buf_retry(rkb, rko->rko_u.xbuf.rkbuf);
- rko->rko_u.xbuf.rkbuf = NULL;
- break;
-
- case RD_KAFKA_OP_PARTITION_JOIN:
- /*
- * Add partition to broker toppars
- */
- rktp = rd_kafka_toppar_s2i(rko->rko_rktp);
- rd_kafka_toppar_lock(rktp);
-
- /* Abort join if instance is terminating */
- if (rd_kafka_terminating(rkb->rkb_rk) ||
- (rktp->rktp_flags & RD_KAFKA_TOPPAR_F_REMOVE)) {
- rd_rkb_dbg(rkb, BROKER | RD_KAFKA_DBG_TOPIC, "TOPBRK",
- "Topic %s [%"PRId32"]: not joining broker: "
- "%s",
- rktp->rktp_rkt->rkt_topic->str,
- rktp->rktp_partition,
- rd_kafka_terminating(rkb->rkb_rk) ?
- "instance is terminating" :
- "partition removed");
-
- rd_kafka_broker_destroy(rktp->rktp_next_leader);
- rktp->rktp_next_leader = NULL;
- rd_kafka_toppar_unlock(rktp);
- break;
- }
-
- /* See if we are still the next leader */
- if (rktp->rktp_next_leader != rkb) {
- rd_rkb_dbg(rkb, BROKER | RD_KAFKA_DBG_TOPIC, "TOPBRK",
- "Topic %s [%"PRId32"]: not joining broker "
- "(next leader %s)",
- rktp->rktp_rkt->rkt_topic->str,
- rktp->rktp_partition,
- rktp->rktp_next_leader ?
- rd_kafka_broker_name(rktp->rktp_next_leader):
- "(none)");
-
- /* Need temporary refcount so we can safely unlock
- * after q_enq(). */
- s_rktp = rd_kafka_toppar_keep(rktp);
-
- /* No, forward this op to the new next leader. */
- rd_kafka_q_enq(rktp->rktp_next_leader->rkb_ops, rko);
- rko = NULL;
-
- rd_kafka_toppar_unlock(rktp);
- rd_kafka_toppar_destroy(s_rktp);
-
- break;
- }
-
- rd_rkb_dbg(rkb, BROKER | RD_KAFKA_DBG_TOPIC, "TOPBRK",
- "Topic %s [%"PRId32"]: joining broker (rktp %p)",
- rktp->rktp_rkt->rkt_topic->str,
- rktp->rktp_partition, rktp);
-
- rd_kafka_assert(NULL, rktp->rktp_s_for_rkb == NULL);
- rktp->rktp_s_for_rkb = rd_kafka_toppar_keep(rktp);
- rd_kafka_broker_lock(rkb);
- TAILQ_INSERT_TAIL(&rkb->rkb_toppars, rktp, rktp_rkblink);
- rkb->rkb_toppar_cnt++;
- rd_kafka_broker_unlock(rkb);
- rktp->rktp_leader = rkb;
- rktp->rktp_msgq_wakeup_fd = rkb->rkb_toppar_wakeup_fd;
- rd_kafka_broker_keep(rkb);
-
- rd_kafka_broker_destroy(rktp->rktp_next_leader);
- rktp->rktp_next_leader = NULL;
-
- rd_kafka_toppar_unlock(rktp);
-
- rd_kafka_brokers_broadcast_state_change(rkb->rkb_rk);
- break;
-
- case RD_KAFKA_OP_PARTITION_LEAVE:
- /*
- * Remove partition from broker toppars
- */
- rktp = rd_kafka_toppar_s2i(rko->rko_rktp);
-
- rd_kafka_toppar_lock(rktp);
-
- /* Multiple PARTITION_LEAVEs are possible during partition
- * migration, make sure we're supposed to handle this one. */
- if (unlikely(rktp->rktp_leader != rkb)) {
- rd_rkb_dbg(rkb, BROKER | RD_KAFKA_DBG_TOPIC, "TOPBRK",
- "Topic %s [%"PRId32"]: "
- "ignoring PARTITION_LEAVE: "
- "broker is not leader (%s)",
- rktp->rktp_rkt->rkt_topic->str,
- rktp->rktp_partition,
- rktp->rktp_leader ?
- rd_kafka_broker_name(rktp->rktp_leader) :
- "none");
- rd_kafka_toppar_unlock(rktp);
- break;
- }
- rd_kafka_toppar_unlock(rktp);
-
- /* Remove from fetcher list */
- rd_kafka_toppar_fetch_decide(rktp, rkb, 1/*force remove*/);
-
- rd_kafka_toppar_lock(rktp);
-
- rd_rkb_dbg(rkb, BROKER | RD_KAFKA_DBG_TOPIC, "TOPBRK",
- "Topic %s [%"PRId32"]: leaving broker "
- "(%d messages in xmitq, next leader %s, rktp %p)",
- rktp->rktp_rkt->rkt_topic->str, rktp->rktp_partition,
- rd_kafka_msgq_len(&rktp->rktp_xmit_msgq),
- rktp->rktp_next_leader ?
- rd_kafka_broker_name(rktp->rktp_next_leader) :
- "(none)", rktp);
-
- /* Prepend xmitq(broker-local) messages to the msgq(global).
- * There is no msgq_prepend() so we append msgq to xmitq
- * and then move the queue altogether back over to msgq. */
- rd_kafka_msgq_concat(&rktp->rktp_xmit_msgq,
- &rktp->rktp_msgq);
- rd_kafka_msgq_move(&rktp->rktp_msgq, &rktp->rktp_xmit_msgq);
-
- rd_kafka_broker_lock(rkb);
- TAILQ_REMOVE(&rkb->rkb_toppars, rktp, rktp_rkblink);
- rkb->rkb_toppar_cnt--;
- rd_kafka_broker_unlock(rkb);
- rd_kafka_broker_destroy(rktp->rktp_leader);
- rktp->rktp_msgq_wakeup_fd = -1;
- rktp->rktp_leader = NULL;
-
- /* Need to hold on to a refcount past q_enq() and
- * unlock() below */
- s_rktp = rktp->rktp_s_for_rkb;
- rktp->rktp_s_for_rkb = NULL;
-
- if (rktp->rktp_next_leader) {
- /* There is a next leader we need to migrate to. */
- rko->rko_type = RD_KAFKA_OP_PARTITION_JOIN;
- rd_kafka_q_enq(rktp->rktp_next_leader->rkb_ops, rko);
- rko = NULL;
- } else {
- rd_rkb_dbg(rkb, BROKER | RD_KAFKA_DBG_TOPIC, "TOPBRK",
- "Topic %s [%"PRId32"]: no next leader, "
- "failing %d message(s) in partition queue",
- rktp->rktp_rkt->rkt_topic->str,
- rktp->rktp_partition,
- rd_kafka_msgq_len(&rktp->rktp_msgq));
- rd_kafka_assert(NULL, rd_kafka_msgq_len(&rktp->rktp_xmit_msgq) == 0);
- rd_kafka_dr_msgq(rktp->rktp_rkt, &rktp->rktp_msgq,
- rd_kafka_terminating(rkb->rkb_rk) ?
- RD_KAFKA_RESP_ERR__DESTROY :
- RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION);
-
- }
-
- rd_kafka_toppar_unlock(rktp);
- rd_kafka_toppar_destroy(s_rktp);
-
- rd_kafka_brokers_broadcast_state_change(rkb->rkb_rk);
- break;
-
- case RD_KAFKA_OP_TERMINATE:
- /* nop: just a wake-up. */
- if (rkb->rkb_blocking_max_ms > 1)
- rkb->rkb_blocking_max_ms = 1; /* Speed up termination*/
- rd_rkb_dbg(rkb, BROKER, "TERM",
- "Received TERMINATE op in state %s: "
- "%d refcnts, %d toppar(s), %d fetch toppar(s), "
- "%d outbufs, %d waitresps, %d retrybufs",
- rd_kafka_broker_state_names[rkb->rkb_state],
- rd_refcnt_get(&rkb->rkb_refcnt),
- rkb->rkb_toppar_cnt, rkb->rkb_fetch_toppar_cnt,
- (int)rd_kafka_bufq_cnt(&rkb->rkb_outbufs),
- (int)rd_kafka_bufq_cnt(&rkb->rkb_waitresps),
- (int)rd_kafka_bufq_cnt(&rkb->rkb_retrybufs));
- ret = 0;
- break;
-
- default:
- rd_kafka_assert(rkb->rkb_rk, !*"unhandled op type");
- break;
- }
-
- if (rko)
- rd_kafka_op_destroy(rko);
-
- return ret;
-}
-
-
-/**
- * @brief Serve broker ops and IOs.
- *
- * @param abs_timeout Maximum block time (absolute time).
- *
- * @locality broker thread
- * @locks none
- */
-static void rd_kafka_broker_serve (rd_kafka_broker_t *rkb,
- rd_ts_t abs_timeout) {
- rd_kafka_op_t *rko;
- rd_ts_t now;
- int initial_state = rkb->rkb_state;
- int remains_ms = rd_timeout_remains(abs_timeout);
-
- /* Serve broker ops */
- while ((rko = rd_kafka_q_pop(rkb->rkb_ops,
- !rkb->rkb_transport ?
- remains_ms : RD_POLL_NOWAIT,
- 0))
- && rd_kafka_broker_op_serve(rkb, rko))
- remains_ms = RD_POLL_NOWAIT;
-
- /* If the broker state changed in op_serve() we minimize
- * the IO timeout since our caller might want to exit out of
- * its loop on state change. */
- if (likely(rkb->rkb_transport != NULL)) {
- int blocking_max_ms;
-
- if ((int)rkb->rkb_state != initial_state)
- blocking_max_ms = 0;
- else {
- int remains_ms = rd_timeout_remains(abs_timeout);
- if (remains_ms == RD_POLL_INFINITE ||
- remains_ms > rkb->rkb_blocking_max_ms)
- remains_ms = rkb->rkb_blocking_max_ms;
- blocking_max_ms = remains_ms;
- }
-
- /* Serve IO events */
- rd_kafka_transport_io_serve(rkb->rkb_transport,
- blocking_max_ms);
- }
-
- /* Scan wait-response queue for timeouts. */
- now = rd_clock();
- if (rd_interval(&rkb->rkb_timeout_scan_intvl, 1000000, now) > 0)
- rd_kafka_broker_timeout_scan(rkb, now);
-}
-
-
-/**
- * @brief Serve the toppar's assigned to this broker.
- *
- * @returns the minimum Fetch backoff time (abs timestamp) for the
- * partitions to fetch.
- *
- * @locality broker thread
- */
-static rd_ts_t rd_kafka_broker_toppars_serve (rd_kafka_broker_t *rkb) {
- rd_kafka_toppar_t *rktp, *rktp_tmp;
- rd_ts_t min_backoff = RD_TS_MAX;
-
- TAILQ_FOREACH_SAFE(rktp, &rkb->rkb_toppars, rktp_rkblink, rktp_tmp) {
- rd_ts_t backoff;
-
- /* Serve toppar to update desired rktp state */
- backoff = rd_kafka_broker_consumer_toppar_serve(rkb, rktp);
- if (backoff < min_backoff)
- min_backoff = backoff;
- }
-
- return min_backoff;
-}
-
-
-/**
- * Idle function for unassigned brokers
- * If \p timeout_ms is not RD_POLL_INFINITE the serve loop will be exited
- * regardless of state after this long (approximately).
- */
-static void rd_kafka_broker_ua_idle (rd_kafka_broker_t *rkb, int timeout_ms) {
- int initial_state = rkb->rkb_state;
- rd_ts_t abs_timeout;
-
- if (rd_kafka_terminating(rkb->rkb_rk))
- timeout_ms = 1;
- else if (timeout_ms == RD_POLL_INFINITE)
- timeout_ms = rkb->rkb_blocking_max_ms;
-
- abs_timeout = rd_timeout_init(timeout_ms);
-
- /* Since ua_idle is used during connection setup
- * in state ..BROKER_STATE_CONNECT we only run this loop
- * as long as the state remains the same as the initial, on a state
- * change - most likely to UP, a correct serve() function
- * should be used instead. */
- while (!rd_kafka_broker_terminating(rkb) &&
- (int)rkb->rkb_state == initial_state &&
- !rd_timeout_expired(rd_timeout_remains(abs_timeout))) {
-
- rd_kafka_broker_toppars_serve(rkb);
- rd_kafka_broker_serve(rkb, abs_timeout);
- }
-}
-
-
-/**
- * @brief Serve a toppar for producing.
- *
- * @param next_wakeup will be updated to when the next wake-up/attempt is
- * desired, only lower (sooner) values will be set.
- *
- * Locks: toppar_lock(rktp) MUST be held.
- * Returns the number of messages produced.
- */
-static int rd_kafka_toppar_producer_serve (rd_kafka_broker_t *rkb,
- rd_kafka_toppar_t *rktp,
- int do_timeout_scan,
- rd_ts_t now,
- rd_ts_t *next_wakeup) {
- int cnt = 0;
- int r;
-
- rd_rkb_dbg(rkb, QUEUE, "TOPPAR",
- "%.*s [%"PRId32"] %i+%i msgs",
- RD_KAFKAP_STR_PR(rktp->rktp_rkt->
- rkt_topic),
- rktp->rktp_partition,
- rd_atomic32_get(&rktp->rktp_msgq.rkmq_msg_cnt),
- rd_atomic32_get(&rktp->rktp_xmit_msgq.
- rkmq_msg_cnt));
-
- if (rd_atomic32_get(&rktp->rktp_msgq.rkmq_msg_cnt) > 0)
- rd_kafka_msgq_concat(&rktp->rktp_xmit_msgq, &rktp->rktp_msgq);
-
- /* Timeout scan */
- if (unlikely(do_timeout_scan)) {
- rd_kafka_msgq_t timedout = RD_KAFKA_MSGQ_INITIALIZER(timedout);
-
- if (rd_kafka_msgq_age_scan(&rktp->rktp_xmit_msgq,
- &timedout, now)) {
- /* Trigger delivery report for timed out messages */
- rd_kafka_dr_msgq(rktp->rktp_rkt, &timedout,
- RD_KAFKA_RESP_ERR__MSG_TIMED_OUT);
- }
- }
-
- r = rd_atomic32_get(&rktp->rktp_xmit_msgq.rkmq_msg_cnt);
- if (r == 0)
- return 0;
-
- /* Attempt to fill the batch size, but limit
- * our waiting to queue.buffering.max.ms
- * and batch.num.messages. */
- if (r < rkb->rkb_rk->rk_conf.batch_num_messages) {
- rd_kafka_msg_t *rkm_oldest;
- rd_ts_t wait_max;
-
- rkm_oldest = TAILQ_FIRST(&rktp->rktp_xmit_msgq.rkmq_msgs);
- if (unlikely(!rkm_oldest))
- return 0;
-
- /* Calculate maximum wait-time to
- * honour queue.buffering.max.ms contract. */
- wait_max = rd_kafka_msg_enq_time(rkm_oldest) +
- (rkb->rkb_rk->rk_conf.buffering_max_ms * 1000);
- if (wait_max > now) {
- if (wait_max < *next_wakeup)
- *next_wakeup = wait_max;
- /* Wait for more messages or queue.buffering.max.ms
- * to expire. */
- return 0;
- }
- }
-
- /* Send Produce requests for this toppar */
- while (1) {
- r = rd_kafka_ProduceRequest(rkb, rktp);
- if (likely(r > 0))
- cnt += r;
- else
- break;
- }
-
- return cnt;
-}
-
-
-/**
- * Producer serving
- */
-static void rd_kafka_broker_producer_serve (rd_kafka_broker_t *rkb) {
- rd_interval_t timeout_scan;
-
- rd_interval_init(&timeout_scan);
-
- rd_kafka_assert(rkb->rkb_rk, thrd_is_current(rkb->rkb_thread));
-
- rd_kafka_broker_lock(rkb);
-
- while (!rd_kafka_broker_terminating(rkb) &&
- rkb->rkb_state == RD_KAFKA_BROKER_STATE_UP) {
- rd_kafka_toppar_t *rktp;
- int cnt;
- rd_ts_t now;
- rd_ts_t next_wakeup;
- int do_timeout_scan = 0;
-
- rd_kafka_broker_unlock(rkb);
-
- now = rd_clock();
- next_wakeup = now + (rkb->rkb_rk->rk_conf.
- socket_blocking_max_ms * 1000);
-
- if (rd_interval(&timeout_scan, 1000*1000, now) >= 0)
- do_timeout_scan = 1;
-
- do {
- cnt = 0;
-
- /* Serve each toppar */
- TAILQ_FOREACH(rktp, &rkb->rkb_toppars, rktp_rkblink) {
- /* Serve toppar op queue */
- rd_kafka_toppar_lock(rktp);
- if (unlikely(rktp->rktp_leader != rkb)) {
- /* Currently migrating away from this
- * broker. */
- rd_kafka_toppar_unlock(rktp);
- continue;
- }
- if (unlikely(RD_KAFKA_TOPPAR_IS_PAUSED(rktp))) {
- /* Partition is paused */
- rd_kafka_toppar_unlock(rktp);
- continue;
- }
- /* Try producing toppar */
- cnt += rd_kafka_toppar_producer_serve(
- rkb, rktp, do_timeout_scan, now,
- &next_wakeup);
-
- rd_kafka_toppar_unlock(rktp);
- }
-
- } while (cnt);
-
- /* Check and move retry buffers */
- if (unlikely(rd_atomic32_get(&rkb->rkb_retrybufs.rkbq_cnt) > 0))
- rd_kafka_broker_retry_bufs_move(rkb);
-
- rkb->rkb_blocking_max_ms = (int)
- (next_wakeup > now ? (next_wakeup - now) / 1000 : 0);
- rd_kafka_broker_serve(rkb, next_wakeup);
-
- rd_kafka_broker_lock(rkb);
- }
-
- rd_kafka_broker_unlock(rkb);
-}
-
-
-
-
-
-
-
-/**
- * Backoff the next Fetch request (due to error).
- */
-static void rd_kafka_broker_fetch_backoff (rd_kafka_broker_t *rkb,
- rd_kafka_resp_err_t err) {
- int backoff_ms = rkb->rkb_rk->rk_conf.fetch_error_backoff_ms;
- rkb->rkb_ts_fetch_backoff = rd_clock() + (backoff_ms * 1000);
- rd_rkb_dbg(rkb, FETCH, "BACKOFF",
- "Fetch backoff for %dms: %s",
- backoff_ms, rd_kafka_err2str(err));
-}
-
-/**
- * @brief Backoff the next Fetch for specific partition
- */
-static void rd_kafka_toppar_fetch_backoff (rd_kafka_broker_t *rkb,
- rd_kafka_toppar_t *rktp,
- rd_kafka_resp_err_t err) {
- int backoff_ms = rkb->rkb_rk->rk_conf.fetch_error_backoff_ms;
- rktp->rktp_ts_fetch_backoff = rd_clock() + (backoff_ms * 1000);
- rd_rkb_dbg(rkb, FETCH, "BACKOFF",
- "%s [%"PRId32"]: Fetch backoff for %dms: %s",
- rktp->rktp_rkt->rkt_topic->str, rktp->rktp_partition,
- backoff_ms, rd_kafka_err2str(err));
-}
-
-
-/**
- * Parses and handles a Fetch reply.
- * Returns 0 on success or an error code on failure.
- */
-static rd_kafka_resp_err_t
-rd_kafka_fetch_reply_handle (rd_kafka_broker_t *rkb,
- rd_kafka_buf_t *rkbuf, rd_kafka_buf_t *request) {
- int32_t TopicArrayCnt;
- int i;
- const int log_decode_errors = LOG_ERR;
- shptr_rd_kafka_itopic_t *s_rkt = NULL;
-
- if (rd_kafka_buf_ApiVersion(request) >= 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);
- }
-
- rd_kafka_buf_read_i32(rkbuf, &TopicArrayCnt);
- /* Verify that TopicArrayCnt seems to be in line with remaining size */
- rd_kafka_buf_check_len(rkbuf,
- TopicArrayCnt * (3/*topic min size*/ +
- 4/*PartitionArrayCnt*/ +
- 4+2+8+4/*inner header*/));
-
- for (i = 0 ; i < TopicArrayCnt ; i++) {
- rd_kafkap_str_t topic;
- int32_t fetch_version;
- int32_t PartitionArrayCnt;
- int j;
-
- rd_kafka_buf_read_str(rkbuf, &topic);
- rd_kafka_buf_read_i32(rkbuf, &PartitionArrayCnt);
-
- s_rkt = rd_kafka_topic_find0(rkb->rkb_rk, &topic);
-
- for (j = 0 ; j < PartitionArrayCnt ; j++) {
- struct rd_kafka_toppar_ver *tver, tver_skel;
- rd_kafka_toppar_t *rktp;
- shptr_rd_kafka_toppar_t *s_rktp = NULL;
- rd_slice_t save_slice;
- struct {
- int32_t Partition;
- int16_t ErrorCode;
- int64_t HighwaterMarkOffset;
- int64_t LastStableOffset; /* v4 */
- int32_t MessageSetSize;
- } hdr;
- rd_kafka_resp_err_t err;
-
- rd_kafka_buf_read_i32(rkbuf, &hdr.Partition);
- rd_kafka_buf_read_i16(rkbuf, &hdr.ErrorCode);
- rd_kafka_buf_read_i64(rkbuf, &hdr.HighwaterMarkOffset);
-
- if (rd_kafka_buf_ApiVersion(request) == 4) {
- int32_t AbortedTxCnt;
- rd_kafka_buf_read_i64(rkbuf,
- &hdr.LastStableOffset);
- rd_kafka_buf_read_i32(rkbuf, &AbortedTxCnt);
- /* Ignore aborted transactions for now */
- if (AbortedTxCnt > 0)
- rd_kafka_buf_skip(rkbuf,
- AbortedTxCnt * (8+8));
- } else
- hdr.LastStableOffset = -1;
-
- rd_kafka_buf_read_i32(rkbuf, &hdr.MessageSetSize);
-
- if (unlikely(hdr.MessageSetSize < 0))
- rd_kafka_buf_parse_fail(
- rkbuf,
- "%.*s [%"PRId32"]: "
- "invalid MessageSetSize %"PRId32,
- RD_KAFKAP_STR_PR(&topic),
- hdr.Partition,
- hdr.MessageSetSize);
-
- /* Look up topic+partition */
- if (likely(s_rkt != NULL)) {
- rd_kafka_itopic_t *rkt;
- rkt = rd_kafka_topic_s2i(s_rkt);
- rd_kafka_topic_rdlock(rkt);
- s_rktp = rd_kafka_toppar_get(
- rkt, hdr.Partition, 0/*no ua-on-miss*/);
- rd_kafka_topic_rdunlock(rkt);
- }
-
- if (unlikely(!s_rkt || !s_rktp)) {
- rd_rkb_dbg(rkb, TOPIC, "UNKTOPIC",
- "Received Fetch response "
- "(error %hu) for unknown topic "
- "%.*s [%"PRId32"]: ignoring",
- hdr.ErrorCode,
- RD_KAFKAP_STR_PR(&topic),
- hdr.Partition);
- rd_kafka_buf_skip(rkbuf, hdr.MessageSetSize);
- continue;
- }
-
- rktp = rd_kafka_toppar_s2i(s_rktp);
-
- rd_kafka_toppar_lock(rktp);
- /* Make sure toppar hasn't moved to another broker
- * during the lifetime of the request. */
- if (unlikely(rktp->rktp_leader != rkb)) {
- rd_kafka_toppar_unlock(rktp);
- rd_rkb_dbg(rkb, MSG, "FETCH",
-
<TRUNCATED>