/* * 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. */ #if defined(__MINGW32__) #include #endif #ifndef _WIN32 #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 #endif #include #include #include #include #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 "rdkafka_idempotence.h" #include "rdkafka_txnmgr.h" #include "rdtime.h" #include "rdcrc32.h" #include "rdrand.h" #include "rdkafka_lz4.h" #if WITH_SSL #include #endif #include "rdendian.h" #include "rdunittest.h" static const int rd_kafka_max_block_ms = 1000; const char *rd_kafka_broker_state_names[] = { "INIT", "DOWN", "TRY_CONNECT", "CONNECT", "SSL_HANDSHAKE", "AUTH_LEGACY", "UP", "UPDATE", "APIVERSION_QUERY", "AUTH_HANDSHAKE", "AUTH_REQ" }; 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 }; /** * @returns true if the broker needs a persistent connection * @locaility broker thread */ static RD_INLINE rd_bool_t rd_kafka_broker_needs_persistent_connection (rd_kafka_broker_t *rkb) { return rkb->rkb_persistconn.internal || rd_atomic32_get(&rkb->rkb_persistconn.coord); } /** * @returns > 0 if a connection to this broker is needed, else 0. * @locality broker thread * @locks none */ static RD_INLINE int rd_kafka_broker_needs_connection (rd_kafka_broker_t *rkb) { return rkb->rkb_state == RD_KAFKA_BROKER_STATE_INIT && !rd_kafka_terminating(rkb->rkb_rk) && !rd_kafka_fatal_error_code(rkb->rkb_rk) && (!rkb->rkb_rk->rk_conf.sparse_connections || rd_kafka_broker_needs_persistent_connection(rkb)); } static void rd_kafka_broker_handle_purge_queues (rd_kafka_broker_t *rkb, rd_kafka_op_t *rko); static void rd_kafka_broker_trigger_monitors (rd_kafka_broker_t *rkb); #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 or logical broker in which case we * omit the protocol part. */ if (proto != RD_KAFKA_PROTO_PLAINTEXT && source != RD_KAFKA_LOGICAL) { 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_LOGICAL ? "" : (source == RD_KAFKA_INTERNAL ? "/internal" : "/bootstrap")); else rd_snprintf(dest, dsize, "%s/%"PRId32, nodename, nodeid); } /** * @brief Enable protocol feature(s) for the current broker. * * @locks broker_lock MUST be held * @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. * * @locks broker_lock MUST be held * @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); if (featuresp) *featuresp = rkb->rkb_features; if (rkb->rkb_features & RD_KAFKA_FEATURE_UNITTEST) { /* For unit tests let the broker support everything. */ rd_kafka_broker_unlock(rkb); return maxver; } retp = bsearch(&skel, rkb->rkb_ApiVersions, rkb->rkb_ApiVersions_cnt, sizeof(*rkb->rkb_ApiVersions), rd_kafka_ApiVersion_key_cmp); if (retp) ret = *retp; 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; } /** * @brief Set broker state. * * \c rkb->rkb_state is the previous state, while * \p state is the new state. * * @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) { /* Propagate ALL_BROKERS_DOWN event if all brokers are * now down, unless we're terminating. */ 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_broker_addrless_cnt) && !rd_kafka_terminating(rkb->rkb_rk)) 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) - rd_atomic32_get(&rkb->rkb_rk-> rk_broker_addrless_cnt)); rkb->rkb_down_reported = 1; } else if (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; } if (rkb->rkb_source != RD_KAFKA_INTERNAL) { if (rd_kafka_broker_state_is_up(state) && !rd_kafka_broker_state_is_up(rkb->rkb_state)) { /* Up -> Down */ rd_atomic32_add(&rkb->rkb_rk->rk_broker_up_cnt, 1); rd_kafka_broker_trigger_monitors(rkb); if (RD_KAFKA_BROKER_IS_LOGICAL(rkb)) rd_atomic32_add(&rkb->rkb_rk-> rk_logical_broker_up_cnt, 1); } else if (rd_kafka_broker_state_is_up(rkb->rkb_state) && !rd_kafka_broker_state_is_up(state)) { /* ~Down(!Up) -> Up */ rd_atomic32_sub(&rkb->rkb_rk->rk_broker_up_cnt, 1); rd_kafka_broker_trigger_monitors(rkb); if (RD_KAFKA_BROKER_IS_LOGICAL(rkb)) rd_atomic32_sub(&rkb->rkb_rk-> rk_logical_broker_up_cnt, 1); } } rkb->rkb_state = state; rkb->rkb_ts_state = rd_clock(); rd_kafka_brokers_broadcast_state_change(rkb->rkb_rk); } /** * @brief Set, log and propagate broker fail error. * * @param rkb Broker connection that failed. * @param level Syslog level. LOG_DEBUG will not be logged unless debugging * is enabled. * @param err The type of error that occurred. * @param fmt Format string. * @param ap Format string arguments. * * @locks none * @locality broker thread */ static void rd_kafka_broker_set_error (rd_kafka_broker_t *rkb, int level, rd_kafka_resp_err_t err, const char *fmt, va_list ap) { char errstr[512]; char extra[128]; size_t of = 0, ofe; rd_bool_t identical, suppress; int state_duration_ms = (int)((rd_clock() - rkb->rkb_ts_state)/1000); /* If this is a logical broker we include its current nodename/address * in the log message. */ rd_kafka_broker_lock(rkb); if (rkb->rkb_source == RD_KAFKA_LOGICAL && *rkb->rkb_nodename) { of = (size_t)rd_snprintf(errstr, sizeof(errstr), "%s: ", rkb->rkb_nodename); if (of > sizeof(errstr)) of = 0; /* If nodename overflows the entire buffer we * skip it completely since the error message * itself is more important. */ } rd_kafka_broker_unlock(rkb); ofe = (size_t)rd_vsnprintf(errstr+of, sizeof(errstr)-of, fmt, ap); if (ofe > sizeof(errstr)-of) ofe = sizeof(errstr)-of; of += ofe; /* Provide more meaningful error messages in certain cases */ if (err == RD_KAFKA_RESP_ERR__TRANSPORT && !strcmp(errstr, "Disconnected")) { if (rkb->rkb_state == RD_KAFKA_BROKER_STATE_APIVERSION_QUERY) { /* A disconnect while requesting ApiVersion typically * means we're connecting to a SSL-listener as * PLAINTEXT, but may also be caused by connecting to * a broker that does not support ApiVersion (<0.10). */ if (rkb->rkb_proto != RD_KAFKA_PROTO_SSL && rkb->rkb_proto != RD_KAFKA_PROTO_SASL_SSL) rd_kafka_broker_set_error( rkb, level, err, "Disconnected while requesting " "ApiVersion: " "might be caused by incorrect " "security.protocol configuration " "(connecting to a SSL listener?) or " "broker version is < 0.10 " "(see api.version.request)", ap/*ignored*/); else rd_kafka_broker_set_error( rkb, level, err, "Disconnected while requesting " "ApiVersion: " "might be caused by broker version " "< 0.10 (see api.version.request)", ap/*ignored*/); return; } else if (rkb->rkb_state == RD_KAFKA_BROKER_STATE_UP && state_duration_ms < 2000/*2s*/ && rkb->rkb_rk->rk_conf.security_protocol != RD_KAFKA_PROTO_SASL_SSL && rkb->rkb_rk->rk_conf.security_protocol != RD_KAFKA_PROTO_SASL_PLAINTEXT) { /* If disconnected shortly after transitioning to UP * state it typically means the broker listener is * configured for SASL authentication but the client * is not. */ rd_kafka_broker_set_error( rkb, level, err, "Disconnected: verify that security.protocol " "is correctly configured, broker might " "require SASL authentication", ap/*ignored*/); return; } } /* Check if error is identical to last error (prior to appending * the variable suffix "after Xms in state Y"), if so we should * suppress it. */ identical = err == rkb->rkb_last_err.err && !strcmp(rkb->rkb_last_err.errstr, errstr); suppress = identical && rd_interval(&rkb->rkb_suppress.fail_error, 30 * 1000 * 1000 /*30s*/, 0) <= 0; /* Copy last error prior to adding extras */ rkb->rkb_last_err.err = err; rd_strlcpy(rkb->rkb_last_err.errstr, errstr, sizeof(rkb->rkb_last_err.errstr)); /* Time since last state change to help debug connection issues */ ofe = rd_snprintf(extra, sizeof(extra), "after %dms in state %s", state_duration_ms, rd_kafka_broker_state_names[rkb->rkb_state]); /* Number of suppressed identical logs */ if (identical && !suppress && rkb->rkb_last_err.cnt >= 1 && ofe + 30 < sizeof(extra)) { size_t r = (size_t)rd_snprintf( extra+ofe, sizeof(extra)-ofe, ", %d identical error(s) suppressed", rkb->rkb_last_err.cnt); if (r < sizeof(extra)-ofe) ofe += r; else ofe = sizeof(extra); } /* Append the extra info if there is enough room */ if (ofe > 0 && of + ofe + 4 < sizeof(errstr)) rd_snprintf(errstr+of, sizeof(errstr)-of, " (%s)", extra); /* Don't log interrupt-wakeups when terminating */ if (err == RD_KAFKA_RESP_ERR__INTR && rd_kafka_terminating(rkb->rkb_rk)) suppress = rd_true; if (!suppress) rkb->rkb_last_err.cnt = 1; else rkb->rkb_last_err.cnt++; rd_rkb_dbg(rkb, BROKER, "FAIL", "%s (%s)%s%s", errstr, rd_kafka_err2name(err), identical ? ": identical to last error" : "", suppress ? ": error log suppressed" : ""); if (level != LOG_DEBUG && (level <= LOG_CRIT || !suppress)) { rd_kafka_log(rkb->rkb_rk, level, "FAIL", "%s: %s", rkb->rkb_name, errstr); /* Send ERR op to application for processing. */ rd_kafka_q_op_err(rkb->rkb_rk->rk_rep, err, "%s: %s", rkb->rkb_name, errstr); } } /** * @brief Failure propagation to application. * * Will tear down connection to broker and trigger a reconnect. * * \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; rd_kafka_bufq_t tmpq_waitresp, tmpq; int old_state; rd_kafka_assert(rkb->rkb_rk, thrd_is_current(rkb->rkb_thread)); if (rkb->rkb_transport) { rd_kafka_transport_close(rkb->rkb_transport); rkb->rkb_transport = NULL; if (rkb->rkb_state >= RD_KAFKA_BROKER_STATE_UP) rd_atomic32_add(&rkb->rkb_c.disconnects, 1); } rkb->rkb_req_timeouts = 0; if (rkb->rkb_recv_buf) { rd_kafka_buf_destroy(rkb->rkb_recv_buf); rkb->rkb_recv_buf = NULL; } va_start(ap, fmt); rd_kafka_broker_set_error(rkb, level, err, fmt, ap); va_end(ap); rd_kafka_broker_lock(rkb); /* 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); rd_atomic64_set(&rkb->rkb_c.ts_send, 0); rd_atomic64_set(&rkb->rkb_c.ts_recv, 0); /* * 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 in-flight buffers (might get re-enqueued in case * of retries). */ rd_kafka_bufq_purge(rkb, &tmpq_waitresp, err); /* Purge the waiting-in-output-queue buffers, * might also get re-enqueued. */ rd_kafka_bufq_purge(rkb, &tmpq, /* If failure was caused by a timeout, * adjust the error code for in-queue requests. */ err == RD_KAFKA_RESP_ERR__TIMED_OUT ? RD_KAFKA_RESP_ERR__TIMED_OUT_QUEUE : err); /* 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); } /* Query for topic leaders to quickly pick up on failover. */ if (err != RD_KAFKA_RESP_ERR__DESTROY && old_state >= RD_KAFKA_BROKER_STATE_UP) rd_kafka_metadata_refresh_known_topics(rkb->rkb_rk, NULL, rd_true/*force*/, "broker down"); } /** * @brief Handle broker connection close. * * @locality broker thread */ void rd_kafka_broker_conn_closed (rd_kafka_broker_t *rkb, rd_kafka_resp_err_t err, const char *errstr) { int log_level = LOG_ERR; if (!rkb->rkb_rk->rk_conf.log_connection_close) { /* Silence all connection closes */ log_level = LOG_DEBUG; } else { /* Silence close logs for connections that are idle, * it is most likely the broker's idle connection * reaper kicking in. * * Indications there might be an error and not an * idle disconnect: * - If the connection age is low a disconnect * typically indicates a failure, such as protocol mismatch. * - If the connection hasn't been idle long enough. * - There are outstanding requests, or requests enqueued. * * For non-idle connections, adjust log level: * - requests in-flight: LOG_WARNING * - else: LOG_INFO */ rd_ts_t now = rd_clock(); rd_ts_t minidle = RD_MAX(60*1000/*60s*/, rkb->rkb_rk->rk_conf.socket_timeout_ms) * 1000; int inflight = rd_kafka_bufq_cnt(&rkb->rkb_waitresps); int inqueue = rd_kafka_bufq_cnt(&rkb->rkb_outbufs); if (rkb->rkb_ts_state + minidle < now && rd_atomic64_get(&rkb->rkb_c.ts_send) + minidle < now && inflight + inqueue == 0) log_level = LOG_DEBUG; else if (inflight > 1) log_level = LOG_WARNING; else log_level = LOG_INFO; } rd_kafka_broker_fail(rkb, log_level, err, "%s", errstr); } /** * @brief Purge requests in \p rkbq matching request \p ApiKey * and partition \p rktp. * * @warning ApiKey must be RD_KAFKAP_Produce * * @returns the number of purged buffers. * * @locality broker thread */ static int rd_kafka_broker_bufq_purge_by_toppar (rd_kafka_broker_t *rkb, rd_kafka_bufq_t *rkbq, int64_t ApiKey, rd_kafka_toppar_t *rktp, rd_kafka_resp_err_t err) { rd_kafka_buf_t *rkbuf, *tmp; int cnt = 0; rd_assert(ApiKey == RD_KAFKAP_Produce); TAILQ_FOREACH_SAFE(rkbuf, &rkbq->rkbq_bufs, rkbuf_link, tmp) { if (rkbuf->rkbuf_reqhdr.ApiKey != ApiKey || rkbuf->rkbuf_u.Produce.batch.rktp != rktp || /* Skip partially sent buffers and let them transmit. * The alternative would be to kill the connection here, * which is more drastic and costly. */ rd_slice_offset(&rkbuf->rkbuf_reader) > 0) continue; rd_kafka_bufq_deq(rkbq, rkbuf); rd_kafka_buf_callback(rkb->rkb_rk, rkb, err, NULL, rkbuf); cnt++; } return cnt; } /** * 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. * * @param ApiKey Only match requests with this ApiKey, or -1 for all. * @param now If 0, all buffers will time out, else the current clock. * @param description "N requests timed out ", e.g., "in flight". * Only used if log_first_n > 0. * @param log_first_n Log the first N request timeouts. * * @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, int16_t ApiKey, rd_kafka_resp_err_t err, rd_ts_t now, const char *description, int log_first_n) { rd_kafka_buf_t *rkbuf, *tmp; int cnt = 0; int idx = -1; const rd_kafka_buf_t *holb; restart: holb = TAILQ_FIRST(&rkbq->rkbq_bufs); TAILQ_FOREACH_SAFE(rkbuf, &rkbq->rkbq_bufs, rkbuf_link, tmp) { rd_kafka_broker_state_t pre_state, post_state; idx++; if (likely(now && rkbuf->rkbuf_ts_timeout > now)) continue; if (ApiKey != -1 && rkbuf->rkbuf_reqhdr.ApiKey != ApiKey) 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 (now && cnt < log_first_n) { char holbstr[128]; /* Head of line blocking: * If this is not the first request in queue, but the * initial first request did not time out, * it typically means the first request is a * long-running blocking one, holding up the * sub-sequent requests. * In this case log what is likely holding up the * requests and what caused this request to time out. */ if (holb && holb == TAILQ_FIRST(&rkbq->rkbq_bufs)) { rd_snprintf(holbstr, sizeof(holbstr), ": possibly held back by " "preceeding%s %sRequest with " "timeout in %dms", (holb->rkbuf_flags & RD_KAFKA_OP_F_BLOCKING) ? " blocking" : "", rd_kafka_ApiKey2str(holb-> rkbuf_reqhdr. ApiKey), (int)((holb->rkbuf_ts_timeout - now) / 1000)); /* Only log the HOLB once */ holb = NULL; } else { *holbstr = '\0'; } rd_rkb_log(rkb, LOG_NOTICE, "REQTMOUT", "Timed out %sRequest %s " "(after %"PRId64"ms, timeout #%d)%s", rd_kafka_ApiKey2str(rkbuf->rkbuf_reqhdr. ApiKey), description, rkbuf->rkbuf_ts_sent/1000, cnt, holbstr); } 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); pre_state = rd_kafka_broker_get_state(rkb); rd_kafka_buf_callback(rkb->rkb_rk, rkb, err, NULL, rkbuf); cnt++; /* If the buf_callback() triggered a broker state change * (typically through broker_fail()) we can't trust the * queue we are scanning to not have been touched, so we * either restart the scan or bail out (if broker is now down), * depending on the new state. #2326 */ post_state = rd_kafka_broker_get_state(rkb); if (pre_state != post_state) { /* If the new state is DOWN it means broker_fail() * was called which may have modified the queues, * to keep things safe we stop scanning this queue. */ if (post_state == RD_KAFKA_BROKER_STATE_DOWN) break; /* Else start scanning the queue from the beginning. */ goto restart; } } 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 inflight_cnt, retry_cnt, outq_cnt; int partial_cnt = 0; rd_kafka_assert(rkb->rkb_rk, thrd_is_current(rkb->rkb_thread)); /* In-flight requests waiting for response */ inflight_cnt = rd_kafka_broker_bufq_timeout_scan( rkb, 1, &rkb->rkb_waitresps, NULL, -1, RD_KAFKA_RESP_ERR__TIMED_OUT, now, "in flight", 5); /* Requests in retry queue */ retry_cnt = rd_kafka_broker_bufq_timeout_scan( rkb, 0, &rkb->rkb_retrybufs, NULL, -1, RD_KAFKA_RESP_ERR__TIMED_OUT_QUEUE, now, "in retry queue", 0); /* Requests in local queue not sent yet. * partial_cnt is included in outq_cnt and denotes a request * that has been partially transmitted. */ outq_cnt = rd_kafka_broker_bufq_timeout_scan( rkb, 0, &rkb->rkb_outbufs, &partial_cnt, -1, RD_KAFKA_RESP_ERR__TIMED_OUT_QUEUE, now, "in output queue", 0); if (inflight_cnt + retry_cnt + outq_cnt + partial_cnt > 0) { rd_rkb_log(rkb, LOG_WARNING, "REQTMOUT", "Timed out %i in-flight, %i retry-queued, " "%i out-queue, %i partially-sent requests", inflight_cnt, retry_cnt, outq_cnt, partial_cnt); rkb->rkb_req_timeouts += inflight_cnt + outq_cnt; rd_atomic64_add(&rkb->rkb_c.req_timeouts, inflight_cnt + outq_cnt); /* If this was a partially sent request that timed out, or the * number of timed out requests have reached the * socket.max.fails threshold, we need to take down the * connection. */ if (partial_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; rd_kafka_broker_fail(rkb, LOG_ERR, RD_KAFKA_RESP_ERR__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 *nodename, rd_bool_t reset_cached_addr) { const char *errstr; int save_idx = 0; if (!*nodename && rkb->rkb_source == RD_KAFKA_LOGICAL) { rd_kafka_broker_fail(rkb, LOG_DEBUG, RD_KAFKA_RESP_ERR__RESOLVE, "Logical broker has no address yet"); return -1; } if (rkb->rkb_rsal && (reset_cached_addr || rkb->rkb_ts_rsal_last + (rkb->rkb_rk->rk_conf.broker_addr_ttl*1000) < rd_clock())) { /* Address list has expired. */ /* Save the address index to make sure we still round-robin * if we get the same address list back */ save_idx = rkb->rkb_rsal->rsal_curr; rd_sockaddr_list_destroy(rkb->rkb_rsal); rkb->rkb_rsal = NULL; } if (!rkb->rkb_rsal) { /* Resolve */ rkb->rkb_rsal = rd_getaddrinfo(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, "Failed to resolve '%s': %s", nodename, errstr); return -1; } else { rkb->rkb_ts_rsal_last = rd_clock(); /* Continue at previous round-robin position */ if (rkb->rkb_rsal->rsal_cnt > save_idx) rkb->rkb_rsal->rsal_curr = save_idx; } } return 0; } static void rd_kafka_broker_buf_enq0 (rd_kafka_broker_t *rkb, rd_kafka_buf_t *rkbuf) { rd_ts_t now; rd_kafka_assert(rkb->rkb_rk, thrd_is_current(rkb->rkb_thread)); if (rkb->rkb_rk->rk_conf.sparse_connections && rkb->rkb_state == RD_KAFKA_BROKER_STATE_INIT) { /* Sparse connections: * Trigger connection when a new request is enqueued. */ rkb->rkb_persistconn.internal++; rd_kafka_broker_lock(rkb); rd_kafka_broker_set_state(rkb, RD_KAFKA_BROKER_STATE_TRY_CONNECT); rd_kafka_broker_unlock(rkb); } now = rd_clock(); rkbuf->rkbuf_ts_enq = now; rkbuf->rkbuf_flags &= ~RD_KAFKA_OP_F_SENT; /* Calculate request attempt timeout */ rd_kafka_buf_calc_timeout(rkb->rkb_rk, rkbuf, now); if (likely(rkbuf->rkbuf_prio == RD_KAFKA_PRIO_NORMAL)) { /* Insert request at tail of queue */ TAILQ_INSERT_TAIL(&rkb->rkb_outbufs.rkbq_bufs, rkbuf, rkbuf_link); } else { /* Insert request after any requests with a higher or * equal priority. * Also make sure the request is after added any partially * sent request (of any prio). * 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. */ rd_kafka_buf_t *prev, *after = NULL; TAILQ_FOREACH(prev, &rkb->rkb_outbufs.rkbq_bufs, rkbuf_link) { if (prev->rkbuf_prio < rkbuf->rkbuf_prio && 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); } rd_atomic32_add(&rkb->rkb_outbufs.rkbq_cnt, 1); if (rkbuf->rkbuf_reqhdr.ApiKey == RD_KAFKAP_Produce) rd_atomic32_add(&rkb->rkb_outbufs.rkbq_msg_cnt, rd_kafka_msgq_len(&rkbuf->rkbuf_batch.msgq)); } /** * 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; if (rkbuf->rkbuf_flags & RD_KAFKA_OP_F_FLEXVER) { /* Empty struct tags */ rd_kafka_buf_write_i8(rkbuf, 0); } /* Calculate total request buffer length. */ totsize = rd_buf_len(&rkbuf->rkbuf_buf) - 4; /* Set up a buffer reader for sending the buffer. */ rd_slice_init_full(&rkbuf->rkbuf_reader, &rkbuf->rkbuf_buf); /** * Update request header fields */ /* Total request 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); } /** * 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); 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 future 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 Same as rd_kafka_brokers_wait_state_change() but will trigger * the wakeup asynchronously through the provided \p eonce. * * If the eonce was added to the wait list its reference count * will have been updated, this reference is later removed by * rd_kafka_broker_state_change_trigger_eonce() by calling trigger(). * * @returns 1 if the \p eonce was added to the wait-broker-state-changes list, * or 0 if the \p stored_version is outdated in which case the * caller should redo the broker lookup. */ int rd_kafka_brokers_wait_state_change_async (rd_kafka_t *rk, int stored_version, rd_kafka_enq_once_t *eonce) { int r = 1; mtx_lock(&rk->rk_broker_state_change_lock); if (stored_version != rk->rk_broker_state_change_version) r = 0; else { rd_kafka_enq_once_add_source(eonce, "wait broker state change"); rd_list_add(&rk->rk_broker_state_change_waiters, eonce); } mtx_unlock(&rk->rk_broker_state_change_lock); return r; } /** * @brief eonce trigger callback for rd_list_apply() call in * rd_kafka_brokers_broadcast_state_change() */ static int rd_kafka_broker_state_change_trigger_eonce (void *elem, void *opaque) { rd_kafka_enq_once_t *eonce = elem; rd_kafka_enq_once_trigger(eonce, RD_KAFKA_RESP_ERR_NO_ERROR, "broker state change"); return 0; /* remove eonce from list */ } /** * @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); /* Bump version */ rk->rk_broker_state_change_version++; /* Trigger waiters */ rd_list_apply(&rk->rk_broker_state_change_waiters, rd_kafka_broker_state_change_trigger_eonce, NULL); /* Broadcast to listeners */ cnd_broadcast(&rk->rk_broker_state_change_cnd); mtx_unlock(&rk->rk_broker_state_change_lock); } /** * @returns a random broker (with refcnt increased) with matching \p state * and where the \p filter function returns 0. * * Uses reservoir sampling. * * @param is_up Any broker that is up (UP or UPDATE state), \p state is ignored. * @param filtered_cnt Optional pointer to integer which will be set to the * number of brokers that matches the \p state or \p is_up but * were filtered out by \p filter. * @param 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_*lock() MUST be held * @locality any */ static rd_kafka_broker_t * rd_kafka_broker_random0 (const char *func, int line, rd_kafka_t *rk, rd_bool_t is_up, int state, int *filtered_cnt, int (*filter) (rd_kafka_broker_t *rk, void *opaque), void *opaque) { rd_kafka_broker_t *rkb, *good = NULL; int cnt = 0; int fcnt = 0; TAILQ_FOREACH(rkb, &rk->rk_brokers, rkb_link) { if (RD_KAFKA_BROKER_IS_LOGICAL(rkb)) continue; rd_kafka_broker_lock(rkb); if ((is_up && rd_kafka_broker_state_is_up(rkb->rkb_state)) || (!is_up && (int)rkb->rkb_state == state)) { if (filter && filter(rkb, opaque)) { /* Filtered out */ fcnt++; } else { if (cnt < 1 || rd_jitter(0, cnt) < 1) { if (good) rd_kafka_broker_destroy(good); rd_kafka_broker_keep_fl(func, line, rkb); good = rkb; } cnt += 1; } } rd_kafka_broker_unlock(rkb); } if (filtered_cnt) *filtered_cnt = fcnt; return good; } #define rd_kafka_broker_random(rk,state,filter,opaque) \ rd_kafka_broker_random0(__FUNCTION__, __LINE__, \ rk, rd_false, state, NULL, filter, opaque) /** * @returns the broker (with refcnt increased) with the highest weight based * based on the provided weighing function. * * If multiple brokers share the same weight reservoir sampling will be used * to randomly select one. * * @param weight_cb Weighing function that should return the sort weight * for the given broker. * Higher weight is better. * A weight of <= 0 will filter out the broker. * The passed broker object is locked. * @param features (optional) Required broker features. * * @locks_required rk(read) * @locality any */ static rd_kafka_broker_t * rd_kafka_broker_weighted (rd_kafka_t *rk, int (*weight_cb) (rd_kafka_broker_t *rkb), int features) { rd_kafka_broker_t *rkb, *good = NULL; int highest = 0; int cnt = 0; TAILQ_FOREACH(rkb, &rk->rk_brokers, rkb_link) { int weight; rd_kafka_broker_lock(rkb); if (features && (rkb->rkb_features & features) != features) weight = 0; else weight = weight_cb(rkb); rd_kafka_broker_unlock(rkb); if (weight <= 0 || weight < highest) continue; if (weight > highest) { highest = weight; cnt = 0; } /* If same weight (cnt > 0), use reservoir sampling */ if (cnt < 1 || rd_jitter(0, cnt) < 1) { if (good) rd_kafka_broker_destroy(good); rd_kafka_broker_keep(rkb); good = rkb; } cnt++; } return good; } /** * @brief Weighing function to select a usable broker connections, * promoting connections according to the scoring below. * * Priority order: * - is not a bootstrap broker * - least idle last 10 minutes (unless blocking) * - least idle hours (if above 10 minutes idle) * - is not a logical broker (these connections have dedicated use and should * preferably not be used for other purposes) * - is not blocking * * Will prefer the most recently used broker connection for two reasons: * - this connection is most likely to function properly. * - allows truly idle connections to be killed by the broker's/LB's * idle connection reaper. * * Connection must be up. * * @locks_required rkb */ static int rd_kafka_broker_weight_usable (rd_kafka_broker_t *rkb) { int weight = 0; if (!rd_kafka_broker_state_is_up(rkb->rkb_state)) return 0; weight += 2000 * (rkb->rkb_nodeid != -1 && !RD_KAFKA_BROKER_IS_LOGICAL(rkb)); weight += 10 * !RD_KAFKA_BROKER_IS_LOGICAL(rkb); if (likely(!rd_atomic32_get(&rkb->rkb_blocking_request_cnt))) { rd_ts_t tx_last = rd_atomic64_get(&rkb->rkb_c.ts_send); int idle = (int)((rd_clock() - (tx_last > 0 ? tx_last : rkb->rkb_ts_state)) / 1000000); weight += 1; /* is not blocking */ /* Prefer least idle broker (based on last 10 minutes use) */ if (idle < 0) ; /*clock going backwards? do nothing */ else if (idle < 600/*10 minutes*/) weight += 1000 + (600 - idle); else /* Else least idle hours (capped to 100h) */ weight += 100 + (100 - RD_MIN((idle / 3600), 100)); } return weight; } /** * @brief Returns a random broker (with refcnt increased) in state \p state. * * Uses Reservoir sampling. * * @param filter is optional, see rd_kafka_broker_random(). * * @sa rd_kafka_broker_random * * @locks rd_kafka_*lock(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, const char *reason) { rd_kafka_broker_t *rkb; rkb = rd_kafka_broker_random(rk, state, filter, opaque); if (!rkb && rk->rk_conf.sparse_connections) { /* Sparse connections: * If no eligible broker was found, schedule * a random broker for connecting. */ rd_kafka_connect_any(rk, reason); } return rkb; } /** * @brief Returns a random broker (with refcnt increased) which is up. * * @param filtered_cnt optional, see rd_kafka_broker_random0(). * @param filter is optional, see rd_kafka_broker_random0(). * * @sa rd_kafka_broker_random * * @locks rd_kafka_*lock(rk) MUST be held. * @locality any thread */ rd_kafka_broker_t * rd_kafka_broker_any_up (rd_kafka_t *rk, int *filtered_cnt, int (*filter) (rd_kafka_broker_t *rkb, void *opaque), void *opaque, const char *reason) { rd_kafka_broker_t *rkb; rkb = rd_kafka_broker_random0(__FUNCTION__, __LINE__, rk, rd_true/*is_up*/, -1, filtered_cnt, filter, opaque); if (!rkb && rk->rk_conf.sparse_connections) { /* Sparse connections: * If no eligible broker was found, schedule * a random broker for connecting. */ rd_kafka_connect_any(rk, reason); } return rkb; } /** * @brief Spend at most \p timeout_ms to acquire a usable (Up) broker. * * Prefers the most recently used broker, see rd_kafka_broker_weight_usable(). * * @param features (optional) Required broker features. * * @returns A probably usable broker with increased refcount, or NULL on timeout * @locks rd_kafka_*lock() if !do_lock * @locality any * * @sa rd_kafka_broker_any_up() */ rd_kafka_broker_t *rd_kafka_broker_any_usable (rd_kafka_t *rk, int timeout_ms, rd_dolock_t do_lock, int features, const char *reason) { 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); if (do_lock) rd_kafka_rdlock(rk); rkb = rd_kafka_broker_weighted(rk, rd_kafka_broker_weight_usable, features); if (!rkb && rk->rk_conf.sparse_connections) { /* Sparse connections: * If no eligible broker was found, schedule * a random broker for connecting. */ rd_kafka_connect_any(rk, reason); } 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 the broker handle for \p broker_id using cached metadata * information (if available) in state == \p state, * with refcount increaesd. * * Otherwise enqueues the \p eonce on the wait-state-change queue * which will be triggered on broker state changes. * It may also be triggered erroneously, so the caller * should call rd_kafka_broker_get_async() again when * the eonce is triggered. * * @locks none * @locality any thread */ rd_kafka_broker_t * rd_kafka_broker_get_async (rd_kafka_t *rk, int32_t broker_id, int state, rd_kafka_enq_once_t *eonce) { int version; do { rd_kafka_broker_t *rkb; version = rd_kafka_brokers_get_state_version(rk); rd_kafka_rdlock(rk); rkb = rd_kafka_broker_find_by_nodeid0(rk, broker_id, state, rd_true); rd_kafka_rdunlock(rk); if (rkb) return rkb; } while (!rd_kafka_brokers_wait_state_change_async(rk, version, eonce)); return NULL; /* eonce added to wait list */ } /** * @returns the current controller using cached metadata information, * and only if the broker's state == \p state. * The reference count is increased for the returned broker. * * @locks none * @locality any thread */ static rd_kafka_broker_t *rd_kafka_broker_controller_nowait (rd_kafka_t *rk, int state) { rd_kafka_broker_t *rkb; rd_kafka_rdlock(rk); if (rk->rk_controllerid == -1) { rd_kafka_rdunlock(rk); rd_kafka_metadata_refresh_brokers(rk, NULL, "lookup controller"); return NULL; } rkb = rd_kafka_broker_find_by_nodeid0(rk, rk->rk_controllerid, state, rd_true); rd_kafka_rdunlock(rk); return rkb; } /** * @returns the current controller using cached metadata information if * available in state == \p state, with refcount increaesd. * * Otherwise enqueues the \p eonce on the wait-controller queue * which will be triggered on controller updates or broker state * changes. It may also be triggered erroneously, so the caller * should call rd_kafka_broker_controller_async() again when * the eonce is triggered. * * @locks none * @locality any thread */ rd_kafka_broker_t * rd_kafka_broker_controller_async (rd_kafka_t *rk, int state, rd_kafka_enq_once_t *eonce) { int version; do { rd_kafka_broker_t *rkb; version = rd_kafka_brokers_get_state_version(rk); rkb = rd_kafka_broker_controller_nowait(rk, state); if (rkb) return rkb; } while (!rd_kafka_brokers_wait_state_change_async(rk, version, eonce)); return NULL; /* eonce added to wait list */ } /** * @returns the current controller using cached metadata information, * blocking up to \p abs_timeout for the controller to be known * and to reach state == \p state. The reference count is increased * for the returned broker. * * @locks none * @locality any thread */ rd_kafka_broker_t *rd_kafka_broker_controller (rd_kafka_t *rk, int state, rd_ts_t abs_timeout) { while (1) { int version = rd_kafka_brokers_get_state_version(rk); rd_kafka_broker_t *rkb; int remains_ms; rkb = rd_kafka_broker_controller_nowait(rk, state); if (rkb) return rkb; remains_ms = rd_timeout_remains(abs_timeout); if (rd_timeout_expired(remains_ms)) return NULL; rd_kafka_brokers_wait_state_change(rk, version, remains_ms); } } /** * 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; int log_decode_errors = LOG_ERR; 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_interceptors_on_response_received( rkb->rkb_rk, -1, rd_kafka_broker_name(rkb), rkb->rkb_nodeid, -1, -1, rkbuf->rkbuf_reshdr.CorrId, rkbuf->rkbuf_totlen, -1, RD_KAFKA_RESP_ERR__NOENT); 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); /* Copy request's header and certain flags to response object's * reqhdr for convenience. */ rkbuf->rkbuf_reqhdr = req->rkbuf_reqhdr; rkbuf->rkbuf_flags |= (req->rkbuf_flags & RD_KAFKA_BUF_FLAGS_RESP_COPY_MASK); rkbuf->rkbuf_ts_sent = req->rkbuf_ts_sent; /* copy rtt */ /* 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); /* In case of flexibleVersion, skip the response header tags. * The ApiVersion request/response is different since it needs * be backwards compatible and thus has no header tags. */ if (req->rkbuf_reqhdr.ApiKey != RD_KAFKAP_ApiVersion) rd_kafka_buf_skip_tags(rkbuf); 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; err_parse: rd_atomic64_add(&rkb->rkb_c.rx_err, 1); rd_kafka_buf_callback(rkb->rkb_rk, rkb, rkbuf->rkbuf_err, NULL, req); rd_kafka_buf_destroy(rkbuf); return -1; } 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; } rd_atomic64_set(&rkb->rkb_c.ts_recv, rd_clock()); 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: if (!strcmp(errstr, "Disconnected")) rd_kafka_broker_conn_closed(rkb, err, errstr); else rd_kafka_broker_fail(rkb, 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 if (fcntl(s, F_SETFD, FD_CLOEXEC, &on) == -1) fprintf(stderr, "WARNING: librdkafka: %s: " "fcntl(FD_CLOEXEC) failed: %s: ignoring\n", __FUNCTION__, rd_strerror(errno)); #endif return s; } /** * @brief Update the reconnect backoff. * Should be called when a connection is made, or all addresses * a broker resolves to has been exhausted without successful connect. * * @locality broker thread * @locks none */ static void rd_kafka_broker_update_reconnect_backoff (rd_kafka_broker_t *rkb, const rd_kafka_conf_t *conf, rd_ts_t now) { int backoff; /* If last connection attempt was more than reconnect.backoff.max.ms * ago, reset the reconnect backoff to the initial * reconnect.backoff.ms value. */ if (rkb->rkb_ts_reconnect + (conf->reconnect_backoff_max_ms * 1000) < now) rkb->rkb_reconnect_backoff_ms = conf->reconnect_backoff_ms; /* Apply -25%...+50% jitter to next backoff. */ backoff = rd_jitter((int)((float)rkb->rkb_reconnect_backoff_ms * 0.75), (int)((float)rkb->rkb_reconnect_backoff_ms * 1.5)); /* Cap to reconnect.backoff.max.ms. */ backoff = RD_MIN(backoff, conf->reconnect_backoff_max_ms); /* Set time of next reconnect */ rkb->rkb_ts_reconnect = now + (backoff * 1000); rkb->rkb_reconnect_backoff_ms = RD_MIN(rkb->rkb_reconnect_backoff_ms * 2, conf->reconnect_backoff_max_ms); } /** * @brief Calculate time until next reconnect attempt. * * @returns the number of milliseconds to the next connection attempt, or 0 * if immediate. * @locality broker thread * @locks none */ static RD_INLINE int rd_kafka_broker_reconnect_backoff (const rd_kafka_broker_t *rkb, rd_ts_t now) { rd_ts_t remains; if (unlikely(rkb->rkb_ts_reconnect == 0)) return 0; /* immediate */ remains = rkb->rkb_ts_reconnect - now; if (remains <= 0) return 0; /* immediate */ return (int)(remains / 1000); } /** * @brief Unittest for reconnect.backoff.ms */ static int rd_ut_reconnect_backoff (void) { rd_kafka_broker_t rkb = RD_ZERO_INIT; rd_kafka_conf_t conf = { .reconnect_backoff_ms = 10, .reconnect_backoff_max_ms = 90 }; rd_ts_t now = 1000000; int backoff; rkb.rkb_reconnect_backoff_ms = conf.reconnect_backoff_ms; /* broker's backoff is the initial reconnect.backoff.ms=10 */ rd_kafka_broker_update_reconnect_backoff(&rkb, &conf, now); backoff = rd_kafka_broker_reconnect_backoff(&rkb, now); RD_UT_ASSERT_RANGE(backoff, 7, 15, "%d"); /* .. 20 */ rd_kafka_broker_update_reconnect_backoff(&rkb, &conf, now); backoff = rd_kafka_broker_reconnect_backoff(&rkb, now); RD_UT_ASSERT_RANGE(backoff, 15, 30, "%d"); /* .. 40 */ rd_kafka_broker_update_reconnect_backoff(&rkb, &conf, now); backoff = rd_kafka_broker_reconnect_backoff(&rkb, now); RD_UT_ASSERT_RANGE(backoff, 30, 60, "%d"); /* .. 80, the jitter is capped at reconnect.backoff.max.ms=90 */ rd_kafka_broker_update_reconnect_backoff(&rkb, &conf, now); backoff = rd_kafka_broker_reconnect_backoff(&rkb, now); RD_UT_ASSERT_RANGE(backoff, 60, conf.reconnect_backoff_max_ms, "%d"); /* .. 90, capped by reconnect.backoff.max.ms */ rd_kafka_broker_update_reconnect_backoff(&rkb, &conf, now); backoff = rd_kafka_broker_reconnect_backoff(&rkb, now); RD_UT_ASSERT_RANGE(backoff, 67, conf.reconnect_backoff_max_ms, "%d"); /* .. 90, should remain at capped value. */ rd_kafka_broker_update_reconnect_backoff(&rkb, &conf, now); backoff = rd_kafka_broker_reconnect_backoff(&rkb, now); RD_UT_ASSERT_RANGE(backoff, 67, conf.reconnect_backoff_max_ms, "%d"); RD_UT_PASS(); } /** * @brief 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, 0 if broker does not have a hostname, or 1 * if the connection is now in progress. */ static int rd_kafka_broker_connect (rd_kafka_broker_t *rkb) { const rd_sockaddr_inx_t *sinx; char errstr[512]; char nodename[RD_KAFKA_NODENAME_SIZE]; rd_bool_t reset_cached_addr = rd_false; rd_rkb_dbg(rkb, BROKER, "CONNECT", "broker in state %s connecting", rd_kafka_broker_state_names[rkb->rkb_state]); rd_atomic32_add(&rkb->rkb_c.connects, 1); rd_kafka_broker_lock(rkb); rd_strlcpy(nodename, rkb->rkb_nodename, sizeof(nodename)); /* If the nodename was changed since the last connect, * reset the address cache. */ reset_cached_addr = (rkb->rkb_connect_epoch != rkb->rkb_nodename_epoch); rkb->rkb_connect_epoch = rkb->rkb_nodename_epoch; /* Logical brokers might not have a hostname set, in which case * we should not try to connect. */ if (*nodename) rd_kafka_broker_set_state(rkb, RD_KAFKA_BROKER_STATE_CONNECT); rd_kafka_broker_unlock(rkb); if (!*nodename) { rd_rkb_dbg(rkb, BROKER, "CONNECT", "broker has no address yet: postponing connect"); return 0; } rd_kafka_broker_update_reconnect_backoff(rkb, &rkb->rkb_rk->rk_conf, rd_clock()); if (rd_kafka_broker_resolve(rkb, nodename, reset_cached_addr) == -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)))) { rd_kafka_broker_fail(rkb, LOG_ERR, RD_KAFKA_RESP_ERR__TRANSPORT, "%s", errstr); return -1; } return 1; } /** * @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; 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, rd_false/*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); if (MechCnt < 0 || MechCnt > 100) rd_kafka_buf_parse_fail(rkbuf, "Invalid MechanismCount %"PRId32, MechCnt); /* Build a CSV string of supported mechanisms. */ msz = RD_MIN(511, 1 + (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) * * @locks_acquired rkb */ 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); } 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, (rkb->rkb_features & RD_KAFKA_FEATURE_SASL_AUTH_REQ) ? RD_KAFKA_BROKER_STATE_AUTH_REQ : RD_KAFKA_BROKER_STATE_AUTH_LEGACY); rd_kafka_broker_unlock(rkb); if (rd_kafka_sasl_client_new( rkb->rkb_transport, sasl_errstr, sizeof(sasl_errstr)) == -1) { rd_kafka_broker_fail( rkb, LOG_ERR, RD_KAFKA_RESP_ERR__AUTHENTICATION, "Failed to initialize " "SASL authentication: %s", sasl_errstr); return; } } 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_required rkb */ static void rd_kafka_broker_set_api_versions (rd_kafka_broker_t *rkb, struct rd_kafka_ApiVersion *apis, size_t api_cnt) { 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)); } /** * 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 = NULL; size_t api_cnt = 0; int16_t retry_ApiVersion = -1; if (err == RD_KAFKA_RESP_ERR__DESTROY) return; err = rd_kafka_handle_ApiVersion(rk, rkb, err, rkbuf, request, &apis, &api_cnt); /* Broker does not support our ApiVersionRequest version, * see if we can downgrade to an older version. */ if (err == RD_KAFKA_RESP_ERR_UNSUPPORTED_VERSION) { size_t i; /* Find the broker's highest supported version for * ApiVersionRequest and use that to retry. */ for (i = 0 ; i < api_cnt ; i++) { if (apis[i].ApiKey == RD_KAFKAP_ApiVersion) { retry_ApiVersion = RD_MIN( request->rkbuf_reqhdr.ApiVersion - 1, apis[i].MaxVer); break; } } /* Before v3 the broker would not return its supported * ApiVersionRequests, so we go straight for version 0. */ if (i == api_cnt && request->rkbuf_reqhdr.ApiVersion > 0) retry_ApiVersion = 0; } else if (err == RD_KAFKA_RESP_ERR_INVALID_REQUEST) { rd_rkb_log(rkb, LOG_ERR, "APIVERSION", "ApiVersionRequest v%hd failed due to " "invalid request: " "check client.software.name (\"%s\") and " "client.software.version (\"%s\") " "for invalid characters: " "falling back to older request version", request->rkbuf_reqhdr.ApiVersion, rk->rk_conf.sw_name, rk->rk_conf.sw_version); retry_ApiVersion = 0; } if (err && apis) rd_free(apis); if (retry_ApiVersion != -1) { /* Retry request with a lower version */ rd_rkb_dbg(rkb, BROKER|RD_KAFKA_DBG_FEATURE|RD_KAFKA_DBG_PROTOCOL, "APIVERSION", "ApiVersionRequest v%hd failed due to %s: " "retrying with v%hd", request->rkbuf_reqhdr.ApiVersion, rd_kafka_err2name(err), retry_ApiVersion); rd_kafka_ApiVersionRequest(rkb, retry_ApiVersion, RD_KAFKA_NO_REPLYQ, rd_kafka_broker_handle_ApiVersion, NULL); return; } if (err) { if (rkb->rkb_transport) rd_kafka_broker_fail( rkb, LOG_WARNING, RD_KAFKA_RESP_ERR__TRANSPORT, "ApiVersionRequest failed: %s: " "probably due to broker version < 0.10 " "(see api.version.request configuration)", rd_kafka_err2str(err)); return; } rd_kafka_broker_lock(rkb); rd_kafka_broker_set_api_versions(rkb, apis, api_cnt); rd_kafka_broker_unlock(rkb); rd_kafka_broker_connect_auth(rkb); } /** * Call when asynchronous connection attempt completes, either succesfully * (if errstr is NULL) or fails. * * @locks_acquired rkb * @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, 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_max_inflight = 1; /* Hold back other requests until * ApiVersion, SaslHandshake, etc * are done. */ rd_kafka_transport_poll_set(rkb->rkb_transport, POLLIN); rd_kafka_broker_lock(rkb); 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)) { /* Use configured broker.version.fallback to * figure out API versions. * In case broker.version.fallback indicates a version * that supports ApiVersionRequest it will update * rkb_features to have FEATURE_APIVERSION set which will * trigger an ApiVersionRequest below. */ rd_kafka_broker_set_api_versions(rkb, NULL, 0); } 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_set_state( rkb, RD_KAFKA_BROKER_STATE_APIVERSION_QUERY); rd_kafka_broker_unlock(rkb); rd_kafka_ApiVersionRequest( rkb, -1 /* Use highest version we support */, RD_KAFKA_NO_REPLYQ, rd_kafka_broker_handle_ApiVersion, NULL); } else { rd_kafka_broker_unlock(rkb); /* 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); rd_ts_t now; /* 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=false 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; now = rd_clock(); rd_atomic64_set(&rkb->rkb_c.ts_send, now); /* 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); rd_atomic64_add(&rkb->rkb_c.reqtype[rkbuf->rkbuf_reqhdr.ApiKey], 1); /* Notify transport layer of full request sent */ if (likely(rkb->rkb_transport != NULL)) rd_kafka_transport_request_sent(rkb, rkbuf); /* Entire buffer sent, unlink from outbuf */ rd_kafka_bufq_deq(&rkb->rkb_outbufs, rkbuf); rkbuf->rkbuf_flags |= RD_KAFKA_OP_F_SENT; /* Store time for RTT calculation */ rkbuf->rkbuf_ts_sent = now; /* Add to outbuf_latency averager */ rd_avg_add(&rkb->rkb_avg_outbuf_latency, rkbuf->rkbuf_ts_sent - rkbuf->rkbuf_ts_enq); 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, " "prev CorrId %"PRId32") in %dms", rd_kafka_ApiKey2str(rkbuf->rkbuf_reqhdr.ApiKey), rkbuf->rkbuf_reqhdr.ApiVersion, rd_slice_size(&rkbuf->rkbuf_reader), rkbuf->rkbuf_retries, rkbuf->rkbuf_max_retries, rkbuf->rkbuf_corrid, rkb->rkb_rk->rk_conf.retry_backoff_ms); rd_atomic64_add(&rkb->rkb_c.tx_retries, 1); rkbuf->rkbuf_ts_retry = rd_clock() + (rkb->rkb_rk->rk_conf.retry_backoff_ms * 1000); /* Precaution: time out the request if it hasn't moved from the * retry queue within the retry interval (such as when the broker is * down). */ // FIXME: implememt this properly. rkbuf->rkbuf_ts_timeout = rkbuf->rkbuf_ts_retry + (5*1000*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 *next_wakeup) { rd_ts_t now = rd_clock(); rd_kafka_buf_t *rkbuf; int cnt = 0; while ((rkbuf = TAILQ_FIRST(&rkb->rkb_retrybufs.rkbq_bufs))) { if (rkbuf->rkbuf_ts_retry > now) { if (rkbuf->rkbuf_ts_retry < *next_wakeup) *next_wakeup = rkbuf->rkbuf_ts_retry; break; } rd_kafka_bufq_deq(&rkb->rkb_retrybufs, rkbuf); rd_kafka_broker_buf_enq0(rkb, rkbuf); cnt++; } if (cnt > 0) rd_rkb_dbg(rkb, BROKER, "RETRY", "Moved %d retry buffer(s) to output queue", cnt); } /** * @brief Propagate delivery report for entire message queue. * * @param err The error which will be set on each message. * @param status The status which will be set on each message. * * To avoid extra iterations, the \p err and \p status are set on * the message as they are popped off the OP_DR msgq in rd_kafka_poll() et.al */ void rd_kafka_dr_msgq (rd_kafka_topic_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; if (err && rd_kafka_is_transactional(rk)) rd_atomic64_add(&rk->rk_eos.txn_dr_fails, rd_kafka_msgq_len(rkmq)); /* Call on_acknowledgement() interceptors */ rd_kafka_interceptors_on_acknowledgement_queue(rk, rkmq, err); if (rk->rk_drmode != RD_KAFKA_DR_MODE_NONE && (!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.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 Trigger delivery reports for implicitly acked messages. * * @locks none * @locality broker thread - either last or current leader */ void rd_kafka_dr_implicit_ack (rd_kafka_broker_t *rkb, rd_kafka_toppar_t *rktp, uint64_t last_msgid) { rd_kafka_msgq_t acked = RD_KAFKA_MSGQ_INITIALIZER(acked); rd_kafka_msgq_t acked2 = RD_KAFKA_MSGQ_INITIALIZER(acked2); rd_kafka_msg_status_t status = RD_KAFKA_MSG_STATUS_POSSIBLY_PERSISTED; if (rktp->rktp_rkt->rkt_conf.required_acks != 0) status = RD_KAFKA_MSG_STATUS_PERSISTED; rd_kafka_msgq_move_acked(&acked, &rktp->rktp_xmit_msgq, last_msgid, status); rd_kafka_msgq_move_acked(&acked2, &rktp->rktp_msgq, last_msgid, status); /* Insert acked2 into acked in correct order */ rd_kafka_msgq_insert_msgq(&acked, &acked2, rktp->rktp_rkt->rkt_conf.msg_order_cmp); if (!rd_kafka_msgq_len(&acked)) return; rd_rkb_dbg(rkb, MSG|RD_KAFKA_DBG_EOS, "IMPLICITACK", "%.*s [%"PRId32"] %d message(s) implicitly acked " "by subsequent batch success " "(msgids %"PRIu64"..%"PRIu64", " "last acked %"PRIu64")", RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic), rktp->rktp_partition, rd_kafka_msgq_len(&acked), rd_kafka_msgq_first(&acked)->rkm_u.producer.msgid, rd_kafka_msgq_last(&acked)->rkm_u.producer.msgid, last_msgid); /* Trigger delivery reports */ rd_kafka_dr_msgq(rktp->rktp_rkt, &acked, RD_KAFKA_RESP_ERR_NO_ERROR); } /** * @brief Map existing partitions to this broker using the * toppar's leader_id. Only undelegated partitions * matching this broker are mapped. * * @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_topic_t *rkt; int cnt = 0; if (rkb->rkb_nodeid == -1 || RD_KAFKA_BROKER_IS_LOGICAL(rkb)) 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++) { rd_kafka_toppar_t *rktp = rkt->rkt_p[i]; /* Only map undelegated partitions matching this broker*/ rd_kafka_toppar_lock(rktp); if (rktp->rktp_leader_id == rkb->rkb_nodeid && !(rktp->rktp_broker && rktp->rktp_next_broker)) { rd_kafka_toppar_broker_update( rktp, rktp->rktp_leader_id, rkb, "broker node information updated"); 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 RD_CMP(a->rkb_nodeid, b->rkb_nodeid); } /** * @brief Set the broker logname (used in logs) to a copy of \p logname. * * @locality any * @locks none */ static void rd_kafka_broker_set_logname (rd_kafka_broker_t *rkb, const char *logname) { mtx_lock(&rkb->rkb_logname_lock); if (rkb->rkb_logname) rd_free(rkb->rkb_logname); rkb->rkb_logname = rd_strdup(logname); mtx_unlock(&rkb->rkb_logname_lock); } /** * @brief Prepare destruction of the broker object. * * Since rd_kafka_broker_terminating() relies on the refcnt of the * broker to reach 1, we need to loose any self-references * to avoid a hang (waiting for refcnt decrease) on destruction. * * @locality broker thread * @locks none */ static void rd_kafka_broker_prepare_destroy (rd_kafka_broker_t *rkb) { rd_kafka_broker_monitor_del(&rkb->rkb_coord_monitor); } /** * @brief Serve a broker op (an op posted by another thread to be handled by * this broker's thread). * * @returns true if calling op loop should break out, else false to continue. * @locality broker thread * @locks none */ static RD_WARN_UNUSED_RESULT rd_bool_t rd_kafka_broker_op_serve (rd_kafka_broker_t *rkb, rd_kafka_op_t *rko) { rd_kafka_toppar_t *rktp; rd_kafka_resp_err_t topic_err; rd_bool_t wakeup = rd_false; 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); rd_strlcpy(rkb->rkb_nodename, rko->rko_u.node.nodename, sizeof(rkb->rkb_nodename)); rkb->rkb_nodename_epoch++; updated |= _UPD_NAME; } if (rko->rko_u.node.nodeid != -1 && !RD_KAFKA_BROKER_IS_LOGICAL(rkb) && 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 system thread name */ rd_kafka_set_thread_sysname("rdk:broker%"PRId32, rkb->rkb_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. */ rd_kafka_broker_set_logname(rkb, brokername); rd_rkb_dbg(rkb, BROKER, "UPDATE", "Name changed from %s to %s", rkb->rkb_name, brokername); rd_strlcpy(rkb->rkb_name, brokername, sizeof(rkb->rkb_name)); } rd_kafka_broker_unlock(rkb); rd_kafka_wrunlock(rkb->rkb_rk); if (updated & _UPD_NAME) rd_kafka_broker_fail(rkb, LOG_DEBUG, RD_KAFKA_RESP_ERR__TRANSPORT, "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); } rd_kafka_brokers_broadcast_state_change(rkb->rkb_rk); 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 = 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_broker); rktp->rktp_next_broker = NULL; rd_kafka_toppar_unlock(rktp); break; } /* See if we are still the next broker */ if (rktp->rktp_next_broker != rkb) { rd_rkb_dbg(rkb, BROKER | RD_KAFKA_DBG_TOPIC, "TOPBRK", "Topic %s [%"PRId32"]: not joining broker " "(next broker %s)", rktp->rktp_rkt->rkt_topic->str, rktp->rktp_partition, rktp->rktp_next_broker ? rd_kafka_broker_name(rktp->rktp_next_broker): "(none)"); /* Need temporary refcount so we can safely unlock * after q_enq(). */ rd_kafka_toppar_keep(rktp); /* No, forward this op to the new next broker. */ rd_kafka_q_enq(rktp->rktp_next_broker->rkb_ops, rko); rko = NULL; rd_kafka_toppar_unlock(rktp); rd_kafka_toppar_destroy(rktp); break; } rd_rkb_dbg(rkb, BROKER | RD_KAFKA_DBG_TOPIC, "TOPBRK", "Topic %s [%"PRId32"]: joining broker " "(rktp %p, %d message(s) queued)", rktp->rktp_rkt->rkt_topic->str, rktp->rktp_partition, rktp, rd_kafka_msgq_len(&rktp->rktp_msgq)); rd_kafka_assert(NULL, !(rktp->rktp_flags & RD_KAFKA_TOPPAR_F_ON_RKB)); rktp->rktp_flags |= RD_KAFKA_TOPPAR_F_ON_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_broker = rkb; rd_assert(!rktp->rktp_msgq_wakeup_q); rktp->rktp_msgq_wakeup_q = rd_kafka_q_keep(rkb->rkb_ops); rd_kafka_broker_keep(rkb); if (rkb->rkb_rk->rk_type == RD_KAFKA_PRODUCER) { rd_kafka_broker_active_toppar_add(rkb, rktp, "joining"); if (rd_kafka_is_idempotent(rkb->rkb_rk)) { /* Wait for all outstanding requests from * the previous leader to finish before * producing anything to this new leader. */ rd_kafka_idemp_drain_toppar( rktp, "wait for outstanding requests to " "finish before producing to " "new leader"); } } rd_kafka_broker_destroy(rktp->rktp_next_broker); rktp->rktp_next_broker = 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 = rko->rko_rktp; /* If there is a topic-wide error, use it as error code * when failing messages below. */ topic_err = rd_kafka_topic_get_error(rktp->rktp_rkt); 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_broker != rkb)) { rd_rkb_dbg(rkb, BROKER | RD_KAFKA_DBG_TOPIC, "TOPBRK", "Topic %s [%"PRId32"]: " "ignoring PARTITION_LEAVE: " "not delegated to broker (%s)", rktp->rktp_rkt->rkt_topic->str, rktp->rktp_partition, rktp->rktp_broker ? rd_kafka_broker_name(rktp->rktp_broker) : "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*/); if (rkb->rkb_rk->rk_type == RD_KAFKA_PRODUCER) { /* Purge any ProduceRequests for this toppar * in the output queue. */ rd_kafka_broker_bufq_purge_by_toppar( rkb, &rkb->rkb_outbufs, RD_KAFKAP_Produce, rktp, RD_KAFKA_RESP_ERR__RETRY); } 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 broker %s, rktp %p)", rktp->rktp_rkt->rkt_topic->str, rktp->rktp_partition, rd_kafka_msgq_len(&rktp->rktp_xmit_msgq), rktp->rktp_next_broker ? rd_kafka_broker_name(rktp->rktp_next_broker) : "(none)", rktp); /* Insert xmitq(broker-local) messages to the msgq(global) * at their sorted position to maintain ordering. */ rd_kafka_msgq_insert_msgq(&rktp->rktp_msgq, &rktp->rktp_xmit_msgq, rktp->rktp_rkt->rkt_conf. msg_order_cmp); if (rkb->rkb_rk->rk_type == RD_KAFKA_PRODUCER) rd_kafka_broker_active_toppar_del(rkb, rktp, "leaving"); 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_broker); if (rktp->rktp_msgq_wakeup_q) { rd_kafka_q_destroy(rktp->rktp_msgq_wakeup_q); rktp->rktp_msgq_wakeup_q = NULL; } rktp->rktp_broker = NULL; rd_assert(rktp->rktp_flags & RD_KAFKA_TOPPAR_F_ON_RKB); rktp->rktp_flags &= ~RD_KAFKA_TOPPAR_F_ON_RKB; if (rktp->rktp_next_broker) { /* There is a next broker we need to migrate to. */ rko->rko_type = RD_KAFKA_OP_PARTITION_JOIN; rd_kafka_q_enq(rktp->rktp_next_broker->rkb_ops, rko); rko = NULL; } else { rd_rkb_dbg(rkb, BROKER | RD_KAFKA_DBG_TOPIC, "TOPBRK", "Topic %s [%"PRId32"]: no next broker, " "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 : (topic_err ? topic_err : RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION)); } rd_kafka_toppar_unlock(rktp); rd_kafka_toppar_destroy(rktp); /* from JOIN */ rd_kafka_brokers_broadcast_state_change(rkb->rkb_rk); break; case RD_KAFKA_OP_TERMINATE: /* nop: just a wake-up. */ rd_rkb_dbg(rkb, BROKER, "TERM", "Received TERMINATE op in state %s: " "%d refcnts, %d toppar(s), %d active 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_active_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)); /* Expedite termination by bringing down the broker * and trigger a state change. * This makes sure any eonce dependent on state changes * are triggered. */ rd_kafka_broker_fail(rkb, LOG_DEBUG, RD_KAFKA_RESP_ERR__DESTROY, "Client is terminating"); rd_kafka_broker_prepare_destroy(rkb); wakeup = rd_true; break; case RD_KAFKA_OP_WAKEUP: wakeup = rd_true; break; case RD_KAFKA_OP_PURGE: rd_kafka_broker_handle_purge_queues(rkb, rko); rko = NULL; /* the rko is reused for the reply */ break; case RD_KAFKA_OP_CONNECT: /* Sparse connections: connection requested, transition * to TRY_CONNECT state to trigger new connection. */ if (rkb->rkb_state == RD_KAFKA_BROKER_STATE_INIT) { rd_rkb_dbg(rkb, BROKER, "CONNECT", "Received CONNECT op"); rkb->rkb_persistconn.internal++; rd_kafka_broker_lock(rkb); rd_kafka_broker_set_state( rkb, RD_KAFKA_BROKER_STATE_TRY_CONNECT); rd_kafka_broker_unlock(rkb); } else if (rkb->rkb_state >= RD_KAFKA_BROKER_STATE_TRY_CONNECT) { rd_bool_t do_disconnect = rd_false; /* If the nodename was changed since the last connect, * close the current connection. */ rd_kafka_broker_lock(rkb); do_disconnect = (rkb->rkb_connect_epoch != rkb->rkb_nodename_epoch); rd_kafka_broker_unlock(rkb); if (do_disconnect) rd_kafka_broker_fail( rkb, LOG_DEBUG, RD_KAFKA_RESP_ERR__TRANSPORT, "Closing connection due to " "nodename change"); } /* Expedite next reconnect */ rkb->rkb_ts_reconnect = 0; wakeup = rd_true; break; default: rd_kafka_assert(rkb->rkb_rk, !*"unhandled op type"); break; } if (rko) rd_kafka_op_reply(rko, RD_KAFKA_RESP_ERR_NO_ERROR); return wakeup; } /** * @brief Serve broker ops. * @returns the number of ops served */ static RD_WARN_UNUSED_RESULT int rd_kafka_broker_ops_serve (rd_kafka_broker_t *rkb, rd_ts_t timeout_us) { rd_kafka_op_t *rko; int cnt = 0; while ((rko = rd_kafka_q_pop(rkb->rkb_ops, timeout_us, 0)) && (cnt++, !rd_kafka_broker_op_serve(rkb, rko))) timeout_us = RD_POLL_NOWAIT; return cnt; } /** * @brief Serve broker ops and IOs. * * If a connection exists, poll IO first based on timeout. * Use remaining timeout for ops queue poll. * * If no connection, poll ops queue using timeout. * * Sparse connections: if there's need for a connection, set * timeout to NOWAIT. * * @param abs_timeout Maximum block time (absolute time). * * @returns true on wakeup (broker state machine needs to be served), * else false. * * @locality broker thread * @locks none */ static RD_WARN_UNUSED_RESULT rd_bool_t rd_kafka_broker_ops_io_serve (rd_kafka_broker_t *rkb, rd_ts_t abs_timeout) { rd_ts_t now; rd_bool_t wakeup; if (unlikely(rd_kafka_terminating(rkb->rkb_rk))) abs_timeout = rd_clock() + 1000; else if (unlikely(rd_kafka_broker_needs_connection(rkb))) abs_timeout = RD_POLL_NOWAIT; else if (unlikely(abs_timeout == RD_POLL_INFINITE)) abs_timeout = rd_clock() + ((rd_ts_t)rd_kafka_max_block_ms * 1000); if (likely(rkb->rkb_transport != NULL)) { /* Serve IO events. * * If there are IO events, cut out the queue ops_serve * timeout (below) since we'll probably have to perform more * duties based on the IO. * IO polling granularity is milliseconds while * queue granularity is microseconds. */ if (rd_kafka_transport_io_serve( rkb->rkb_transport, rd_timeout_remains(abs_timeout))) abs_timeout = RD_POLL_NOWAIT; } /* Serve broker ops */ wakeup = rd_kafka_broker_ops_serve(rkb, rd_timeout_remains_us(abs_timeout)); /* An op might have triggered the need for a connection, if so * transition to TRY_CONNECT state. */ if (unlikely(rd_kafka_broker_needs_connection(rkb) && rkb->rkb_state == RD_KAFKA_BROKER_STATE_INIT)) { rd_kafka_broker_lock(rkb); rd_kafka_broker_set_state( rkb, RD_KAFKA_BROKER_STATE_TRY_CONNECT); rd_kafka_broker_unlock(rkb); wakeup = rd_true; } /* Scan queues for timeouts. */ now = rd_clock(); if (rd_interval(&rkb->rkb_timeout_scan_intvl, 1000000, now) > 0) rd_kafka_broker_timeout_scan(rkb, now); return wakeup; } /** * @brief Consumer: Serve the toppars 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_consumer_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; } /** * @brief Scan toppar's xmit and producer queue for message timeouts and * enqueue delivery reports for timed out messages. * * @param abs_next_timeout will be set to the next message timeout, or 0 * if no timeout. * * @returns the number of messages timed out. * * @locality toppar's broker handler thread * @locks toppar_lock MUST be held */ static int rd_kafka_broker_toppar_msgq_scan (rd_kafka_broker_t *rkb, rd_kafka_toppar_t *rktp, rd_ts_t now, rd_ts_t *abs_next_timeout) { rd_kafka_msgq_t xtimedout = RD_KAFKA_MSGQ_INITIALIZER(xtimedout); rd_kafka_msgq_t qtimedout = RD_KAFKA_MSGQ_INITIALIZER(qtimedout); int xcnt, qcnt, cnt; uint64_t first, last; rd_ts_t next; *abs_next_timeout = 0; xcnt = rd_kafka_msgq_age_scan(rktp, &rktp->rktp_xmit_msgq, &xtimedout, now, &next); if (next && next < *abs_next_timeout) *abs_next_timeout = next; qcnt = rd_kafka_msgq_age_scan(rktp, &rktp->rktp_msgq, &qtimedout, now, &next); if (next && (!*abs_next_timeout || next < *abs_next_timeout)) *abs_next_timeout = next; cnt = xcnt + qcnt; if (likely(cnt == 0)) return 0; /* Insert queue-timedout into xmitqueue-timedout in a sorted fashion */ rd_kafka_msgq_insert_msgq(&xtimedout, &qtimedout, rktp->rktp_rkt->rkt_conf.msg_order_cmp); first = rd_kafka_msgq_first(&xtimedout)->rkm_u.producer.msgid; last = rd_kafka_msgq_last(&xtimedout)->rkm_u.producer.msgid; rd_rkb_dbg(rkb, MSG, "TIMEOUT", "%s [%"PRId32"]: timed out %d+%d message(s) " "(MsgId %"PRIu64"..%"PRIu64"): message.timeout.ms exceeded", rktp->rktp_rkt->rkt_topic->str, rktp->rktp_partition, xcnt, qcnt, first, last); /* Trigger delivery report for timed out messages */ rd_kafka_dr_msgq(rktp->rktp_rkt, &xtimedout, RD_KAFKA_RESP_ERR__MSG_TIMED_OUT); return cnt; } /** * @brief Producer: Check this broker's toppars for message timeouts. * * This is only used by the internal broker to enforce message timeouts. * * @returns the next absolute scan time. * * @locality internal broker thread. */ static rd_ts_t rd_kafka_broker_toppars_timeout_scan (rd_kafka_broker_t *rkb, rd_ts_t now) { rd_kafka_toppar_t *rktp; rd_ts_t next = now + (1000*1000); TAILQ_FOREACH(rktp, &rkb->rkb_toppars, rktp_rkblink) { rd_ts_t this_next; rd_kafka_toppar_lock(rktp); if (unlikely(rktp->rktp_broker != rkb)) { /* Currently migrating away from this * broker. */ rd_kafka_toppar_unlock(rktp); continue; } /* Scan queues for msg timeouts */ rd_kafka_broker_toppar_msgq_scan(rkb, rktp, now, &this_next); rd_kafka_toppar_unlock(rktp); if (this_next && this_next < next) next = this_next; } return next; } /** * @brief Idle function for the internal broker handle. */ static void rd_kafka_broker_internal_serve (rd_kafka_broker_t *rkb, rd_ts_t abs_timeout) { int initial_state = rkb->rkb_state; rd_bool_t wakeup; if (rkb->rkb_rk->rk_type == RD_KAFKA_CONSUMER) { /* Consumer */ do { rd_kafka_broker_consumer_toppars_serve(rkb); wakeup = rd_kafka_broker_ops_io_serve(rkb, abs_timeout); } while (!rd_kafka_broker_terminating(rkb) && (int)rkb->rkb_state == initial_state && !wakeup && !rd_timeout_expired(rd_timeout_remains(abs_timeout))); } else { /* Producer */ rd_ts_t next_timeout_scan = 0; do { rd_ts_t now = rd_clock(); if (now >= next_timeout_scan) next_timeout_scan = rd_kafka_broker_toppars_timeout_scan( rkb, now); wakeup = rd_kafka_broker_ops_io_serve( rkb, RD_MIN(abs_timeout, next_timeout_scan)); } while (!rd_kafka_broker_terminating(rkb) && (int)rkb->rkb_state == initial_state && !wakeup && !rd_timeout_expired(rd_timeout_remains(abs_timeout))); } } /** * @returns the number of requests that may be enqueued before * queue.backpressure.threshold is reached. */ static RD_INLINE unsigned int rd_kafka_broker_outbufs_space (rd_kafka_broker_t *rkb) { int r = rkb->rkb_rk->rk_conf.queue_backpressure_thres - rd_atomic32_get(&rkb->rkb_outbufs.rkbq_cnt); return r < 0 ? 0 : (unsigned int)r; } /** * @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. * @param do_timeout_scan perform msg timeout scan * @param may_send if set to false there is something on the global level * that prohibits sending messages, such as a transactional * state. * * @returns the number of messages produced. * * @locks none * @locality broker thread */ static int rd_kafka_toppar_producer_serve (rd_kafka_broker_t *rkb, rd_kafka_toppar_t *rktp, const rd_kafka_pid_t pid, rd_ts_t now, rd_ts_t *next_wakeup, rd_bool_t do_timeout_scan, rd_bool_t may_send) { int cnt = 0; int r; rd_kafka_msg_t *rkm; int move_cnt = 0; int max_requests; int reqcnt; int inflight = 0; uint64_t epoch_base_msgid = 0; /* By limiting the number of not-yet-sent buffers (rkb_outbufs) we * provide a backpressure mechanism to the producer loop * which allows larger message batches to accumulate and thus * increase throughput. * This comes at no latency cost since there are already * buffers enqueued waiting for transmission. */ max_requests = rd_kafka_broker_outbufs_space(rkb); rd_kafka_toppar_lock(rktp); if (unlikely(rktp->rktp_broker != rkb)) { /* Currently migrating away from this * broker. */ rd_kafka_toppar_unlock(rktp); return 0; } if (unlikely(do_timeout_scan)) { int timeoutcnt; rd_ts_t next; /* Scan queues for msg timeouts */ timeoutcnt = rd_kafka_broker_toppar_msgq_scan(rkb, rktp, now, &next); if (next && next < *next_wakeup) *next_wakeup = next; if (rd_kafka_is_idempotent(rkb->rkb_rk)) { if (!rd_kafka_pid_valid(pid)) { /* If we don't have a PID, we can't transmit * any messages. */ rd_kafka_toppar_unlock(rktp); return 0; } else if (timeoutcnt > 0) { /* Message timeouts will lead to gaps the in * the message sequence and thus trigger * OutOfOrderSequence errors from the broker. * Bump the epoch to reset the base msgid after * draining all partitions. */ /* Must not hold toppar lock */ rd_kafka_toppar_unlock(rktp); rd_kafka_idemp_drain_epoch_bump( rkb->rkb_rk, "%d message(s) timed out " "on %s [%"PRId32"]", timeoutcnt, rktp->rktp_rkt->rkt_topic->str, rktp->rktp_partition); return 0; } } } if (unlikely(!may_send)) { /* Sends prohibited on the broker or instance level */ max_requests = 0; } else if (unlikely(rd_kafka_fatal_error_code(rkb->rkb_rk))) { /* Fatal error has been raised, don't produce. */ max_requests = 0; } else if (unlikely(RD_KAFKA_TOPPAR_IS_PAUSED(rktp))) { /* Partition is paused */ max_requests = 0; } else if (unlikely(rd_kafka_is_transactional(rkb->rkb_rk) && !rd_kafka_txn_toppar_may_send_msg(rktp))) { /* Partition not registered in transaction yet */ max_requests = 0; } else if (max_requests > 0) { /* Move messages from locked partition produce queue * to broker-local xmit queue. */ if ((move_cnt = rktp->rktp_msgq.rkmq_msg_cnt) > 0) rd_kafka_msgq_insert_msgq(&rktp->rktp_xmit_msgq, &rktp->rktp_msgq, rktp->rktp_rkt->rkt_conf. msg_order_cmp); } rd_kafka_toppar_unlock(rktp); if (rd_kafka_is_idempotent(rkb->rkb_rk)) { /* Update the partition's cached PID, and reset the * base msg sequence if necessary */ rd_bool_t did_purge = rd_false; if (unlikely(!rd_kafka_pid_eq(pid, rktp->rktp_eos.pid))) { /* Flush any ProduceRequests for this partition in the * output buffer queue to speed up recovery. */ rd_kafka_broker_bufq_purge_by_toppar( rkb, &rkb->rkb_outbufs, RD_KAFKAP_Produce, rktp, RD_KAFKA_RESP_ERR__RETRY); did_purge = rd_true; if (rd_kafka_pid_valid(rktp->rktp_eos.pid)) rd_rkb_dbg(rkb, QUEUE, "TOPPAR", "%.*s [%"PRId32"] PID has changed: " "must drain requests for all " "partitions before resuming reset " "of PID", RD_KAFKAP_STR_PR(rktp->rktp_rkt-> rkt_topic), rktp->rktp_partition); } inflight = rd_atomic32_get(&rktp->rktp_msgs_inflight); if (unlikely(rktp->rktp_eos.wait_drain)) { if (inflight) { /* Waiting for in-flight requests to * drain/finish before producing anything more. * This is used to recover to a consistent * state when the partition leader * has changed, or timed out messages * have been removed from the queue. */ rd_rkb_dbg(rkb, QUEUE, "TOPPAR", "%.*s [%"PRId32"] waiting for " "%d in-flight request(s) to drain " "from queue before continuing " "to produce", RD_KAFKAP_STR_PR(rktp->rktp_rkt-> rkt_topic), rktp->rktp_partition, inflight); /* Flush any ProduceRequests for this * partition in the output buffer queue to * speed up draining. */ if (!did_purge) rd_kafka_broker_bufq_purge_by_toppar( rkb, &rkb->rkb_outbufs, RD_KAFKAP_Produce, rktp, RD_KAFKA_RESP_ERR__RETRY); return 0; } rd_rkb_dbg(rkb, QUEUE, "TOPPAR", "%.*s [%"PRId32"] all in-flight requests " "drained from queue", RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic), rktp->rktp_partition); rktp->rktp_eos.wait_drain = rd_false; } /* Limit the number of in-flight requests (per partition) * to the broker's sequence de-duplication window. */ max_requests = RD_MIN(max_requests, RD_KAFKA_IDEMP_MAX_INFLIGHT - inflight); } /* Check if allowed to create and enqueue a ProduceRequest */ if (max_requests <= 0) return 0; r = rktp->rktp_xmit_msgq.rkmq_msg_cnt; if (r == 0) return 0; rd_kafka_msgq_verify_order(rktp, &rktp->rktp_xmit_msgq, 0, rd_false); rd_rkb_dbg(rkb, QUEUE, "TOPPAR", "%.*s [%"PRId32"] %d message(s) in " "xmit queue (%d added from partition queue)", RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic), rktp->rktp_partition, r, move_cnt); rkm = TAILQ_FIRST(&rktp->rktp_xmit_msgq.rkmq_msgs); rd_dassert(rkm != NULL); if (rd_kafka_is_idempotent(rkb->rkb_rk)) { /* Update the partition's cached PID, and reset the * base msg sequence if necessary */ if (unlikely(!rd_kafka_pid_eq(pid, rktp->rktp_eos.pid))) { /* Attempt to change the pid, it will fail if there * are outstanding messages in-flight, in which case * we eventually come back here to retry. */ if (!rd_kafka_toppar_pid_change( rktp, pid, rkm->rkm_u.producer.msgid)) return 0; } rd_kafka_toppar_lock(rktp); /* Idempotent producer epoch base msgid, this is passed to the * ProduceRequest and msgset writer to adjust the protocol-level * per-message sequence number. */ epoch_base_msgid = rktp->rktp_eos.epoch_base_msgid; rd_kafka_toppar_unlock(rktp); } if (unlikely(rkb->rkb_state != RD_KAFKA_BROKER_STATE_UP)) { /* There are messages to send but connection is not up. */ rd_rkb_dbg(rkb, BROKER, "TOPPAR", "%.*s [%"PRId32"] " "%d message(s) queued but broker not up", RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic), rktp->rktp_partition, r); rkb->rkb_persistconn.internal++; return 0; } /* Attempt to fill the batch size, but limit our waiting * to queue.buffering.max.ms, batch.num.messages, and batch.size. */ if (r < rkb->rkb_rk->rk_conf.batch_num_messages && rktp->rktp_xmit_msgq.rkmq_msg_bytes < (int64_t)rkb->rkb_rk->rk_conf.batch_size) { rd_ts_t wait_max; /* Calculate maximum wait-time to honour * queue.buffering.max.ms contract. */ wait_max = rd_kafka_msg_enq_time(rkm) + rkb->rkb_rk->rk_conf.buffering_max_us; if (wait_max > now) { /* Wait for more messages or queue.buffering.max.ms * to expire. */ if (wait_max < *next_wakeup) *next_wakeup = wait_max; return 0; } } /* Honour retry.backoff.ms. */ if (unlikely(rkm->rkm_u.producer.ts_backoff > now)) { if (rkm->rkm_u.producer.ts_backoff < *next_wakeup) *next_wakeup = rkm->rkm_u.producer.ts_backoff; /* Wait for backoff to expire */ return 0; } /* Send Produce requests for this toppar, honouring the * queue backpressure threshold. */ for (reqcnt = 0 ; reqcnt < max_requests ; reqcnt++) { r = rd_kafka_ProduceRequest(rkb, rktp, pid, epoch_base_msgid); if (likely(r > 0)) cnt += r; else break; } /* If there are messages still in the queue, make the next * wakeup immediate. */ if (rd_kafka_msgq_len(&rktp->rktp_xmit_msgq) > 0) *next_wakeup = now; return cnt; } /** * @brief Produce from all toppars assigned to this broker. * * @param next_wakeup is updated if the next IO/ops timeout should be * less than the input value. * * @returns the total number of messages produced. */ static int rd_kafka_broker_produce_toppars (rd_kafka_broker_t *rkb, rd_ts_t now, rd_ts_t *next_wakeup, rd_bool_t do_timeout_scan) { rd_kafka_toppar_t *rktp; int cnt = 0; rd_ts_t ret_next_wakeup = *next_wakeup; rd_kafka_pid_t pid = RD_KAFKA_PID_INITIALIZER; rd_bool_t may_send = rd_true; /* Round-robin serve each toppar. */ rktp = rkb->rkb_active_toppar_next; if (unlikely(!rktp)) return 0; if (rd_kafka_is_idempotent(rkb->rkb_rk)) { /* Idempotent producer: get a copy of the current pid. */ pid = rd_kafka_idemp_get_pid(rkb->rkb_rk); /* If we don't have a valid pid, or the transaction state * prohibits sending messages, return immedatiely, * unless the per-partition timeout scan needs to run. * The broker threads are woken up when a PID is acquired * or the transaction state changes. */ if (!rd_kafka_pid_valid(pid)) may_send = rd_false; else if (rd_kafka_is_transactional(rkb->rkb_rk) && !rd_kafka_txn_may_send_msg(rkb->rkb_rk)) may_send = rd_false; if (!may_send && !do_timeout_scan) return 0; } do { rd_ts_t this_next_wakeup = ret_next_wakeup; /* Try producing toppar */ cnt += rd_kafka_toppar_producer_serve( rkb, rktp, pid, now, &this_next_wakeup, do_timeout_scan, may_send); if (this_next_wakeup < ret_next_wakeup) ret_next_wakeup = this_next_wakeup; } while ((rktp = CIRCLEQ_LOOP_NEXT(&rkb-> rkb_active_toppars, rktp, rktp_activelink)) != rkb->rkb_active_toppar_next); /* Update next starting toppar to produce in round-robin list. */ rd_kafka_broker_active_toppar_next( rkb, CIRCLEQ_LOOP_NEXT(&rkb->rkb_active_toppars, rktp, rktp_activelink)); *next_wakeup = ret_next_wakeup; return cnt; } /** * @brief Producer serving */ static void rd_kafka_broker_producer_serve (rd_kafka_broker_t *rkb, rd_ts_t abs_timeout) { rd_interval_t timeout_scan; unsigned int initial_state = rkb->rkb_state; rd_ts_t now; int cnt = 0; 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 == initial_state && (abs_timeout > (now = rd_clock()))) { rd_bool_t do_timeout_scan; rd_ts_t next_wakeup = abs_timeout; rd_kafka_broker_unlock(rkb); /* Perform timeout scan on first iteration, thus * on each state change, to make sure messages in * partition rktp_xmit_msgq are timed out before * being attempted to re-transmit. */ do_timeout_scan = cnt++ == 0 || rd_interval(&timeout_scan, 1000*1000, now) >= 0; rd_kafka_broker_produce_toppars(rkb, now, &next_wakeup, do_timeout_scan); /* Check and move retry buffers */ if (unlikely(rd_atomic32_get(&rkb->rkb_retrybufs.rkbq_cnt) > 0)) rd_kafka_broker_retry_bufs_move(rkb, &next_wakeup); if (rd_kafka_broker_ops_io_serve(rkb, next_wakeup)) return; /* 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; /* Don't back off on reaching end of partition */ if (err == RD_KAFKA_RESP_ERR__PARTITION_EOF) return; /* Certain errors that may require manual intervention should have * a longer backoff time. */ if (err == RD_KAFKA_RESP_ERR_TOPIC_AUTHORIZATION_FAILED) backoff_ms = RD_MAX(1000, backoff_ms * 10); rktp->rktp_ts_fetch_backoff = rd_clock() + (backoff_ms * 1000); rd_rkb_dbg(rkb, FETCH, "BACKOFF", "%s [%"PRId32"]: Fetch backoff for %dms%s%s", rktp->rktp_rkt->rkt_topic->str, rktp->rktp_partition, backoff_ms, err ? ": " : "", err ? rd_kafka_err2str(err) : ""); } /** * @brief Handle preferred replica in fetch response. * * @locks rd_kafka_toppar_lock(rktp) and * rd_kafka_rdlock(rk) must NOT be held. * * @locality broker thread */ static void rd_kafka_fetch_preferred_replica_handle (rd_kafka_toppar_t *rktp, rd_kafka_buf_t *rkbuf, rd_kafka_broker_t *rkb, int32_t preferred_id) { const rd_ts_t one_minute = 60*1000*1000; const rd_ts_t five_seconds = 5*1000*1000; rd_kafka_broker_t *preferred_rkb; rd_kafka_t *rk = rktp->rktp_rkt->rkt_rk; rd_ts_t new_intvl = rd_interval_immediate(&rktp->rktp_new_lease_intvl, one_minute, 0); if (new_intvl < 0) { /* In lieu of KIP-320, the toppar is delegated back to * the leader in the event of an offset out-of-range * error (KIP-392 error case #4) because this scenario * implies the preferred replica is out-of-sync. * * If program execution reaches here, the leader has * relatively quickly instructed the client back to * a preferred replica, quite possibly the same one * as before (possibly resulting from stale metadata), * so we back off the toppar to slow down potential * back-and-forth. */ if (rd_interval_immediate(&rktp->rktp_new_lease_log_intvl, one_minute, 0) > 0) rd_rkb_log(rkb, LOG_NOTICE, "FETCH", "%.*s [%"PRId32"]: preferred replica " "(%"PRId32") lease changing too quickly " "(%"PRId64"s < 60s): possibly due to " "unavailable replica or stale cluster " "state: backing off next fetch", RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic), rktp->rktp_partition, preferred_id, (one_minute - -new_intvl)/(1000*1000)); rd_kafka_toppar_fetch_backoff(rkb, rktp, RD_KAFKA_RESP_ERR_NO_ERROR); } rd_kafka_rdlock(rk); preferred_rkb = rd_kafka_broker_find_by_nodeid(rk, preferred_id); rd_kafka_rdunlock(rk); if (preferred_rkb) { rd_interval_reset_to_now(&rktp->rktp_lease_intvl, 0); rd_kafka_toppar_lock(rktp); rd_kafka_toppar_broker_update(rktp, preferred_id, preferred_rkb, "preferred replica updated"); rd_kafka_toppar_unlock(rktp); rd_kafka_broker_destroy(preferred_rkb); return; } if (rd_interval_immediate(&rktp->rktp_metadata_intvl, five_seconds, 0) > 0) { rd_rkb_log(rkb, LOG_NOTICE, "FETCH", "%.*s [%"PRId32"]: preferred replica (%"PRId32") " "is unknown: refreshing metadata", RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic), rktp->rktp_partition, preferred_id); rd_kafka_metadata_refresh_brokers( rktp->rktp_rkt->rkt_rk, NULL, "preferred replica unavailable"); } rd_kafka_toppar_fetch_backoff( rkb, rktp, RD_KAFKA_RESP_ERR_REPLICA_NOT_AVAILABLE); } /** * @brief Handle partition-specific Fetch error. */ static void rd_kafka_fetch_reply_handle_partition_error ( rd_kafka_broker_t *rkb, rd_kafka_toppar_t *rktp, const struct rd_kafka_toppar_ver *tver, rd_kafka_resp_err_t err, int64_t HighwaterMarkOffset) { /* Some errors should be passed to the * application while some handled by rdkafka */ switch (err) { /* Errors handled by rdkafka */ case RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PART: case RD_KAFKA_RESP_ERR_LEADER_NOT_AVAILABLE: case RD_KAFKA_RESP_ERR_NOT_LEADER_FOR_PARTITION: case RD_KAFKA_RESP_ERR_BROKER_NOT_AVAILABLE: case RD_KAFKA_RESP_ERR_REPLICA_NOT_AVAILABLE: case RD_KAFKA_RESP_ERR_KAFKA_STORAGE_ERROR: case RD_KAFKA_RESP_ERR_FENCED_LEADER_EPOCH: /* Request metadata information update*/ rd_kafka_toppar_leader_unavailable(rktp, "fetch", err); break; case RD_KAFKA_RESP_ERR_OFFSET_NOT_AVAILABLE: /* Occurs when: * - Msg exists on broker but * offset > HWM, or: * - HWM is >= offset, but msg not * yet available at that offset * (replica is out of sync). * * Handle by retrying FETCH (with backoff). */ rd_rkb_dbg(rkb, MSG, "FETCH", "Topic %s [%"PRId32"]: Offset %"PRId64" not " "available on broker %"PRId32" (leader %"PRId32"): " "retrying", rktp->rktp_rkt->rkt_topic->str, rktp->rktp_partition, rktp->rktp_offsets. fetch_offset, rktp->rktp_broker_id, rktp->rktp_leader_id); break; case RD_KAFKA_RESP_ERR_OFFSET_OUT_OF_RANGE: { int64_t err_offset; if (rktp->rktp_broker_id != rktp->rktp_leader_id && rktp->rktp_offsets.fetch_offset > HighwaterMarkOffset) { rd_kafka_log(rkb->rkb_rk, LOG_WARNING, "FETCH", "Topic %s [%"PRId32"]: Offset %"PRId64 " out of range (HighwaterMark %"PRId64 " fetching from " "broker %"PRId32" (leader %"PRId32"): " "reverting to leader", rktp->rktp_rkt->rkt_topic->str, rktp->rktp_partition, rktp->rktp_offsets.fetch_offset, HighwaterMarkOffset, rktp->rktp_broker_id, rktp->rktp_leader_id); /* Out of range error cannot be taken as definitive * when fetching from follower. * Revert back to the leader in lieu of KIP-320. */ rd_kafka_toppar_delegate_to_leader(rktp); break; } /* Application error */ err_offset = rktp->rktp_offsets.fetch_offset; rktp->rktp_offsets.fetch_offset = RD_KAFKA_OFFSET_INVALID; rd_kafka_offset_reset(rktp, err_offset, err, "fetch failed due to requested offset " "not available on the broker"); } break; case RD_KAFKA_RESP_ERR_TOPIC_AUTHORIZATION_FAILED: /* If we're not authorized to access the * topic mark it as errored to deny * further Fetch requests. */ if (rktp->rktp_last_error != err) { rd_kafka_consumer_err( rktp->rktp_fetchq, rd_kafka_broker_id(rkb), err, tver->version, NULL, rktp, rktp->rktp_offsets.fetch_offset, "Fetch from broker %"PRId32" failed: %s", rd_kafka_broker_id(rkb), rd_kafka_err2str(err)); rktp->rktp_last_error = err; } break; /* Application errors */ case RD_KAFKA_RESP_ERR__PARTITION_EOF: if (rkb->rkb_rk->rk_conf.enable_partition_eof) rd_kafka_consumer_err( rktp->rktp_fetchq, rd_kafka_broker_id(rkb), err, tver->version, NULL, rktp, rktp->rktp_offsets.fetch_offset, "Fetch from broker %"PRId32" reached end of " "partition at offset %"PRId64 " (HighwaterMark %"PRId64")", rd_kafka_broker_id(rkb), rktp->rktp_offsets.fetch_offset, HighwaterMarkOffset); break; case RD_KAFKA_RESP_ERR_MSG_SIZE_TOO_LARGE: default: /* and all other errors */ rd_dassert(tver->version > 0); rd_kafka_consumer_err( rktp->rktp_fetchq, rd_kafka_broker_id(rkb), err, tver->version, NULL, rktp, rktp->rktp_offsets.fetch_offset, "Fetch from broker %"PRId32" failed: %s", rd_kafka_broker_id(rkb), rd_kafka_err2str(err)); break; } /* Back off the next fetch for this partition */ rd_kafka_toppar_fetch_backoff(rkb, rktp, 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; rd_kafka_topic_t *rkt = NULL; int16_t ErrorCode = RD_KAFKA_RESP_ERR_NO_ERROR; 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); } if (rd_kafka_buf_ApiVersion(request) >= 7) { int32_t SessionId; rd_kafka_buf_read_i16(rkbuf, &ErrorCode); rd_kafka_buf_read_i32(rkbuf, &SessionId); } 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); 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 = NULL; rd_kafka_aborted_txns_t *aborted_txns = NULL; rd_slice_t save_slice; struct { int32_t Partition; int16_t ErrorCode; int64_t HighwaterMarkOffset; int64_t LastStableOffset; /* v4 */ int64_t LogStartOffset; /* v5 */ int32_t MessageSetSize; int32_t PreferredReadReplica; /* v11 */ } hdr; rd_kafka_resp_err_t err; int64_t end_offset; rd_kafka_buf_read_i32(rkbuf, &hdr.Partition); rd_kafka_buf_read_i16(rkbuf, &hdr.ErrorCode); if (ErrorCode) hdr.ErrorCode = ErrorCode; rd_kafka_buf_read_i64(rkbuf, &hdr.HighwaterMarkOffset); end_offset = hdr.HighwaterMarkOffset; hdr.LastStableOffset = RD_KAFKA_OFFSET_INVALID; hdr.LogStartOffset = RD_KAFKA_OFFSET_INVALID; if (rd_kafka_buf_ApiVersion(request) >= 4) { int32_t AbortedTxnCnt; rd_kafka_buf_read_i64(rkbuf, &hdr.LastStableOffset); if (rd_kafka_buf_ApiVersion(request) >= 5) rd_kafka_buf_read_i64(rkbuf, &hdr.LogStartOffset); rd_kafka_buf_read_i32(rkbuf, &AbortedTxnCnt); if (rkb->rkb_rk->rk_conf.isolation_level == RD_KAFKA_READ_UNCOMMITTED) { if (unlikely(AbortedTxnCnt > 0)) { rd_rkb_log(rkb, LOG_ERR, "FETCH", "%.*s [%"PRId32"]: " "%"PRId32" aborted " "transaction(s) " "encountered in " "READ_UNCOMMITTED " "fetch response: " "ignoring.", RD_KAFKAP_STR_PR( &topic), hdr.Partition, AbortedTxnCnt); rd_kafka_buf_skip(rkbuf, AbortedTxnCnt * (8+8)); } } else { /* Older brokers may return LSO -1, * in which case we use the HWM. */ if (hdr.LastStableOffset >= 0) end_offset = hdr.LastStableOffset; if (AbortedTxnCnt > 0) { int k; if (unlikely(AbortedTxnCnt > 1000000)) rd_kafka_buf_parse_fail( rkbuf, "%.*s [%"PRId32"]: " "invalid AbortedTxnCnt %"PRId32, RD_KAFKAP_STR_PR(&topic), hdr.Partition, AbortedTxnCnt); aborted_txns = rd_kafka_aborted_txns_new(AbortedTxnCnt); for (k = 0 ; k < AbortedTxnCnt; k++) { int64_t PID; int64_t FirstOffset; rd_kafka_buf_read_i64(rkbuf, &PID); rd_kafka_buf_read_i64(rkbuf, &FirstOffset); rd_kafka_aborted_txns_add(aborted_txns, PID, FirstOffset); } rd_kafka_aborted_txns_sort(aborted_txns); } } } if (rd_kafka_buf_ApiVersion(request) >= 11) rd_kafka_buf_read_i32(rkbuf, &hdr.PreferredReadReplica); else hdr.PreferredReadReplica = -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(rkt != NULL)) { rd_kafka_topic_rdlock(rkt); rktp = rd_kafka_toppar_get( rkt, hdr.Partition, 0/*no ua-on-miss*/); rd_kafka_topic_rdunlock(rkt); } if (unlikely(!rkt || !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); if (aborted_txns) rd_kafka_aborted_txns_destroy( aborted_txns); continue; } rd_kafka_toppar_lock(rktp); rktp->rktp_lo_offset = hdr.LogStartOffset; rktp->rktp_hi_offset = hdr.HighwaterMarkOffset; /* Let the LastStable offset be the effective * end_offset based on protocol version, that is: * if connected to a broker that does not support * LastStableOffset we use the HighwaterMarkOffset. */ rktp->rktp_ls_offset = end_offset; rd_kafka_toppar_unlock(rktp); if (hdr.PreferredReadReplica != -1) { rd_kafka_fetch_preferred_replica_handle( rktp, rkbuf, rkb, hdr.PreferredReadReplica); if (unlikely(hdr.MessageSetSize != 0)) { rd_rkb_log(rkb, LOG_WARNING, "FETCH", "%.*s [%"PRId32"]: Fetch " "response has both " "preferred read replica " "and non-zero message set " "size: %"PRId32": " "skipping messages", RD_KAFKAP_STR_PR(rktp-> rktp_rkt->rkt_topic), rktp->rktp_partition, hdr.MessageSetSize); rd_kafka_buf_skip(rkbuf, hdr.MessageSetSize); } if (aborted_txns) rd_kafka_aborted_txns_destroy( aborted_txns); rd_kafka_toppar_destroy(rktp); /* from get */ continue; } rd_kafka_toppar_lock(rktp); /* Make sure toppar hasn't moved to another broker * during the lifetime of the request. */ if (unlikely(rktp->rktp_broker != rkb)) { rd_kafka_toppar_unlock(rktp); rd_rkb_dbg(rkb, MSG, "FETCH", "%.*s [%"PRId32"]: " "partition broker has changed: " "discarding fetch response", RD_KAFKAP_STR_PR(&topic), hdr.Partition); rd_kafka_toppar_destroy(rktp); /* from get */ rd_kafka_buf_skip(rkbuf, hdr.MessageSetSize); if (aborted_txns) rd_kafka_aborted_txns_destroy( aborted_txns); continue; } fetch_version = rktp->rktp_fetch_version; rd_kafka_toppar_unlock(rktp); /* Check if this Fetch is for an outdated fetch version, * or the original rktp was removed and a new one * created (due to partition count decreasing and * then increasing again, which can happen in * desynchronized clusters): if so ignore it. */ tver_skel.rktp = rktp; tver = rd_list_find(request->rkbuf_rktp_vers, &tver_skel, rd_kafka_toppar_ver_cmp); rd_kafka_assert(NULL, tver); if (tver->rktp != rktp || tver->version < fetch_version) { rd_rkb_dbg(rkb, MSG, "DROP", "%s [%"PRId32"]: " "dropping outdated fetch response " "(v%d < %d or old rktp)", rktp->rktp_rkt->rkt_topic->str, rktp->rktp_partition, tver->version, fetch_version); rd_atomic64_add(&rktp->rktp_c. rx_ver_drops, 1); rd_kafka_toppar_destroy(rktp); /* from get */ rd_kafka_buf_skip(rkbuf, hdr.MessageSetSize); if (aborted_txns) rd_kafka_aborted_txns_destroy( aborted_txns); continue; } rd_rkb_dbg(rkb, MSG, "FETCH", "Topic %.*s [%"PRId32"] MessageSet " "size %"PRId32", error \"%s\", " "MaxOffset %"PRId64", " "LSO %"PRId64", " "Ver %"PRId32"/%"PRId32, RD_KAFKAP_STR_PR(&topic), hdr.Partition, hdr.MessageSetSize, rd_kafka_err2str(hdr.ErrorCode), hdr.HighwaterMarkOffset, hdr.LastStableOffset, tver->version, fetch_version); /* If this is the last message of the queue, * signal EOF back to the application. */ if (end_offset == rktp->rktp_offsets.fetch_offset && rktp->rktp_offsets.eof_offset != rktp->rktp_offsets.fetch_offset) { hdr.ErrorCode = RD_KAFKA_RESP_ERR__PARTITION_EOF; rktp->rktp_offsets.eof_offset = rktp->rktp_offsets.fetch_offset; } if (unlikely(hdr.ErrorCode != RD_KAFKA_RESP_ERR_NO_ERROR)) { /* Handle partition-level errors. */ rd_kafka_fetch_reply_handle_partition_error( rkb, rktp, tver, hdr.ErrorCode, hdr.HighwaterMarkOffset); rd_kafka_toppar_destroy(rktp); /* from get()*/ rd_kafka_buf_skip(rkbuf, hdr.MessageSetSize); if (aborted_txns) rd_kafka_aborted_txns_destroy( aborted_txns); continue; } /* No error, clear any previous fetch error. */ rktp->rktp_last_error = RD_KAFKA_RESP_ERR_NO_ERROR; if (unlikely(hdr.MessageSetSize <= 0)) { rd_kafka_toppar_destroy(rktp); /*from get()*/ if (aborted_txns) rd_kafka_aborted_txns_destroy( aborted_txns); continue; } /** * Parse MessageSet */ if (!rd_slice_narrow_relative( &rkbuf->rkbuf_reader, &save_slice, (size_t)hdr.MessageSetSize)) rd_kafka_buf_check_len(rkbuf, hdr.MessageSetSize); /* Parse messages */ err = rd_kafka_msgset_parse( rkbuf, request, rktp, aborted_txns, tver); if (aborted_txns) rd_kafka_aborted_txns_destroy( aborted_txns); rd_slice_widen(&rkbuf->rkbuf_reader, &save_slice); /* Continue with next partition regardless of * parse errors (which are partition-specific) */ /* On error: back off the fetcher for this partition */ if (unlikely(err)) rd_kafka_toppar_fetch_backoff(rkb, rktp, err); rd_kafka_toppar_destroy(rktp); /* from get */ } if (rkt) { rd_kafka_topic_destroy0(rkt); rkt = NULL; } } if (rd_kafka_buf_read_remain(rkbuf) != 0) { rd_kafka_buf_parse_fail(rkbuf, "Remaining data after message set " "parse: %"PRIusz" bytes", rd_kafka_buf_read_remain(rkbuf)); RD_NOTREACHED(); } return 0; err_parse: if (rkt) rd_kafka_topic_destroy0(rkt); rd_rkb_dbg(rkb, MSG, "BADMSG", "Bad message (Fetch v%d): " "is broker.version.fallback incorrectly set?", (int)request->rkbuf_reqhdr.ApiVersion); return rkbuf->rkbuf_err; } static void rd_kafka_broker_fetch_reply (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) { if (err == RD_KAFKA_RESP_ERR__DESTROY) return; /* Terminating */ rd_kafka_assert(rkb->rkb_rk, rkb->rkb_fetching > 0); rkb->rkb_fetching = 0; /* Parse and handle the messages (unless the request errored) */ if (!err && reply) err = rd_kafka_fetch_reply_handle(rkb, reply, request); if (unlikely(err)) { char tmp[128]; rd_rkb_dbg(rkb, MSG, "FETCH", "Fetch reply: %s", rd_kafka_err2str(err)); switch (err) { case RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PART: case RD_KAFKA_RESP_ERR_LEADER_NOT_AVAILABLE: case RD_KAFKA_RESP_ERR_NOT_LEADER_FOR_PARTITION: case RD_KAFKA_RESP_ERR_BROKER_NOT_AVAILABLE: case RD_KAFKA_RESP_ERR_REPLICA_NOT_AVAILABLE: /* Request metadata information update */ rd_snprintf(tmp, sizeof(tmp), "FetchRequest failed: %s", rd_kafka_err2str(err)); rd_kafka_metadata_refresh_known_topics( rkb->rkb_rk, NULL, rd_true/*force*/, tmp); /* FALLTHRU */ case RD_KAFKA_RESP_ERR__TRANSPORT: case RD_KAFKA_RESP_ERR_REQUEST_TIMED_OUT: case RD_KAFKA_RESP_ERR__MSG_TIMED_OUT: /* The fetch is already intervalled from * consumer_serve() so dont retry. */ break; default: break; } rd_kafka_broker_fetch_backoff(rkb, err); /* FALLTHRU */ } } /** * Build and send a Fetch request message for all underflowed toppars * for a specific broker. */ static int rd_kafka_broker_fetch_toppars (rd_kafka_broker_t *rkb, rd_ts_t now) { rd_kafka_toppar_t *rktp; rd_kafka_buf_t *rkbuf; int cnt = 0; size_t of_TopicArrayCnt = 0; int TopicArrayCnt = 0; size_t of_PartitionArrayCnt = 0; int PartitionArrayCnt = 0; rd_kafka_topic_t *rkt_last = NULL; int16_t ApiVersion = 0; /* Create buffer and segments: * 1 x ReplicaId MaxWaitTime MinBytes TopicArrayCnt * N x topic name * N x PartitionArrayCnt Partition FetchOffset MaxBytes * where N = number of toppars. * Since we dont keep track of the number of topics served by * this broker, only the partition count, we do a worst-case calc * when allocating and assume each partition is on its own topic */ if (unlikely(rkb->rkb_active_toppar_cnt == 0)) return 0; rkbuf = rd_kafka_buf_new_request( rkb, RD_KAFKAP_Fetch, 1, /* ReplicaId+MaxWaitTime+MinBytes+MaxBytes+IsolationLevel+ * SessionId+Epoch+TopicCnt */ 4+4+4+4+1+4+4+4+ /* N x PartCnt+Partition+CurrentLeaderEpoch+FetchOffset+ * LogStartOffset+MaxBytes+?TopicNameLen?*/ (rkb->rkb_active_toppar_cnt * (4+4+4+8+8+4+40)) + /* ForgottenTopicsCnt */ 4+ /* N x ForgottenTopicsData */ 0); ApiVersion = rd_kafka_broker_ApiVersion_supported( rkb, RD_KAFKAP_Fetch, 0, 11, NULL); if (rkb->rkb_features & RD_KAFKA_FEATURE_MSGVER2) rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, RD_KAFKA_FEATURE_MSGVER2); else if (rkb->rkb_features & RD_KAFKA_FEATURE_MSGVER1) rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, RD_KAFKA_FEATURE_MSGVER1); else if (rkb->rkb_features & RD_KAFKA_FEATURE_THROTTLETIME) rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, RD_KAFKA_FEATURE_THROTTLETIME); /* FetchRequest header */ /* ReplicaId */ rd_kafka_buf_write_i32(rkbuf, -1); /* MaxWaitTime */ rd_kafka_buf_write_i32(rkbuf, rkb->rkb_rk->rk_conf.fetch_wait_max_ms); /* MinBytes */ rd_kafka_buf_write_i32(rkbuf, rkb->rkb_rk->rk_conf.fetch_min_bytes); if (rd_kafka_buf_ApiVersion(rkbuf) >= 3) /* MaxBytes */ rd_kafka_buf_write_i32(rkbuf, rkb->rkb_rk->rk_conf.fetch_max_bytes); if (rd_kafka_buf_ApiVersion(rkbuf) >= 4) /* IsolationLevel */ rd_kafka_buf_write_i8(rkbuf, rkb->rkb_rk->rk_conf.isolation_level); if (rd_kafka_buf_ApiVersion(rkbuf) >= 7) { /* SessionId */ rd_kafka_buf_write_i32(rkbuf, 0); /* Epoch */ rd_kafka_buf_write_i32(rkbuf, -1); } /* Write zero TopicArrayCnt but store pointer for later update */ of_TopicArrayCnt = rd_kafka_buf_write_i32(rkbuf, 0); /* Prepare map for storing the fetch version for each partition, * this will later be checked in Fetch response to purge outdated * responses (e.g., after a seek). */ rkbuf->rkbuf_rktp_vers = rd_list_new( 0, (void *)rd_kafka_toppar_ver_destroy); rd_list_prealloc_elems(rkbuf->rkbuf_rktp_vers, sizeof(struct rd_kafka_toppar_ver), rkb->rkb_active_toppar_cnt, 0); /* Round-robin start of the list. */ rktp = rkb->rkb_active_toppar_next; do { struct rd_kafka_toppar_ver *tver; if (rkt_last != rktp->rktp_rkt) { if (rkt_last != NULL) { /* Update PartitionArrayCnt */ rd_kafka_buf_update_i32(rkbuf, of_PartitionArrayCnt, PartitionArrayCnt); } /* Topic name */ rd_kafka_buf_write_kstr(rkbuf, rktp->rktp_rkt->rkt_topic); TopicArrayCnt++; rkt_last = rktp->rktp_rkt; /* Partition count */ of_PartitionArrayCnt = rd_kafka_buf_write_i32(rkbuf, 0); PartitionArrayCnt = 0; } PartitionArrayCnt++; /* Partition */ rd_kafka_buf_write_i32(rkbuf, rktp->rktp_partition); if (rd_kafka_buf_ApiVersion(rkbuf) >= 9) /* CurrentLeaderEpoch */ rd_kafka_buf_write_i32(rkbuf, -1); /* FetchOffset */ rd_kafka_buf_write_i64(rkbuf, rktp->rktp_offsets.fetch_offset); if (rd_kafka_buf_ApiVersion(rkbuf) >= 5) /* LogStartOffset - only used by follower replica */ rd_kafka_buf_write_i64(rkbuf, -1); /* MaxBytes */ rd_kafka_buf_write_i32(rkbuf, rktp->rktp_fetch_msg_max_bytes); rd_rkb_dbg(rkb, FETCH, "FETCH", "Fetch topic %.*s [%"PRId32"] at offset %"PRId64 " (v%d)", RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic), rktp->rktp_partition, rktp->rktp_offsets.fetch_offset, rktp->rktp_fetch_version); /* We must have a valid fetch offset when we get here */ rd_dassert(rktp->rktp_offsets.fetch_offset >= 0); /* Add toppar + op version mapping. */ tver = rd_list_add(rkbuf->rkbuf_rktp_vers, NULL); tver->rktp = rd_kafka_toppar_keep(rktp); tver->version = rktp->rktp_fetch_version; cnt++; } while ((rktp = CIRCLEQ_LOOP_NEXT(&rkb->rkb_active_toppars, rktp, rktp_activelink)) != rkb->rkb_active_toppar_next); /* Update next toppar to fetch in round-robin list. */ rd_kafka_broker_active_toppar_next( rkb, rktp ? CIRCLEQ_LOOP_NEXT(&rkb->rkb_active_toppars, rktp, rktp_activelink) : NULL); rd_rkb_dbg(rkb, FETCH, "FETCH", "Fetch %i/%i/%i toppar(s)", cnt, rkb->rkb_active_toppar_cnt, rkb->rkb_toppar_cnt); if (!cnt) { rd_kafka_buf_destroy(rkbuf); return cnt; } if (rkt_last != NULL) { /* Update last topic's PartitionArrayCnt */ rd_kafka_buf_update_i32(rkbuf, of_PartitionArrayCnt, PartitionArrayCnt); } /* Update TopicArrayCnt */ rd_kafka_buf_update_i32(rkbuf, of_TopicArrayCnt, TopicArrayCnt); if (rd_kafka_buf_ApiVersion(rkbuf) >= 7) /* Length of the ForgottenTopics list (KIP-227). Broker * use only - not used by the consumer. */ rd_kafka_buf_write_i32(rkbuf, 0); if (rd_kafka_buf_ApiVersion(rkbuf) >= 11) /* RackId */ rd_kafka_buf_write_kstr(rkbuf, rkb->rkb_rk->rk_conf.client_rack); /* Consider Fetch requests blocking if fetch.wait.max.ms >= 1s */ if (rkb->rkb_rk->rk_conf.fetch_wait_max_ms >= 1000) rkbuf->rkbuf_flags |= RD_KAFKA_OP_F_BLOCKING; /* Use configured timeout */ rd_kafka_buf_set_timeout(rkbuf, rkb->rkb_rk->rk_conf.socket_timeout_ms + rkb->rkb_rk->rk_conf.fetch_wait_max_ms, now); /* Sort toppar versions for quicker lookups in Fetch response. */ rd_list_sort(rkbuf->rkbuf_rktp_vers, rd_kafka_toppar_ver_cmp); rkb->rkb_fetching = 1; rd_kafka_broker_buf_enq1(rkb, rkbuf, rd_kafka_broker_fetch_reply, NULL); return cnt; } /** * Consumer serving */ static void rd_kafka_broker_consumer_serve (rd_kafka_broker_t *rkb, rd_ts_t abs_timeout) { unsigned int initial_state = rkb->rkb_state; rd_ts_t now; 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 == initial_state && abs_timeout > (now = rd_clock())) { rd_ts_t min_backoff; rd_kafka_broker_unlock(rkb); /* Serve toppars */ min_backoff = rd_kafka_broker_consumer_toppars_serve(rkb); if (rkb->rkb_ts_fetch_backoff > now && rkb->rkb_ts_fetch_backoff < min_backoff) min_backoff = rkb->rkb_ts_fetch_backoff; if (min_backoff < RD_TS_MAX && rkb->rkb_state != RD_KAFKA_BROKER_STATE_UP) { /* There are partitions to fetch but the * connection is not up. */ rkb->rkb_persistconn.internal++; } /* Send Fetch request message for all underflowed toppars * if the connection is up and there are no outstanding * fetch requests for this connection. */ if (!rkb->rkb_fetching && rkb->rkb_state == RD_KAFKA_BROKER_STATE_UP) { if (min_backoff < now) { rd_kafka_broker_fetch_toppars(rkb, now); min_backoff = abs_timeout; } else if (min_backoff < RD_TS_MAX) rd_rkb_dbg(rkb, FETCH, "FETCH", "Fetch backoff for %"PRId64 "ms", (min_backoff-now)/1000); } else { /* Nothing needs to be done, next wakeup * is from ops, state change, IO, or this timeout */ min_backoff = abs_timeout; } /* Check and move retry buffers */ if (unlikely(rd_atomic32_get(&rkb->rkb_retrybufs.rkbq_cnt) > 0)) rd_kafka_broker_retry_bufs_move(rkb, &min_backoff); if (min_backoff > abs_timeout) min_backoff = abs_timeout; if (rd_kafka_broker_ops_io_serve(rkb, min_backoff)) return; /* Wakeup */ rd_kafka_broker_lock(rkb); } rd_kafka_broker_unlock(rkb); } /** * @brief Check if connections.max.idle.ms has been exceeded and if so * close the connection. * * @remark Must only be called if connections.max.idle.ms > 0 and * the current broker state is UP (or UPDATE). * * @locality broker thread */ static RD_INLINE void rd_kafka_broker_idle_check (rd_kafka_broker_t *rkb) { rd_ts_t ts_send = rd_atomic64_get(&rkb->rkb_c.ts_send); rd_ts_t ts_recv = rd_atomic64_get(&rkb->rkb_c.ts_recv); rd_ts_t ts_last_activity = RD_MAX(ts_send, ts_recv); int idle_ms; /* If nothing has been sent yet, use the connection time as * last activity. */ if (unlikely(!ts_last_activity)) ts_last_activity = rkb->rkb_ts_state; idle_ms = (int)((rd_clock() - ts_last_activity) / 1000); if (likely(idle_ms < rkb->rkb_rk->rk_conf.connections_max_idle_ms)) return; rd_kafka_broker_fail(rkb, LOG_DEBUG, RD_KAFKA_RESP_ERR__TRANSPORT, "Connection max idle time exceeded " "(%dms since last activity)", idle_ms); } /** * @brief Serve broker thread according to client type. * May be called in any broker state. * * This function is to be called from the state machine in * rd_kafka_broker_thread_main, and will return when * there was a state change, or the handle is terminating. * * Broker threads are triggered by three things: * - Ops from other parts of librdkafka / app. * This is the rkb_ops queue which is served from * rd_kafka_broker_ops_io_serve(). * - IO from broker socket. * The ops queue is also IO-triggered to provide * quick wakeup when thread is blocking on IO. * Also serverd from rd_kafka_broker_ops_io_serve(). * When there is no broker socket only the ops * queue is served. * - Ops/IO timeout when there were no ops or * IO events within a variable timeout. * * For each iteration of the loops in producer_serve(), consumer_serve(), * etc, the Ops and IO are polled, and the client type specific * logic is executed. For the consumer this logic checks which partitions * to fetch or backoff, and sends Fetch requests. * The producer checks for messages to batch and transmit. * All types check for request timeouts, etc. * * Wakeups * ======= * The logic returns a next wakeup time which controls how long the * next Ops/IO poll may block before the logic wants to run again; * this is typically controlled by `linger.ms` on the Producer * and fetch backoffs on the consumer. * * Remote threads may also want to wake up the Ops/IO poll so that * the logic is run more quickly. For example when a new message * is enqueued by produce() it is important that it is batched * and transmitted within the configured `linger.ms`. * * Any op enqueued on the broker ops queue (rkb_ops) will automatically * trigger a wakeup of the broker thread (either by wakeup_fd IO event * or by the conditional variable of rkb_ops being triggered - or both). * * Produced messages are not enqueued on the rkb_ops queue but on * the partition's rktp_msgq message queue. To provide quick wakeups * the partition has a reference to the partition's current leader broker * thread's rkb_ops queue, rktp_msgq_wakeup_q. * When enqueuing a message on the partition queue and the queue was * previously empty, the rktp_msgq_wakeup_q (which is rkb_ops) is woken up * by rd_kafka_q_yield(), which sets a YIELD flag and triggers the cond var * to wake up the broker thread (without allocating and enqueuing an rko). * This also triggers the wakeup_fd of rkb_ops, if necessary. * * When sparse connections is enabled the broker will linger in the * INIT state until there's a need for a connection, in which case * it will set its state to DOWN to trigger the connection. * This is controlled both by the shared rkb_persistconn atomic counters * that may be updated from other parts of the code, as well as the * temporary per broker_serve() rkb_persistconn.internal counter which * is used by the broker handler code to detect if a connection is needed, * such as when a partition is being produced to. * * * @param timeout_ms The maximum timeout for blocking Ops/IO. * * @locality broker thread * @locks none */ static void rd_kafka_broker_serve (rd_kafka_broker_t *rkb, int timeout_ms) { rd_ts_t abs_timeout; if (unlikely(rd_kafka_terminating(rkb->rkb_rk) || timeout_ms == RD_POLL_NOWAIT)) timeout_ms = 1; else if (timeout_ms == RD_POLL_INFINITE) timeout_ms = rd_kafka_max_block_ms; abs_timeout = rd_timeout_init(timeout_ms); /* Must be a valid absolute time from here on. */ rd_assert(abs_timeout > 0); /* rkb_persistconn.internal is the per broker_serve() * automatic counter that keeps track of anything * in the producer/consumer logic needs this broker connection * to be up. * The value is reset here on each serve(). If there are queued * requests we know right away that a connection is needed. */ rkb->rkb_persistconn.internal = rd_atomic32_get(&rkb->rkb_outbufs.rkbq_cnt) > 0; if (rkb->rkb_source == RD_KAFKA_INTERNAL) { rd_kafka_broker_internal_serve(rkb, abs_timeout); return; } if (rkb->rkb_rk->rk_type == RD_KAFKA_PRODUCER) rd_kafka_broker_producer_serve(rkb, abs_timeout); else if (rkb->rkb_rk->rk_type == RD_KAFKA_CONSUMER) rd_kafka_broker_consumer_serve(rkb, abs_timeout); if (rkb->rkb_rk->rk_conf.connections_max_idle_ms && rkb->rkb_state == RD_KAFKA_BROKER_STATE_UP) rd_kafka_broker_idle_check(rkb); } /** * @returns true if all broker addresses have been tried. * * @locality broker thread * @locks_required none * @locks_acquired none */ static rd_bool_t rd_kafka_broker_addresses_exhausted (const rd_kafka_broker_t *rkb) { return !rkb->rkb_rsal || rkb->rkb_rsal->rsal_cnt == 0 || rkb->rkb_rsal->rsal_curr + 1 == rkb->rkb_rsal->rsal_cnt; } static int rd_kafka_broker_thread_main (void *arg) { rd_kafka_broker_t *rkb = arg; rd_kafka_t *rk = rkb->rkb_rk; rd_kafka_set_thread_name("%s", rkb->rkb_name); rd_kafka_set_thread_sysname("rdk:broker%"PRId32, rkb->rkb_nodeid); rd_kafka_interceptors_on_thread_start(rk, RD_KAFKA_THREAD_BROKER); (void)rd_atomic32_add(&rd_kafka_thread_cnt_curr, 1); /* Our own refcount was increased just prior to thread creation, * when refcount drops to 1 it is just us left and the broker * thread should terminate. */ /* Acquire lock (which was held by thread creator during creation) * to synchronise state. */ rd_kafka_broker_lock(rkb); rd_kafka_broker_unlock(rkb); rd_rkb_dbg(rkb, BROKER, "BRKMAIN", "Enter main broker thread"); while (!rd_kafka_broker_terminating(rkb)) { int backoff; int r; redo: switch (rkb->rkb_state) { case RD_KAFKA_BROKER_STATE_INIT: /* Check if there is demand for a connection * to this broker, if so jump to TRY_CONNECT state. */ if (!rd_kafka_broker_needs_connection(rkb)) { rd_kafka_broker_serve(rkb, rd_kafka_max_block_ms); break; } /* The INIT state also exists so that an initial * connection failure triggers a state transition * which might trigger a ALL_BROKERS_DOWN error. */ rd_kafka_broker_lock(rkb); rd_kafka_broker_set_state( rkb, RD_KAFKA_BROKER_STATE_TRY_CONNECT); rd_kafka_broker_unlock(rkb); goto redo; /* effectively a fallthru to TRY_CONNECT */ case RD_KAFKA_BROKER_STATE_DOWN: rd_kafka_broker_lock(rkb); if (rkb->rkb_rk->rk_conf.sparse_connections) rd_kafka_broker_set_state( rkb, RD_KAFKA_BROKER_STATE_INIT); else rd_kafka_broker_set_state( rkb, RD_KAFKA_BROKER_STATE_TRY_CONNECT); rd_kafka_broker_unlock(rkb); goto redo; /* effectively a fallthru to TRY_CONNECT */ case RD_KAFKA_BROKER_STATE_TRY_CONNECT: if (rkb->rkb_source == RD_KAFKA_INTERNAL) { rd_kafka_broker_lock(rkb); rd_kafka_broker_set_state(rkb, RD_KAFKA_BROKER_STATE_UP); rd_kafka_broker_unlock(rkb); break; } if (unlikely(rd_kafka_terminating(rkb->rkb_rk))) rd_kafka_broker_serve(rkb, 1000); if (!rd_kafka_sasl_ready(rkb->rkb_rk)) { /* SASL provider not yet ready. */ rd_kafka_broker_serve(rkb, rd_kafka_max_block_ms); /* Continue while loop to try again (as long as * we are not terminating). */ continue; } /* Throttle & jitter reconnects to avoid * thundering horde of reconnecting clients after * a broker / network outage. Issue #403 */ backoff = rd_kafka_broker_reconnect_backoff(rkb, rd_clock()); if (backoff > 0) { rd_rkb_dbg(rkb, BROKER, "RECONNECT", "Delaying next reconnect by %dms", backoff); rd_kafka_broker_serve(rkb, (int)backoff); continue; } /* Initiate asynchronous connection attempt. * Only the host lookup is blocking here. */ r = rd_kafka_broker_connect(rkb); if (r == -1) { /* Immediate failure, most likely host * resolving failed. * Try the next resolve result until we've * tried them all, in which case we sleep a * short while to avoid busy looping. */ if (rd_kafka_broker_addresses_exhausted(rkb)) rd_kafka_broker_serve( rkb, rd_kafka_max_block_ms); } else if (r == 0) { /* Broker has no hostname yet, wait * for hostname to be set and connection * triggered by received OP_CONNECT. */ rd_kafka_broker_serve(rkb, rd_kafka_max_block_ms); } else { /* Connection in progress, state will * have changed to STATE_CONNECT. */ } break; case RD_KAFKA_BROKER_STATE_CONNECT: case RD_KAFKA_BROKER_STATE_SSL_HANDSHAKE: case RD_KAFKA_BROKER_STATE_AUTH_LEGACY: case RD_KAFKA_BROKER_STATE_AUTH_REQ: case RD_KAFKA_BROKER_STATE_AUTH_HANDSHAKE: case RD_KAFKA_BROKER_STATE_APIVERSION_QUERY: /* Asynchronous connect in progress. */ rd_kafka_broker_serve(rkb, rd_kafka_max_block_ms); /* Connect failure. * Try the next resolve result until we've * tried them all, in which case we back off the next * connection attempt to avoid busy looping. */ if (rkb->rkb_state == RD_KAFKA_BROKER_STATE_DOWN && rd_kafka_broker_addresses_exhausted(rkb)) rd_kafka_broker_update_reconnect_backoff( rkb, &rkb->rkb_rk->rk_conf, rd_clock()); break; case RD_KAFKA_BROKER_STATE_UPDATE: /* FALLTHRU */ case RD_KAFKA_BROKER_STATE_UP: rd_kafka_broker_serve(rkb, rd_kafka_max_block_ms); if (rkb->rkb_state == RD_KAFKA_BROKER_STATE_UPDATE) { rd_kafka_broker_lock(rkb); rd_kafka_broker_set_state(rkb, RD_KAFKA_BROKER_STATE_UP); rd_kafka_broker_unlock(rkb); } break; } if (rd_kafka_terminating(rkb->rkb_rk)) { /* Handle is terminating: fail the send+retry queue * to speed up termination, otherwise we'll * need to wait for request timeouts. */ r = rd_kafka_broker_bufq_timeout_scan( rkb, 0, &rkb->rkb_outbufs, NULL, -1, RD_KAFKA_RESP_ERR__DESTROY, 0, NULL, 0); r += rd_kafka_broker_bufq_timeout_scan( rkb, 0, &rkb->rkb_retrybufs, NULL, -1, RD_KAFKA_RESP_ERR__DESTROY, 0, NULL, 0); rd_rkb_dbg(rkb, BROKER, "TERMINATE", "Handle is terminating in state %s: " "%d refcnts (%p), %d toppar(s), " "%d active toppar(s), " "%d outbufs, %d waitresps, %d retrybufs: " "failed %d request(s) in retry+outbuf", rd_kafka_broker_state_names[rkb->rkb_state], rd_refcnt_get(&rkb->rkb_refcnt), &rkb->rkb_refcnt, rkb->rkb_toppar_cnt, rkb->rkb_active_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), r); } } if (rkb->rkb_source != RD_KAFKA_INTERNAL) { rd_kafka_wrlock(rkb->rkb_rk); TAILQ_REMOVE(&rkb->rkb_rk->rk_brokers, rkb, rkb_link); if (rkb->rkb_nodeid != -1 && !RD_KAFKA_BROKER_IS_LOGICAL(rkb)) rd_list_remove(&rkb->rkb_rk->rk_broker_by_id, rkb); (void)rd_atomic32_sub(&rkb->rkb_rk->rk_broker_cnt, 1); rd_kafka_wrunlock(rkb->rkb_rk); } rd_kafka_broker_fail(rkb, LOG_DEBUG, RD_KAFKA_RESP_ERR__DESTROY, "Broker handle is terminating"); /* Disable and drain ops queue. * Simply purging the ops queue risks leaving dangling references * for ops such as PARTITION_JOIN/PARTITION_LEAVE where the broker * reference is not maintained in the rko (but in rktp_next_leader). * #1596 */ rd_kafka_q_disable(rkb->rkb_ops); while (rd_kafka_broker_ops_serve(rkb, RD_POLL_NOWAIT)) ; rd_kafka_broker_destroy(rkb); #if WITH_SSL /* Remove OpenSSL per-thread error state to avoid memory leaks */ #if OPENSSL_VERSION_NUMBER >= 0x10100000L && !defined(LIBRESSL_VERSION_NUMBER) /*(OpenSSL libraries handle thread init and deinit) * https://github.com/openssl/openssl/pull/1048 */ #elif OPENSSL_VERSION_NUMBER >= 0x10000000L ERR_remove_thread_state(NULL); #endif #endif rd_kafka_interceptors_on_thread_exit(rk, RD_KAFKA_THREAD_BROKER); rd_atomic32_sub(&rd_kafka_thread_cnt_curr, 1); return 0; } /** * Final destructor. Refcnt must be 0. */ void rd_kafka_broker_destroy_final (rd_kafka_broker_t *rkb) { rd_assert(thrd_is_current(rkb->rkb_thread)); rd_assert(TAILQ_EMPTY(&rkb->rkb_monitors)); rd_assert(TAILQ_EMPTY(&rkb->rkb_outbufs.rkbq_bufs)); rd_assert(TAILQ_EMPTY(&rkb->rkb_waitresps.rkbq_bufs)); rd_assert(TAILQ_EMPTY(&rkb->rkb_retrybufs.rkbq_bufs)); rd_assert(TAILQ_EMPTY(&rkb->rkb_toppars)); if (rkb->rkb_source != RD_KAFKA_INTERNAL && (rkb->rkb_rk->rk_conf.security_protocol == RD_KAFKA_PROTO_SASL_PLAINTEXT || rkb->rkb_rk->rk_conf.security_protocol == RD_KAFKA_PROTO_SASL_SSL)) rd_kafka_sasl_broker_term(rkb); if (rkb->rkb_wakeup_fd[0] != -1) rd_close(rkb->rkb_wakeup_fd[0]); if (rkb->rkb_wakeup_fd[1] != -1) rd_close(rkb->rkb_wakeup_fd[1]); if (rkb->rkb_recv_buf) rd_kafka_buf_destroy(rkb->rkb_recv_buf); if (rkb->rkb_rsal) rd_sockaddr_list_destroy(rkb->rkb_rsal); if (rkb->rkb_ApiVersions) rd_free(rkb->rkb_ApiVersions); rd_free(rkb->rkb_origname); rd_kafka_q_purge(rkb->rkb_ops); rd_kafka_q_destroy_owner(rkb->rkb_ops); rd_avg_destroy(&rkb->rkb_avg_int_latency); rd_avg_destroy(&rkb->rkb_avg_outbuf_latency); rd_avg_destroy(&rkb->rkb_avg_rtt); rd_avg_destroy(&rkb->rkb_avg_throttle); mtx_lock(&rkb->rkb_logname_lock); rd_free(rkb->rkb_logname); rkb->rkb_logname = NULL; mtx_unlock(&rkb->rkb_logname_lock); mtx_destroy(&rkb->rkb_logname_lock); mtx_destroy(&rkb->rkb_lock); rd_refcnt_destroy(&rkb->rkb_refcnt); rd_free(rkb); } /** * Returns the internal broker with refcnt increased. */ rd_kafka_broker_t *rd_kafka_broker_internal (rd_kafka_t *rk) { rd_kafka_broker_t *rkb; mtx_lock(&rk->rk_internal_rkb_lock); rkb = rk->rk_internal_rkb; if (rkb) rd_kafka_broker_keep(rkb); mtx_unlock(&rk->rk_internal_rkb_lock); return rkb; } /** * Adds a broker with refcount set to 1. * If 'source' is RD_KAFKA_INTERNAL an internal broker is added * that does not actually represent or connect to a real broker, it is used * for serving unassigned toppar's op queues. * * Locks: rd_kafka_wrlock(rk) must be held */ rd_kafka_broker_t *rd_kafka_broker_add (rd_kafka_t *rk, rd_kafka_confsource_t source, rd_kafka_secproto_t proto, const char *name, uint16_t port, int32_t nodeid) { rd_kafka_broker_t *rkb; int r; #ifndef _WIN32 sigset_t newset, oldset; #endif rkb = rd_calloc(1, sizeof(*rkb)); if (source != RD_KAFKA_LOGICAL) { rd_kafka_mk_nodename(rkb->rkb_nodename, sizeof(rkb->rkb_nodename), name, port); rd_kafka_mk_brokername(rkb->rkb_name, sizeof(rkb->rkb_name), proto, rkb->rkb_nodename, nodeid, source); } else { /* Logical broker does not have a nodename (address) or port * at initialization. */ rd_snprintf(rkb->rkb_name, sizeof(rkb->rkb_name), "%s", name); } rkb->rkb_source = source; rkb->rkb_rk = rk; rkb->rkb_ts_state = rd_clock(); rkb->rkb_nodeid = nodeid; rkb->rkb_proto = proto; rkb->rkb_port = port; rkb->rkb_origname = rd_strdup(name); mtx_init(&rkb->rkb_lock, mtx_plain); mtx_init(&rkb->rkb_logname_lock, mtx_plain); rkb->rkb_logname = rd_strdup(rkb->rkb_name); TAILQ_INIT(&rkb->rkb_toppars); CIRCLEQ_INIT(&rkb->rkb_active_toppars); TAILQ_INIT(&rkb->rkb_monitors); rd_kafka_bufq_init(&rkb->rkb_outbufs); rd_kafka_bufq_init(&rkb->rkb_waitresps); rd_kafka_bufq_init(&rkb->rkb_retrybufs); rkb->rkb_ops = rd_kafka_q_new(rk); rd_avg_init(&rkb->rkb_avg_int_latency, RD_AVG_GAUGE, 0, 100*1000, 2, rk->rk_conf.stats_interval_ms ? 1 : 0); rd_avg_init(&rkb->rkb_avg_outbuf_latency, RD_AVG_GAUGE, 0, 100*1000, 2, rk->rk_conf.stats_interval_ms ? 1 : 0); rd_avg_init(&rkb->rkb_avg_rtt, RD_AVG_GAUGE, 0, 500*1000, 2, rk->rk_conf.stats_interval_ms ? 1 : 0); rd_avg_init(&rkb->rkb_avg_throttle, RD_AVG_GAUGE, 0, 5000*1000, 2, rk->rk_conf.stats_interval_ms ? 1 : 0); rd_refcnt_init(&rkb->rkb_refcnt, 0); rd_kafka_broker_keep(rkb); /* rk_broker's refcount */ rkb->rkb_reconnect_backoff_ms = rk->rk_conf.reconnect_backoff_ms; rd_atomic32_init(&rkb->rkb_persistconn.coord, 0); rd_atomic64_init(&rkb->rkb_c.ts_send, 0); rd_atomic64_init(&rkb->rkb_c.ts_recv, 0); /* ApiVersion fallback interval */ if (rkb->rkb_rk->rk_conf.api_version_request) { rd_interval_init(&rkb->rkb_ApiVersion_fail_intvl); rd_interval_fixed( &rkb->rkb_ApiVersion_fail_intvl, (rd_ts_t)rkb->rkb_rk->rk_conf.api_version_fallback_ms * 1000); } rd_interval_init(&rkb->rkb_suppress.unsupported_compression); rd_interval_init(&rkb->rkb_suppress.unsupported_kip62); rd_interval_init(&rkb->rkb_suppress.fail_error); #ifndef _WIN32 /* Block all signals in newly created thread. * To avoid race condition we block all signals in the calling * thread, which the new thread will inherit its sigmask from, * and then restore the original sigmask of the calling thread when * we're done creating the thread. * NOTE: term_sig remains unblocked since we use it on termination * to quickly interrupt system calls. */ sigemptyset(&oldset); sigfillset(&newset); if (rkb->rkb_rk->rk_conf.term_sig) sigdelset(&newset, rkb->rkb_rk->rk_conf.term_sig); pthread_sigmask(SIG_SETMASK, &newset, &oldset); #endif /* * Fd-based queue wake-ups using a non-blocking pipe. * Writes are best effort, if the socket queue is full * the write fails (silently) but this has no effect on latency * since the POLLIN flag will already have been raised for fd. */ rkb->rkb_wakeup_fd[0] = -1; rkb->rkb_wakeup_fd[1] = -1; rkb->rkb_toppar_wakeup_fd = -1; if ((r = rd_pipe_nonblocking(rkb->rkb_wakeup_fd)) == -1) { rd_rkb_log(rkb, LOG_ERR, "WAKEUPFD", "Failed to setup broker queue wake-up fds: " "%s: disabling low-latency mode", rd_strerror(r)); } else if (source == RD_KAFKA_INTERNAL) { /* nop: internal broker has no IO transport. */ } else { char onebyte = 1; rd_rkb_dbg(rkb, QUEUE, "WAKEUPFD", "Enabled low-latency ops queue wake-ups"); rd_kafka_q_io_event_enable(rkb->rkb_ops, rkb->rkb_wakeup_fd[1], &onebyte, sizeof(onebyte)); } /* Lock broker's lock here to synchronise state, i.e., hold off * the broker thread until we've finalized the rkb. */ rd_kafka_broker_lock(rkb); rd_kafka_broker_keep(rkb); /* broker thread's refcnt */ if (thrd_create(&rkb->rkb_thread, rd_kafka_broker_thread_main, rkb) != thrd_success) { rd_kafka_broker_unlock(rkb); rd_kafka_log(rk, LOG_CRIT, "THREAD", "Unable to create broker thread"); /* Send ERR op back to application for processing. */ rd_kafka_op_err(rk, RD_KAFKA_RESP_ERR__CRIT_SYS_RESOURCE, "Unable to create broker thread"); rd_free(rkb); #ifndef _WIN32 /* Restore sigmask of caller */ pthread_sigmask(SIG_SETMASK, &oldset, NULL); #endif return NULL; } if (rkb->rkb_source != RD_KAFKA_INTERNAL) { if (rk->rk_conf.security_protocol == RD_KAFKA_PROTO_SASL_PLAINTEXT || rk->rk_conf.security_protocol == RD_KAFKA_PROTO_SASL_SSL) rd_kafka_sasl_broker_init(rkb); /* Insert broker at head of list, idea is that * newer brokers are more relevant than old ones, * and in particular LEARNED brokers are more relevant * than CONFIGURED (bootstrap) and LOGICAL brokers. */ TAILQ_INSERT_HEAD(&rkb->rkb_rk->rk_brokers, rkb, rkb_link); (void)rd_atomic32_add(&rkb->rkb_rk->rk_broker_cnt, 1); if (rkb->rkb_nodeid != -1 && !RD_KAFKA_BROKER_IS_LOGICAL(rkb)) { 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); } rd_rkb_dbg(rkb, BROKER, "BROKER", "Added new broker with NodeId %"PRId32, rkb->rkb_nodeid); } rd_kafka_broker_unlock(rkb); /* Add broker state monitor for the coordinator request to use. * This is needed by the transactions implementation and DeleteGroups. */ rd_kafka_broker_monitor_add(&rkb->rkb_coord_monitor, rkb, rk->rk_ops, rd_kafka_coord_rkb_monitor_cb); #ifndef _WIN32 /* Restore sigmask of caller */ pthread_sigmask(SIG_SETMASK, &oldset, NULL); #endif return rkb; } /** * @brief Adds a logical broker. * * Logical brokers act just like any broker handle, but will not have * an initial address set. The address (or nodename is it is called * internally) can be set from another broker handle * by calling rd_kafka_broker_set_nodename(). * * This allows maintaining a logical group coordinator broker * handle that can ambulate between real broker addresses. * * Logical broker constraints: * - will not have a broker-id set (-1). * - will not have a port set (0). * - the address for the broker may change. * - the name of broker will not correspond to the address, * but the \p name given here. * * @returns a new broker, holding a refcount for the caller. * * @locality any rdkafka thread * @locks none */ rd_kafka_broker_t *rd_kafka_broker_add_logical (rd_kafka_t *rk, const char *name) { rd_kafka_broker_t *rkb; rd_kafka_wrlock(rk); rkb = rd_kafka_broker_add(rk, RD_KAFKA_LOGICAL, rk->rk_conf.security_protocol, name, 0/*port*/, -1/*brokerid*/); rd_assert(rkb && *"failed to create broker thread"); rd_kafka_wrunlock(rk); rd_atomic32_add(&rk->rk_broker_addrless_cnt, 1); rd_dassert(RD_KAFKA_BROKER_IS_LOGICAL(rkb)); rd_kafka_broker_keep(rkb); return rkb; } /** * @brief Update the nodename (address) of broker \p rkb * with the nodename from broker \p from_rkb (may be NULL). * * If \p rkb is connected, the connection will be torn down. * A new connection may be attempted to the new address * if a persistent connection is needed (standard connection rules). * * The broker's logname is also updated to include \p from_rkb's * broker id. * * @param from_rkb Use the nodename from this broker. If NULL, clear * the \p rkb nodename. * * @remark Must only be called for logical brokers. * * @locks none */ void rd_kafka_broker_set_nodename (rd_kafka_broker_t *rkb, rd_kafka_broker_t *from_rkb) { char nodename[RD_KAFKA_NODENAME_SIZE]; char brokername[RD_KAFKA_NODENAME_SIZE]; int32_t nodeid; rd_bool_t changed = rd_false; rd_assert(RD_KAFKA_BROKER_IS_LOGICAL(rkb)); rd_assert(rkb != from_rkb); /* Get nodename from from_rkb */ if (from_rkb) { rd_kafka_broker_lock(from_rkb); rd_strlcpy(nodename, from_rkb->rkb_nodename, sizeof(nodename)); nodeid = from_rkb->rkb_nodeid; rd_kafka_broker_unlock(from_rkb); } else { *nodename = '\0'; nodeid = -1; } /* Set nodename on rkb */ rd_kafka_broker_lock(rkb); if (strcmp(rkb->rkb_nodename, nodename)) { rd_rkb_dbg(rkb, BROKER, "NODENAME", "Broker nodename changed from \"%s\" to \"%s\"", rkb->rkb_nodename, nodename); rd_strlcpy(rkb->rkb_nodename, nodename, sizeof(rkb->rkb_nodename)); rkb->rkb_nodename_epoch++; changed = rd_true; } if (rkb->rkb_nodeid != nodeid) { rd_rkb_dbg(rkb, BROKER, "NODEID", "Broker nodeid changed from %"PRId32" to %"PRId32, rkb->rkb_nodeid, nodeid); rkb->rkb_nodeid = nodeid; } rd_kafka_broker_unlock(rkb); /* Update the log name to include (or exclude) the nodeid. * The nodeid is appended as "..logname../nodeid" */ rd_kafka_mk_brokername(brokername, sizeof(brokername), rkb->rkb_proto, rkb->rkb_name, nodeid, rkb->rkb_source); rd_kafka_broker_set_logname(rkb, brokername); if (!changed) return; if (*rkb->rkb_nodename) rd_atomic32_sub(&rkb->rkb_rk->rk_broker_addrless_cnt, 1); else rd_atomic32_add(&rkb->rkb_rk->rk_broker_addrless_cnt, 1); /* Trigger a disconnect & reconnect */ rd_kafka_broker_schedule_connection(rkb); } /** * @brief Find broker by nodeid (not -1) and * possibly filtered by state (unless -1). * * @param do_connect If sparse connections are enabled and the broker is found * but not up, a connection will be triggered. * * @locks: rd_kafka_*lock() MUST be held * @remark caller must release rkb reference by rd_kafka_broker_destroy() */ rd_kafka_broker_t * rd_kafka_broker_find_by_nodeid0_fl (const char *func, int line, rd_kafka_t *rk, int32_t nodeid, int state, rd_bool_t do_connect) { rd_kafka_broker_t *rkb; rd_kafka_broker_t skel = { .rkb_nodeid = nodeid }; if (rd_kafka_terminating(rk)) return NULL; rkb = rd_list_find(&rk->rk_broker_by_id, &skel, rd_kafka_broker_cmp_by_id); if (!rkb) return NULL; if (state != -1) { int broker_state; rd_kafka_broker_lock(rkb); broker_state = (int)rkb->rkb_state; rd_kafka_broker_unlock(rkb); if (broker_state != state) { if (do_connect && broker_state == RD_KAFKA_BROKER_STATE_INIT && rk->rk_conf.sparse_connections) rd_kafka_broker_schedule_connection(rkb); return NULL; } } rd_kafka_broker_keep_fl(func, line, rkb); return rkb; } /** * Locks: rd_kafka_rdlock(rk) must be held * NOTE: caller must release rkb reference by rd_kafka_broker_destroy() */ static rd_kafka_broker_t *rd_kafka_broker_find (rd_kafka_t *rk, rd_kafka_secproto_t proto, const char *name, uint16_t port) { rd_kafka_broker_t *rkb; char nodename[RD_KAFKA_NODENAME_SIZE]; rd_kafka_mk_nodename(nodename, sizeof(nodename), name, port); TAILQ_FOREACH(rkb, &rk->rk_brokers, rkb_link) { if (RD_KAFKA_BROKER_IS_LOGICAL(rkb)) continue; rd_kafka_broker_lock(rkb); if (!rd_kafka_terminating(rk) && rkb->rkb_proto == proto && !strcmp(rkb->rkb_nodename, nodename)) { rd_kafka_broker_keep(rkb); rd_kafka_broker_unlock(rkb); return rkb; } rd_kafka_broker_unlock(rkb); } return NULL; } /** * Parse a broker host name. * The string 'name' is modified and null-terminated portions of it * are returned in 'proto', 'host', and 'port'. * * Returns 0 on success or -1 on parse error. */ static int rd_kafka_broker_name_parse (rd_kafka_t *rk, char **name, rd_kafka_secproto_t *proto, const char **host, uint16_t *port) { char *s = *name; char *orig; char *n, *t, *t2; /* Save a temporary copy of the original name for logging purposes */ rd_strdupa(&orig, *name); /* Find end of this name (either by delimiter or end of string */ if ((n = strchr(s, ','))) *n = '\0'; else n = s + strlen(s)-1; /* Check if this looks like an url. */ if ((t = strstr(s, "://"))) { int i; /* "proto://host[:port]" */ if (t == s) { rd_kafka_log(rk, LOG_WARNING, "BROKER", "Broker name \"%s\" parse error: " "empty protocol name", orig); return -1; } /* Make protocol uppercase */ for (t2 = s ; t2 < t ; t2++) *t2 = toupper(*t2); *t = '\0'; /* Find matching protocol by name. */ for (i = 0 ; i < RD_KAFKA_PROTO_NUM ; i++) if (!rd_strcasecmp(s, rd_kafka_secproto_names[i])) break; /* Unsupported protocol */ if (i == RD_KAFKA_PROTO_NUM) { rd_kafka_log(rk, LOG_WARNING, "BROKER", "Broker name \"%s\" parse error: " "unsupported protocol \"%s\"", orig, s); return -1; } *proto = i; /* Enforce protocol */ if (rk->rk_conf.security_protocol != *proto) { rd_kafka_log(rk, LOG_WARNING, "BROKER", "Broker name \"%s\" parse error: " "protocol \"%s\" does not match " "security.protocol setting \"%s\"", orig, s, rd_kafka_secproto_names[ rk->rk_conf.security_protocol]); return -1; } /* Hostname starts here */ s = t+3; /* Ignore anything that looks like the path part of an URL */ if ((t = strchr(s, '/'))) *t = '\0'; } else *proto = rk->rk_conf.security_protocol; /* Default protocol */ *port = RD_KAFKA_PORT; /* Check if port has been specified, but try to identify IPv6 * addresses first: * t = last ':' in string * t2 = first ':' in string * If t and t2 are equal then only one ":" exists in name * and thus an IPv4 address with port specified. * Else if not equal and t is prefixed with "]" then it's an * IPv6 address with port specified. * Else no port specified. */ if ((t = strrchr(s, ':')) && ((t2 = strchr(s, ':')) == t || *(t-1) == ']')) { *t = '\0'; *port = atoi(t+1); } /* Empty host name -> localhost */ if (!*s) s = "localhost"; *host = s; *name = n+1; /* past this name. e.g., next name/delimiter to parse */ return 0; } /** * @brief Adds a (csv list of) broker(s). * Returns the number of brokers succesfully added. * * @locality any thread * @locks none */ int rd_kafka_brokers_add0 (rd_kafka_t *rk, const char *brokerlist) { char *s_copy = rd_strdup(brokerlist); char *s = s_copy; int cnt = 0; rd_kafka_broker_t *rkb; int pre_cnt = rd_atomic32_get(&rk->rk_broker_cnt); /* Parse comma-separated list of brokers. */ while (*s) { uint16_t port; const char *host; rd_kafka_secproto_t proto; if (*s == ',' || *s == ' ') { s++; continue; } if (rd_kafka_broker_name_parse(rk, &s, &proto, &host, &port) == -1) break; rd_kafka_wrlock(rk); if ((rkb = rd_kafka_broker_find(rk, proto, host, port)) && rkb->rkb_source == RD_KAFKA_CONFIGURED) { cnt++; } else if (rd_kafka_broker_add(rk, RD_KAFKA_CONFIGURED, proto, host, port, RD_KAFKA_NODEID_UA) != NULL) cnt++; /* If rd_kafka_broker_find returned a broker its * reference needs to be released * See issue #193 */ if (rkb) rd_kafka_broker_destroy(rkb); rd_kafka_wrunlock(rk); } rd_free(s_copy); if (rk->rk_conf.sparse_connections && cnt > 0 && pre_cnt == 0) { /* Sparse connections: * If this was the first set of brokers added, * select a random one to trigger the initial cluster * connection. */ rd_kafka_rdlock(rk); rd_kafka_connect_any(rk, "bootstrap servers added"); rd_kafka_rdunlock(rk); } return cnt; } int rd_kafka_brokers_add (rd_kafka_t *rk, const char *brokerlist) { return rd_kafka_brokers_add0(rk, brokerlist); } /** * @brief Adds a new broker or updates an existing one. * * @param rkbp if non-NULL, will be set to the broker object with * refcount increased, or NULL on error. * * @locks none * @locality any */ void rd_kafka_broker_update (rd_kafka_t *rk, rd_kafka_secproto_t proto, const struct rd_kafka_metadata_broker *mdb, rd_kafka_broker_t **rkbp) { rd_kafka_broker_t *rkb; char nodename[RD_KAFKA_NODENAME_SIZE]; int needs_update = 0; rd_kafka_mk_nodename(nodename, sizeof(nodename), mdb->host, mdb->port); rd_kafka_wrlock(rk); if (unlikely(rd_kafka_terminating(rk))) { /* Dont update metadata while terminating, do this * after acquiring lock for proper synchronisation */ rd_kafka_wrunlock(rk); if (rkbp) *rkbp = NULL; return; } if ((rkb = rd_kafka_broker_find_by_nodeid(rk, mdb->id))) { /* Broker matched by nodeid, see if we need to update * the hostname. */ if (strcmp(rkb->rkb_nodename, nodename)) needs_update = 1; } else if ((rkb = rd_kafka_broker_find(rk, proto, mdb->host, mdb->port))) { /* Broker matched by hostname (but not by nodeid), * update the nodeid. */ needs_update = 1; } else if ((rkb = rd_kafka_broker_add(rk, RD_KAFKA_LEARNED, proto, mdb->host, mdb->port, mdb->id))){ rd_kafka_broker_keep(rkb); } rd_kafka_wrunlock(rk); if (rkb) { /* Existing broker */ if (needs_update) { rd_kafka_op_t *rko; rko = rd_kafka_op_new(RD_KAFKA_OP_NODE_UPDATE); rd_strlcpy(rko->rko_u.node.nodename, nodename, sizeof(rko->rko_u.node.nodename)); rko->rko_u.node.nodeid = mdb->id; /* Perform a blocking op request so that all * broker-related state, such as the rk broker list, * is up to date by the time this call returns. * Ignore&destroy the response. */ rd_kafka_op_err_destroy( rd_kafka_op_req(rkb->rkb_ops, rko, -1)); } } if (rkbp) *rkbp = rkb; else if (rkb) rd_kafka_broker_destroy(rkb); } /** * @returns the broker id, or RD_KAFKA_NODEID_UA if \p rkb is NULL. * * @locality any * @locks_required none * @locks_acquired rkb_lock */ int32_t rd_kafka_broker_id (rd_kafka_broker_t *rkb) { int32_t broker_id; if (unlikely(!rkb)) return RD_KAFKA_NODEID_UA; /* Avoid locking if already on the broker thread */ if (thrd_is_current(rkb->rkb_thread)) return rkb->rkb_nodeid; rd_kafka_broker_lock(rkb); broker_id = rkb->rkb_nodeid; rd_kafka_broker_unlock(rkb); return broker_id; } /** * Returns a thread-safe temporary copy of the broker name. * Must not be called more than 4 times from the same expression. * * Locks: none * Locality: any thread */ const char *rd_kafka_broker_name (rd_kafka_broker_t *rkb) { static RD_TLS char ret[4][RD_KAFKA_NODENAME_SIZE]; static RD_TLS int reti = 0; reti = (reti + 1) % 4; mtx_lock(&rkb->rkb_logname_lock); rd_snprintf(ret[reti], sizeof(ret[reti]), "%s", rkb->rkb_logname); mtx_unlock(&rkb->rkb_logname_lock); return ret[reti]; } /** * @brief Send dummy OP to broker thread to wake it up from IO sleep. * * @locality any * @locks any */ void rd_kafka_broker_wakeup (rd_kafka_broker_t *rkb) { rd_kafka_op_t *rko = rd_kafka_op_new(RD_KAFKA_OP_WAKEUP); rd_kafka_op_set_prio(rko, RD_KAFKA_PRIO_FLASH); rd_kafka_q_enq(rkb->rkb_ops, rko); rd_rkb_dbg(rkb, QUEUE, "WAKEUP", "Wake-up"); } /** * @brief Wake up all broker threads that are in at least state \p min_state * * @locality any * @locks none: rd_kafka_*lock() MUST NOT be held * * @returns the number of broker threads woken up */ int rd_kafka_all_brokers_wakeup (rd_kafka_t *rk, int min_state) { int cnt = 0; rd_kafka_broker_t *rkb; rd_kafka_rdlock(rk); TAILQ_FOREACH(rkb, &rk->rk_brokers, rkb_link) { int do_wakeup; rd_kafka_broker_lock(rkb); do_wakeup = (int)rkb->rkb_state >= min_state; rd_kafka_broker_unlock(rkb); if (do_wakeup) { rd_kafka_broker_wakeup(rkb); cnt += 1; } } rd_kafka_rdunlock(rk); return cnt; } /** * @brief Filter out brokers that have at least one connection attempt. */ static int rd_kafka_broker_filter_never_connected (rd_kafka_broker_t *rkb, void *opaque) { return rd_atomic32_get(&rkb->rkb_c.connects); } /** * @brief Sparse connections: * Select a random broker to connect to if no brokers are up. * * This is a non-blocking call, the connection is * performed by the selected broker thread. * * @locality any * @locks rd_kafka_rdlock() MUST be held */ void rd_kafka_connect_any (rd_kafka_t *rk, const char *reason) { rd_kafka_broker_t *rkb; rd_ts_t suppr; /* Don't count connections to logical brokers since they serve * a specific purpose (group coordinator) and their connections * should not be reused for other purposes. * rd_kafka_broker_random() will not return LOGICAL brokers. */ if (rd_atomic32_get(&rk->rk_broker_up_cnt) - rd_atomic32_get(&rk->rk_logical_broker_up_cnt) > 0 || rd_atomic32_get(&rk->rk_broker_cnt) - rd_atomic32_get(&rk->rk_broker_addrless_cnt) == 0) return; mtx_lock(&rk->rk_suppress.sparse_connect_lock); suppr = rd_interval(&rk->rk_suppress.sparse_connect_random, rk->rk_conf.sparse_connect_intvl*1000, 0); mtx_unlock(&rk->rk_suppress.sparse_connect_lock); if (suppr <= 0) { rd_kafka_dbg(rk, BROKER|RD_KAFKA_DBG_GENERIC, "CONNECT", "Not selecting any broker for cluster connection: " "still suppressed for %"PRId64"ms: %s", -suppr/1000, reason); return; } /* First pass: only match brokers never connected to, * to try to exhaust the available brokers * so that an ERR_ALL_BROKERS_DOWN error can be raised. */ rkb = rd_kafka_broker_random(rk, RD_KAFKA_BROKER_STATE_INIT, rd_kafka_broker_filter_never_connected, NULL); /* Second pass: match any non-connected/non-connecting broker. */ if (!rkb) rkb = rd_kafka_broker_random(rk, RD_KAFKA_BROKER_STATE_INIT, NULL, NULL); if (!rkb) { /* No brokers matched: * this happens if there are brokers in > INIT state, * in which case they're already connecting. */ rd_kafka_dbg(rk, BROKER|RD_KAFKA_DBG_GENERIC, "CONNECT", "Cluster connection already in progress: %s", reason); return; } rd_rkb_dbg(rkb, BROKER|RD_KAFKA_DBG_GENERIC, "CONNECT", "Selected for cluster connection: " "%s (broker has %d connection attempt(s))", reason, rd_atomic32_get(&rkb->rkb_c.connects)); rd_kafka_broker_schedule_connection(rkb); rd_kafka_broker_destroy(rkb); /* refcnt from ..broker_random() */ } /** * @brief Send PURGE queue request to broker. * * @locality any * @locks none */ void rd_kafka_broker_purge_queues (rd_kafka_broker_t *rkb, int purge_flags, rd_kafka_replyq_t replyq) { rd_kafka_op_t *rko = rd_kafka_op_new(RD_KAFKA_OP_PURGE); rd_kafka_op_set_prio(rko, RD_KAFKA_PRIO_FLASH); rko->rko_replyq = replyq; rko->rko_u.purge.flags = purge_flags; rd_kafka_q_enq(rkb->rkb_ops, rko); } /** * @brief Handle purge queues request * * @locality broker thread * @locks none */ static void rd_kafka_broker_handle_purge_queues (rd_kafka_broker_t *rkb, rd_kafka_op_t *rko) { int purge_flags = rko->rko_u.purge.flags; int inflight_cnt = 0, retry_cnt = 0, outq_cnt = 0, partial_cnt = 0; rd_rkb_dbg(rkb, QUEUE|RD_KAFKA_DBG_TOPIC, "PURGE", "Purging queues with flags %s", rd_kafka_purge_flags2str(purge_flags)); /** * First purge any Produce requests to move the * messages from the request's message queue to delivery reports. */ /* Purge in-flight ProduceRequests */ if (purge_flags & RD_KAFKA_PURGE_F_INFLIGHT) inflight_cnt = rd_kafka_broker_bufq_timeout_scan( rkb, 1, &rkb->rkb_waitresps, NULL, RD_KAFKAP_Produce, RD_KAFKA_RESP_ERR__PURGE_INFLIGHT, 0, NULL, 0); if (purge_flags & RD_KAFKA_PURGE_F_QUEUE) { /* Requests in retry queue */ retry_cnt = rd_kafka_broker_bufq_timeout_scan( rkb, 0, &rkb->rkb_retrybufs, NULL, RD_KAFKAP_Produce, RD_KAFKA_RESP_ERR__PURGE_QUEUE, 0, NULL, 0); /* Requests in transmit queue not completely sent yet. * partial_cnt is included in outq_cnt and denotes a request * that has been partially transmitted. */ outq_cnt = rd_kafka_broker_bufq_timeout_scan( rkb, 0, &rkb->rkb_outbufs, &partial_cnt, RD_KAFKAP_Produce, RD_KAFKA_RESP_ERR__PURGE_QUEUE, 0, NULL, 0); /* Purging a partially transmitted request will mess up * the protocol stream, so we need to disconnect from the broker * to get a clean protocol socket. */ if (partial_cnt) rd_kafka_broker_fail( rkb, LOG_DEBUG, RD_KAFKA_RESP_ERR__PURGE_QUEUE, "Purged %d partially sent request: " "forcing disconnect", partial_cnt); } rd_rkb_dbg(rkb, QUEUE|RD_KAFKA_DBG_TOPIC, "PURGEQ", "Purged %i in-flight, %i retry-queued, " "%i out-queue, %i partially-sent requests", inflight_cnt, retry_cnt, outq_cnt, partial_cnt); /* Purge partition queues */ if (purge_flags & RD_KAFKA_PURGE_F_QUEUE) { rd_kafka_toppar_t *rktp; int msg_cnt = 0; int part_cnt = 0; TAILQ_FOREACH(rktp, &rkb->rkb_toppars, rktp_rkblink) { int r; r = rd_kafka_toppar_purge_queues( rktp, purge_flags, rd_true/*include xmit msgq*/); if (r > 0) { msg_cnt += r; part_cnt++; } } rd_rkb_dbg(rkb, QUEUE|RD_KAFKA_DBG_TOPIC, "PURGEQ", "Purged %i message(s) from %d partition(s)", msg_cnt, part_cnt); } rd_kafka_op_reply(rko, RD_KAFKA_RESP_ERR_NO_ERROR); } /** * @brief Add toppar to broker's active list list. * * For consumer this means the fetch list. * For producers this is all partitions assigned to this broker. * * @locality broker thread * @locks rktp_lock MUST be held */ void rd_kafka_broker_active_toppar_add (rd_kafka_broker_t *rkb, rd_kafka_toppar_t *rktp, const char *reason) { int is_consumer = rkb->rkb_rk->rk_type == RD_KAFKA_CONSUMER; if (is_consumer && rktp->rktp_fetch) return; /* Already added */ CIRCLEQ_INSERT_TAIL(&rkb->rkb_active_toppars, rktp, rktp_activelink); rkb->rkb_active_toppar_cnt++; if (is_consumer) rktp->rktp_fetch = 1; if (unlikely(rkb->rkb_active_toppar_cnt == 1)) rd_kafka_broker_active_toppar_next(rkb, rktp); rd_rkb_dbg(rkb, TOPIC, "FETCHADD", "Added %.*s [%"PRId32"] to %s list (%d entries, opv %d, " "%d messages queued): %s", RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic), rktp->rktp_partition, is_consumer ? "fetch" : "active", rkb->rkb_active_toppar_cnt, rktp->rktp_fetch_version, rd_kafka_msgq_len(&rktp->rktp_msgq), reason); } /** * @brief Remove toppar from active list. * * Locality: broker thread * Locks: none */ void rd_kafka_broker_active_toppar_del (rd_kafka_broker_t *rkb, rd_kafka_toppar_t *rktp, const char *reason) { int is_consumer = rkb->rkb_rk->rk_type == RD_KAFKA_CONSUMER; if (is_consumer && !rktp->rktp_fetch) return; /* Not added */ CIRCLEQ_REMOVE(&rkb->rkb_active_toppars, rktp, rktp_activelink); rd_kafka_assert(NULL, rkb->rkb_active_toppar_cnt > 0); rkb->rkb_active_toppar_cnt--; if (is_consumer) rktp->rktp_fetch = 0; if (rkb->rkb_active_toppar_next == rktp) { /* Update next pointer */ rd_kafka_broker_active_toppar_next( rkb, CIRCLEQ_LOOP_NEXT(&rkb->rkb_active_toppars, rktp, rktp_activelink)); } rd_rkb_dbg(rkb, TOPIC, "FETCHADD", "Removed %.*s [%"PRId32"] from %s list " "(%d entries, opv %d): %s", RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic), rktp->rktp_partition, is_consumer ? "fetch" : "active", rkb->rkb_active_toppar_cnt, rktp->rktp_fetch_version, reason); } /** * @brief Schedule connection for \p rkb. * Will trigger disconnection for logical brokers whose nodename * was changed. * * @locality any * @locks none */ void rd_kafka_broker_schedule_connection (rd_kafka_broker_t *rkb) { rd_kafka_op_t *rko; rko = rd_kafka_op_new(RD_KAFKA_OP_CONNECT); rd_kafka_op_set_prio(rko, RD_KAFKA_PRIO_FLASH); rd_kafka_q_enq(rkb->rkb_ops, rko); } /** * @brief Add need for persistent connection to \p rkb * with rkb_persistconn atomic counter \p acntp * * @locality any * @locks none */ void rd_kafka_broker_persistent_connection_add (rd_kafka_broker_t *rkb, rd_atomic32_t *acntp) { if (rd_atomic32_add(acntp, 1) == 1) { /* First one, trigger event. */ rd_kafka_broker_schedule_connection(rkb); } } /** * @brief Remove need for persistent connection to \p rkb * with rkb_persistconn atomic counter \p acntp * * @locality any * @locks none */ void rd_kafka_broker_persistent_connection_del (rd_kafka_broker_t *rkb, rd_atomic32_t *acntp) { int32_t r = rd_atomic32_sub(acntp, 1); rd_assert(r >= 0); } /** * @brief OP_BROKER_MONITOR callback trampoline which * calls the rkbmon's callback. * * @locality monitoree's op handler thread * @locks none */ static rd_kafka_op_res_t rd_kafka_broker_monitor_op_cb (rd_kafka_t *rk, rd_kafka_q_t *rkq, rd_kafka_op_t *rko) { if (rko->rko_err != RD_KAFKA_RESP_ERR__DESTROY) rko->rko_u.broker_monitor.cb(rko->rko_u.broker_monitor.rkb); return RD_KAFKA_OP_RES_HANDLED; } /** * @brief Trigger ops for registered monitors when the broker * state goes from or to UP. * * @locality broker thread * @locks rkb_lock MUST be held */ static void rd_kafka_broker_trigger_monitors (rd_kafka_broker_t *rkb) { rd_kafka_broker_monitor_t *rkbmon; TAILQ_FOREACH(rkbmon, &rkb->rkb_monitors, rkbmon_link) { rd_kafka_op_t *rko = rd_kafka_op_new_cb( rkb->rkb_rk, RD_KAFKA_OP_BROKER_MONITOR, rd_kafka_broker_monitor_op_cb); rd_kafka_broker_keep(rkb); rko->rko_u.broker_monitor.rkb = rkb; rko->rko_u.broker_monitor.cb = rkbmon->rkbmon_cb; rd_kafka_q_enq(rkbmon->rkbmon_q, rko); } } /** * @brief Adds a monitor for when the broker goes up or down. * * The callback will be triggered on the caller's op queue handler thread. * * Use rd_kafka_broker_is_up() in your callback to get the current * state of the broker, since it might have changed since the event * was enqueued. * * @param rkbmon monitoree's monitor. * @param rkb broker to monitor. * @param rkq queue for event op. * @param callback callback to be triggered from \p rkq's op handler. * @opaque opaque passed to callback. * * @locks none * @locality any */ void rd_kafka_broker_monitor_add (rd_kafka_broker_monitor_t *rkbmon, rd_kafka_broker_t *rkb, rd_kafka_q_t *rkq, void (*callback) (rd_kafka_broker_t *rkb)) { rd_assert(!rkbmon->rkbmon_rkb); rkbmon->rkbmon_rkb = rkb; rkbmon->rkbmon_q = rkq; rd_kafka_q_keep(rkbmon->rkbmon_q); rkbmon->rkbmon_cb = callback; rd_kafka_broker_keep(rkb); rd_kafka_broker_lock(rkb); TAILQ_INSERT_TAIL(&rkb->rkb_monitors, rkbmon, rkbmon_link); rd_kafka_broker_unlock(rkb); } /** * @brief Removes a monitor previously added with * rd_kafka_broker_monitor_add(). * * @warning The rkbmon's callback may still be called after * _del() has been called due to the buffering nature * of op queues. * * @locks none * @locality any */ void rd_kafka_broker_monitor_del (rd_kafka_broker_monitor_t *rkbmon) { rd_kafka_broker_t *rkb = rkbmon->rkbmon_rkb; if (!rkb) return; rd_kafka_broker_lock(rkb); rkbmon->rkbmon_rkb = NULL; rd_kafka_q_destroy(rkbmon->rkbmon_q); TAILQ_REMOVE(&rkb->rkb_monitors, rkbmon, rkbmon_link); rd_kafka_broker_unlock(rkb); rd_kafka_broker_destroy(rkb); } /** * @name Unit tests * @{ * */ int unittest_broker (void) { int fails = 0; fails += rd_ut_reconnect_backoff(); return fails; } /**@}*/