/* * librdkafka - Apache Kafka C library * * Copyright (c) 2012,2013 Magnus Edenhill * All rights reserved. * * Redistribution and use in source and binary forms, with or without * modification, are permitted provided that the following conditions are met: * * 1. Redistributions of source code must retain the above copyright notice, * this list of conditions and the following disclaimer. * 2. Redistributions in binary form must reproduce the above copyright notice, * this list of conditions and the following disclaimer in the documentation * and/or other materials provided with the distribution. * * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE * POSSIBILITY OF SUCH DAMAGE. */ #include "rdkafka_int.h" #include "rd.h" #include #include #include #include "rdkafka_int.h" #include "rdkafka_feature.h" #include "rdkafka_interceptor.h" #if WITH_PLUGINS #include "rdkafka_plugin.h" #endif struct rd_kafka_property { rd_kafka_conf_scope_t scope; const char *name; enum { _RK_C_STR, _RK_C_INT, _RK_C_S2I, /* String to Integer mapping. * Supports limited canonical str->int mappings * using s2i[] */ _RK_C_S2F, /* CSV String to Integer flag mapping (OR:ed) */ _RK_C_BOOL, _RK_C_PTR, /* Only settable through special set functions */ _RK_C_PATLIST, /* Pattern list */ _RK_C_KSTR, /* Kafka string */ _RK_C_ALIAS, /* Alias: points to other property through .sdef */ _RK_C_INTERNAL, /* Internal, don't expose to application */ _RK_C_INVALID, /* Invalid property, used to catch known * but unsupported Java properties. */ } type; int offset; const char *desc; int vmin; int vmax; int vdef; /* Default value (int) */ const char *sdef; /* Default value (string) */ void *pdef; /* Default value (pointer) */ struct { int val; const char *str; } s2i[16]; /* _RK_C_S2I and _RK_C_S2F */ /* Value validator (STR) */ int (*validate) (const struct rd_kafka_property *prop, const char *val, int ival); /* Configuration object constructors and destructor for use when * the property value itself is not used, or needs extra care. */ void (*ctor) (int scope, void *pconf); void (*dtor) (int scope, void *pconf); void (*copy) (int scope, void *pdst, const void *psrc, void *dstptr, const void *srcptr, size_t filter_cnt, const char **filter); rd_kafka_conf_res_t (*set) (int scope, void *pconf, const char *name, const char *value, void *dstptr, rd_kafka_conf_set_mode_t set_mode, char *errstr, size_t errstr_size); }; #define _RK(field) offsetof(rd_kafka_conf_t, field) #define _RKT(field) offsetof(rd_kafka_topic_conf_t, field) static rd_kafka_conf_res_t rd_kafka_anyconf_get0 (const void *conf, const struct rd_kafka_property *prop, char *dest, size_t *dest_size); /** * @brief Validate \p broker.version.fallback property. */ static int rd_kafka_conf_validate_broker_version (const struct rd_kafka_property *prop, const char *val, int ival) { struct rd_kafka_ApiVersion *apis; size_t api_cnt; return rd_kafka_get_legacy_ApiVersions(val, &apis, &api_cnt, NULL); } /** * @brief Validate that string is a single item, without delimters (, space). */ static RD_UNUSED int rd_kafka_conf_validate_single (const struct rd_kafka_property *prop, const char *val, int ival) { return !strchr(val, ',') && !strchr(val, ' '); } /** * @brief Validate builtin partitioner string */ static RD_UNUSED int rd_kafka_conf_validate_partitioner (const struct rd_kafka_property *prop, const char *val, int ival) { return !strcmp(val, "random") || !strcmp(val, "consistent") || !strcmp(val, "consistent_random") || !strcmp(val, "murmur2") || !strcmp(val, "murmur2_random"); } /** * librdkafka configuration property definitions. */ static const struct rd_kafka_property rd_kafka_properties[] = { /* Global properties */ { _RK_GLOBAL, "builtin.features", _RK_C_S2F, _RK(builtin_features), "Indicates the builtin features for this build of librdkafka. " "An application can either query this value or attempt to set it " "with its list of required features to check for library support.", 0, 0x7fffffff, 0xffff, .s2i = { #if WITH_ZLIB { 0x1, "gzip" }, #endif #if WITH_SNAPPY { 0x2, "snappy" }, #endif #if WITH_SSL { 0x4, "ssl" }, #endif { 0x8, "sasl" }, { 0x10, "regex" }, { 0x20, "lz4" }, #if defined(_MSC_VER) || WITH_SASL_CYRUS { 0x40, "sasl_gssapi" }, #endif { 0x80, "sasl_plain" }, #if WITH_SASL_SCRAM { 0x100, "sasl_scram" }, #endif #if WITH_PLUGINS { 0x200, "plugins" }, #endif { 0, NULL } } }, { _RK_GLOBAL, "client.id", _RK_C_STR, _RK(client_id_str), "Client identifier.", .sdef = "rdkafka" }, { _RK_GLOBAL, "metadata.broker.list", _RK_C_STR, _RK(brokerlist), "Initial list of brokers as a CSV list of broker host or host:port. " "The application may also use `rd_kafka_brokers_add()` to add " "brokers during runtime." }, { _RK_GLOBAL, "bootstrap.servers", _RK_C_ALIAS, 0, "See metadata.broker.list", .sdef = "metadata.broker.list" }, { _RK_GLOBAL, "message.max.bytes", _RK_C_INT, _RK(max_msg_size), "Maximum Kafka protocol request message size.", 1000, 1000000000, 1000000 }, { _RK_GLOBAL, "message.copy.max.bytes", _RK_C_INT, _RK(msg_copy_max_size), "Maximum size for message to be copied to buffer. " "Messages larger than this will be passed by reference (zero-copy) " "at the expense of larger iovecs.", 0, 1000000000, 0xffff }, { _RK_GLOBAL, "receive.message.max.bytes", _RK_C_INT, _RK(recv_max_msg_size), "Maximum Kafka protocol response message size. " "This serves as a safety precaution to avoid memory exhaustion in " "case of protocol hickups. " "This value is automatically adjusted upwards to be at least " "`fetch.max.bytes` + 512 to allow for protocol overhead.", 1000, INT_MAX, 100000000 }, { _RK_GLOBAL, "max.in.flight.requests.per.connection", _RK_C_INT, _RK(max_inflight), "Maximum number of in-flight requests per broker connection. " "This is a generic property applied to all broker communication, " "however it is primarily relevant to produce requests. " "In particular, note that other mechanisms limit the number " "of outstanding consumer fetch request per broker to one.", 1, 1000000, 1000000 }, { _RK_GLOBAL, "max.in.flight", _RK_C_ALIAS, .sdef = "max.in.flight.requests.per.connection" }, { _RK_GLOBAL, "metadata.request.timeout.ms", _RK_C_INT, _RK(metadata_request_timeout_ms), "Non-topic request timeout in milliseconds. " "This is for metadata requests, etc.", 10, 900*1000, 60*1000}, { _RK_GLOBAL, "topic.metadata.refresh.interval.ms", _RK_C_INT, _RK(metadata_refresh_interval_ms), "Topic metadata refresh interval in milliseconds. " "The metadata is automatically refreshed on error and connect. " "Use -1 to disable the intervalled refresh.", -1, 3600*1000, 5*60*1000 }, { _RK_GLOBAL, "metadata.max.age.ms", _RK_C_INT, _RK(metadata_max_age_ms), "Metadata cache max age. " "Defaults to metadata.refresh.interval.ms * 3", 1, 24*3600*1000, -1 }, { _RK_GLOBAL, "topic.metadata.refresh.fast.interval.ms", _RK_C_INT, _RK(metadata_refresh_fast_interval_ms), "When a topic loses its leader a new metadata request will be " "enqueued with this initial interval, exponentially increasing " "until the topic metadata has been refreshed. " "This is used to recover quickly from transitioning leader brokers.", 1, 60*1000, 250 }, { _RK_GLOBAL, "topic.metadata.refresh.fast.cnt", _RK_C_INT, _RK(metadata_refresh_fast_cnt), "*Deprecated: No longer used.*", 0, 1000, 10 }, { _RK_GLOBAL, "topic.metadata.refresh.sparse", _RK_C_BOOL, _RK(metadata_refresh_sparse), "Sparse metadata requests (consumes less network bandwidth)", 0, 1, 1 }, { _RK_GLOBAL, "topic.blacklist", _RK_C_PATLIST, _RK(topic_blacklist), "Topic blacklist, a comma-separated list of regular expressions " "for matching topic names that should be ignored in " "broker metadata information as if the topics did not exist." }, { _RK_GLOBAL, "debug", _RK_C_S2F, _RK(debug), "A comma-separated list of debug contexts to enable. " "Detailed Producer debugging: broker,topic,msg. " "Consumer: consumer,cgrp,topic,fetch", .s2i = { { RD_KAFKA_DBG_GENERIC, "generic" }, { RD_KAFKA_DBG_BROKER, "broker" }, { RD_KAFKA_DBG_TOPIC, "topic" }, { RD_KAFKA_DBG_METADATA, "metadata" }, { RD_KAFKA_DBG_FEATURE, "feature" }, { RD_KAFKA_DBG_QUEUE, "queue" }, { RD_KAFKA_DBG_MSG, "msg" }, { RD_KAFKA_DBG_PROTOCOL, "protocol" }, { RD_KAFKA_DBG_CGRP, "cgrp" }, { RD_KAFKA_DBG_SECURITY, "security" }, { RD_KAFKA_DBG_FETCH, "fetch" }, { RD_KAFKA_DBG_INTERCEPTOR, "interceptor" }, { RD_KAFKA_DBG_PLUGIN, "plugin" }, { RD_KAFKA_DBG_CONSUMER, "consumer" }, { RD_KAFKA_DBG_ALL, "all" } } }, { _RK_GLOBAL, "socket.timeout.ms", _RK_C_INT, _RK(socket_timeout_ms), "Default timeout for network requests. " "Producer: ProduceRequests will use the lesser value of " "socket.timeout.ms and remaining message.timeout.ms for the " "first message in the batch. " "Consumer: FetchRequests will use " "fetch.wait.max.ms + socket.timeout.ms. ", 10, 300*1000, 60*1000 }, { _RK_GLOBAL, "socket.blocking.max.ms", _RK_C_INT, _RK(socket_blocking_max_ms), "Maximum time a broker socket operation may block. " "A lower value improves responsiveness at the expense of " "slightly higher CPU usage. **Deprecated**", 1, 60*1000, 1000 }, { _RK_GLOBAL, "socket.send.buffer.bytes", _RK_C_INT, _RK(socket_sndbuf_size), "Broker socket send buffer size. System default is used if 0.", 0, 100000000, 0 }, { _RK_GLOBAL, "socket.receive.buffer.bytes", _RK_C_INT, _RK(socket_rcvbuf_size), "Broker socket receive buffer size. System default is used if 0.", 0, 100000000, 0 }, { _RK_GLOBAL, "socket.keepalive.enable", _RK_C_BOOL, _RK(socket_keepalive), "Enable TCP keep-alives (SO_KEEPALIVE) on broker sockets", 0, 1, 0 }, { _RK_GLOBAL, "socket.nagle.disable", _RK_C_BOOL, _RK(socket_nagle_disable), "Disable the Nagle algorithm (TCP_NODELAY).", 0, 1, 0 }, { _RK_GLOBAL, "socket.max.fails", _RK_C_INT, _RK(socket_max_fails), "Disconnect from broker when this number of send failures " "(e.g., timed out requests) is reached. Disable with 0. " "NOTE: The connection is automatically re-established.", 0, 1000000, 1 }, { _RK_GLOBAL, "broker.address.ttl", _RK_C_INT, _RK(broker_addr_ttl), "How long to cache the broker address resolving " "results (milliseconds).", 0, 86400*1000, 1*1000 }, { _RK_GLOBAL, "broker.address.family", _RK_C_S2I, _RK(broker_addr_family), "Allowed broker IP address families: any, v4, v6", .vdef = AF_UNSPEC, .s2i = { { AF_UNSPEC, "any" }, { AF_INET, "v4" }, { AF_INET6, "v6" }, } }, { _RK_GLOBAL, "reconnect.backoff.jitter.ms", _RK_C_INT, _RK(reconnect_jitter_ms), "Throttle broker reconnection attempts by this value +-50%.", 0, 60*60*1000, 500 }, { _RK_GLOBAL, "statistics.interval.ms", _RK_C_INT, _RK(stats_interval_ms), "librdkafka statistics emit interval. The application also needs to " "register a stats callback using `rd_kafka_conf_set_stats_cb()`. " "The granularity is 1000ms. A value of 0 disables statistics.", 0, 86400*1000, 0 }, { _RK_GLOBAL, "enabled_events", _RK_C_INT, _RK(enabled_events), "See `rd_kafka_conf_set_events()`", 0, 0x7fffffff, 0 }, { _RK_GLOBAL, "error_cb", _RK_C_PTR, _RK(error_cb), "Error callback (set with rd_kafka_conf_set_error_cb())" }, { _RK_GLOBAL, "throttle_cb", _RK_C_PTR, _RK(throttle_cb), "Throttle callback (set with rd_kafka_conf_set_throttle_cb())" }, { _RK_GLOBAL, "stats_cb", _RK_C_PTR, _RK(stats_cb), "Statistics callback (set with rd_kafka_conf_set_stats_cb())" }, { _RK_GLOBAL, "log_cb", _RK_C_PTR, _RK(log_cb), "Log callback (set with rd_kafka_conf_set_log_cb())", .pdef = rd_kafka_log_print }, { _RK_GLOBAL, "log_level", _RK_C_INT, _RK(log_level), "Logging level (syslog(3) levels)", 0, 7, 6 }, { _RK_GLOBAL, "log.queue", _RK_C_BOOL, _RK(log_queue), "Disable spontaneous log_cb from internal librdkafka " "threads, instead enqueue log messages on queue set with " "`rd_kafka_set_log_queue()` and serve log callbacks or " "events through the standard poll APIs. " "**NOTE**: Log messages will linger in a temporary queue " "until the log queue has been set.", 0, 1, 0 }, { _RK_GLOBAL, "log.thread.name", _RK_C_BOOL, _RK(log_thread_name), "Print internal thread name in log messages " "(useful for debugging librdkafka internals)", 0, 1, 1 }, { _RK_GLOBAL, "log.connection.close", _RK_C_BOOL, _RK(log_connection_close), "Log broker disconnects. " "It might be useful to turn this off when interacting with " "0.9 brokers with an aggressive `connection.max.idle.ms` value.", 0, 1, 1 }, { _RK_GLOBAL, "socket_cb", _RK_C_PTR, _RK(socket_cb), "Socket creation callback to provide race-free CLOEXEC", .pdef = #ifdef __linux__ rd_kafka_socket_cb_linux #else rd_kafka_socket_cb_generic #endif }, { _RK_GLOBAL, "connect_cb", _RK_C_PTR, _RK(connect_cb), "Socket connect callback", }, { _RK_GLOBAL, "closesocket_cb", _RK_C_PTR, _RK(closesocket_cb), "Socket close callback", }, { _RK_GLOBAL, "open_cb", _RK_C_PTR, _RK(open_cb), "File open callback to provide race-free CLOEXEC", .pdef = #ifdef __linux__ rd_kafka_open_cb_linux #else rd_kafka_open_cb_generic #endif }, { _RK_GLOBAL, "opaque", _RK_C_PTR, _RK(opaque), "Application opaque (set with rd_kafka_conf_set_opaque())" }, { _RK_GLOBAL, "default_topic_conf", _RK_C_PTR, _RK(topic_conf), "Default topic configuration for automatically subscribed topics" }, { _RK_GLOBAL, "internal.termination.signal", _RK_C_INT, _RK(term_sig), "Signal that librdkafka will use to quickly terminate on " "rd_kafka_destroy(). If this signal is not set then there will be a " "delay before rd_kafka_wait_destroyed() returns true " "as internal threads are timing out their system calls. " "If this signal is set however the delay will be minimal. " "The application should mask this signal as an internal " "signal handler is installed.", 0, 128, 0 }, { _RK_GLOBAL, "api.version.request", _RK_C_BOOL, _RK(api_version_request), "Request broker's supported API versions to adjust functionality to " "available protocol features. If set to false, or the " "ApiVersionRequest fails, the fallback version " "`broker.version.fallback` will be used. " "**NOTE**: Depends on broker version >=0.10.0. If the request is not " "supported by (an older) broker the `broker.version.fallback` fallback is used.", 0, 1, 1 }, { _RK_GLOBAL, "api.version.request.timeout.ms", _RK_C_INT, _RK(api_version_request_timeout_ms), "Timeout for broker API version requests.", 1, 5*60*1000, 10*1000 }, { _RK_GLOBAL, "api.version.fallback.ms", _RK_C_INT, _RK(api_version_fallback_ms), "Dictates how long the `broker.version.fallback` fallback is used " "in the case the ApiVersionRequest fails. " "**NOTE**: The ApiVersionRequest is only issued when a new connection " "to the broker is made (such as after an upgrade).", 0, 86400*7*1000, 20*60*1000 /* longer than default Idle timeout (10m)*/ }, { _RK_GLOBAL, "broker.version.fallback", _RK_C_STR, _RK(broker_version_fallback), "Older broker versions (<0.10.0) provides no way for a client to query " "for supported protocol features " "(ApiVersionRequest, see `api.version.request`) making it impossible " "for the client to know what features it may use. " "As a workaround a user may set this property to the expected broker " "version and the client will automatically adjust its feature set " "accordingly if the ApiVersionRequest fails (or is disabled). " "The fallback broker version will be used for `api.version.fallback.ms`. " "Valid values are: 0.9.0, 0.8.2, 0.8.1, 0.8.0. Any other value, " "such as 0.10.2.1, enables ApiVersionRequests.", .sdef = "0.9.0", .validate = rd_kafka_conf_validate_broker_version }, /* Security related global properties */ { _RK_GLOBAL, "security.protocol", _RK_C_S2I, _RK(security_protocol), "Protocol used to communicate with brokers.", .vdef = RD_KAFKA_PROTO_PLAINTEXT, .s2i = { { RD_KAFKA_PROTO_PLAINTEXT, "plaintext" }, #if WITH_SSL { RD_KAFKA_PROTO_SSL, "ssl" }, #endif { RD_KAFKA_PROTO_SASL_PLAINTEXT, "sasl_plaintext" }, #if WITH_SSL { RD_KAFKA_PROTO_SASL_SSL, "sasl_ssl" }, #endif { 0, NULL } } }, #if WITH_SSL { _RK_GLOBAL, "ssl.cipher.suites", _RK_C_STR, _RK(ssl.cipher_suites), "A cipher suite is a named combination of authentication, " "encryption, MAC and key exchange algorithm used to negotiate the " "security settings for a network connection using TLS or SSL network " "protocol. See manual page for `ciphers(1)` and " "`SSL_CTX_set_cipher_list(3)." }, { _RK_GLOBAL, "ssl.key.location", _RK_C_STR, _RK(ssl.key_location), "Path to client's private key (PEM) used for authentication." }, { _RK_GLOBAL, "ssl.key.password", _RK_C_STR, _RK(ssl.key_password), "Private key passphrase" }, { _RK_GLOBAL, "ssl.certificate.location", _RK_C_STR, _RK(ssl.cert_location), "Path to client's public key (PEM) used for authentication." }, { _RK_GLOBAL, "ssl.ca.location", _RK_C_STR, _RK(ssl.ca_location), "File or directory path to CA certificate(s) for verifying " "the broker's key." }, { _RK_GLOBAL, "ssl.crl.location", _RK_C_STR, _RK(ssl.crl_location), "Path to CRL for verifying broker's certificate validity." }, { _RK_GLOBAL, "ssl.keystore.location", _RK_C_STR, _RK(ssl.keystore_location), "Path to client's keystore (PKCS#12) used for authentication." }, { _RK_GLOBAL, "ssl.keystore.password", _RK_C_STR, _RK(ssl.keystore_password), "Client's keystore (PKCS#12) password." }, #endif /* WITH_SSL */ /* Point user in the right direction if they try to apply * Java client SSL / JAAS properties. */ { _RK_GLOBAL, "ssl.truststore.location", _RK_C_INVALID, _RK(dummy), "Java TrustStores are not supported, use `ssl.ca.location` " "and a certificate file instead. " "See https://github.com/edenhill/librdkafka/wiki/Using-SSL-with-librdkafka for more information." }, { _RK_GLOBAL, "sasl.jaas.config", _RK_C_INVALID, _RK(dummy), "Java JAAS configuration is not supported, see " "https://github.com/edenhill/librdkafka/wiki/Using-SASL-with-librdkafka " "for more information." }, {_RK_GLOBAL,"sasl.mechanisms", _RK_C_STR, _RK(sasl.mechanisms), "SASL mechanism to use for authentication. " "Supported: GSSAPI, PLAIN, SCRAM-SHA-256, SCRAM-SHA-512. " "**NOTE**: Despite the name only one mechanism must be configured.", .sdef = "GSSAPI", .validate = rd_kafka_conf_validate_single }, {_RK_GLOBAL,"sasl.mechanism", _RK_C_ALIAS, .sdef = "sasl.mechanisms" }, { _RK_GLOBAL, "sasl.kerberos.service.name", _RK_C_STR, _RK(sasl.service_name), "Kerberos principal name that Kafka runs as, " "not including /hostname@REALM", .sdef = "kafka" }, { _RK_GLOBAL, "sasl.kerberos.principal", _RK_C_STR, _RK(sasl.principal), "This client's Kerberos principal name. " "(Not supported on Windows, will use the logon user's principal).", .sdef = "kafkaclient" }, #ifndef _MSC_VER { _RK_GLOBAL, "sasl.kerberos.kinit.cmd", _RK_C_STR, _RK(sasl.kinit_cmd), "Full kerberos kinit command string, %{config.prop.name} is replaced " "by corresponding config object value, %{broker.name} returns the " "broker's hostname.", .sdef = "kinit -S \"%{sasl.kerberos.service.name}/%{broker.name}\" " "-k -t \"%{sasl.kerberos.keytab}\" %{sasl.kerberos.principal}" }, { _RK_GLOBAL, "sasl.kerberos.keytab", _RK_C_STR, _RK(sasl.keytab), "Path to Kerberos keytab file. Uses system default if not set." "**NOTE**: This is not automatically used but must be added to the " "template in sasl.kerberos.kinit.cmd as " "` ... -t %{sasl.kerberos.keytab}`." }, { _RK_GLOBAL, "sasl.kerberos.min.time.before.relogin", _RK_C_INT, _RK(sasl.relogin_min_time), "Minimum time in milliseconds between key refresh attempts.", 1, 86400*1000, 60*1000 }, #endif { _RK_GLOBAL, "sasl.username", _RK_C_STR, _RK(sasl.username), "SASL username for use with the PLAIN and SASL-SCRAM-.. mechanisms" }, { _RK_GLOBAL, "sasl.password", _RK_C_STR, _RK(sasl.password), "SASL password for use with the PLAIN and SASL-SCRAM-.. mechanism" }, #if WITH_PLUGINS /* Plugins */ { _RK_GLOBAL, "plugin.library.paths", _RK_C_STR, _RK(plugin_paths), "List of plugin libaries to load (; separated). " "The library search path is platform dependent (see dlopen(3) for Unix and LoadLibrary() for Windows). If no filename extension is specified the " "platform-specific extension (such as .dll or .so) will be appended automatically.", .set = rd_kafka_plugins_conf_set }, #endif /* Interceptors are added through specific API and not exposed * as configuration properties. * The interceptor property must be defined after plugin.library.paths * so that the plugin libraries are properly loaded before * interceptors are configured when duplicating configuration objects.*/ { _RK_GLOBAL, "interceptors", _RK_C_INTERNAL, _RK(interceptors), "Interceptors added through rd_kafka_conf_interceptor_add_..() " "and any configuration handled by interceptors.", .ctor = rd_kafka_conf_interceptor_ctor, .dtor = rd_kafka_conf_interceptor_dtor, .copy = rd_kafka_conf_interceptor_copy }, /* Global client group properties */ { _RK_GLOBAL|_RK_CGRP, "group.id", _RK_C_STR, _RK(group_id_str), "Client group id string. All clients sharing the same group.id " "belong to the same group." }, { _RK_GLOBAL|_RK_CGRP, "partition.assignment.strategy", _RK_C_STR, _RK(partition_assignment_strategy), "Name of partition assignment strategy to use when elected " "group leader assigns partitions to group members.", .sdef = "range,roundrobin" }, { _RK_GLOBAL|_RK_CGRP, "session.timeout.ms", _RK_C_INT, _RK(group_session_timeout_ms), "Client group session and failure detection timeout.", 1, 3600*1000, 30*1000 }, { _RK_GLOBAL|_RK_CGRP, "heartbeat.interval.ms", _RK_C_INT, _RK(group_heartbeat_intvl_ms), "Group session keepalive heartbeat interval.", 1, 3600*1000, 1*1000 }, { _RK_GLOBAL|_RK_CGRP, "group.protocol.type", _RK_C_KSTR, _RK(group_protocol_type), "Group protocol type", .sdef = "consumer" }, { _RK_GLOBAL|_RK_CGRP, "coordinator.query.interval.ms", _RK_C_INT, _RK(coord_query_intvl_ms), "How often to query for the current client group coordinator. " "If the currently assigned coordinator is down the configured " "query interval will be divided by ten to more quickly recover " "in case of coordinator reassignment.", 1, 3600*1000, 10*60*1000 }, /* Global consumer properties */ { _RK_GLOBAL|_RK_CONSUMER, "enable.auto.commit", _RK_C_BOOL, _RK(enable_auto_commit), "Automatically and periodically commit offsets in the background. " "Note: setting this to false does not prevent the consumer from " "fetching previously committed start offsets. To circumvent this " "behaviour set specific start offsets per partition in the call " "to assign().", 0, 1, 1 }, { _RK_GLOBAL|_RK_CONSUMER, "auto.commit.interval.ms", _RK_C_INT, _RK(auto_commit_interval_ms), "The frequency in milliseconds that the consumer offsets " "are committed (written) to offset storage. (0 = disable). " "This setting is used by the high-level consumer.", 0, 86400*1000, 5*1000 }, { _RK_GLOBAL|_RK_CONSUMER, "enable.auto.offset.store", _RK_C_BOOL, _RK(enable_auto_offset_store), "Automatically store offset of last message provided to " "application.", 0, 1, 1 }, { _RK_GLOBAL|_RK_CONSUMER, "queued.min.messages", _RK_C_INT, _RK(queued_min_msgs), "Minimum number of messages per topic+partition " "librdkafka tries to maintain in the local consumer queue.", 1, 10000000, 100000 }, { _RK_GLOBAL|_RK_CONSUMER, "queued.max.messages.kbytes", _RK_C_INT, _RK(queued_max_msg_kbytes), "Maximum number of kilobytes per topic+partition in the " "local consumer queue. " "This value may be overshot by fetch.message.max.bytes. " "This property has higher priority than queued.min.messages.", 1, INT_MAX/1024, 0x100000/*1GB*/ }, { _RK_GLOBAL|_RK_CONSUMER, "fetch.wait.max.ms", _RK_C_INT, _RK(fetch_wait_max_ms), "Maximum time the broker may wait to fill the response " "with fetch.min.bytes.", 0, 300*1000, 100 }, { _RK_GLOBAL|_RK_CONSUMER, "fetch.message.max.bytes", _RK_C_INT, _RK(fetch_msg_max_bytes), "Initial maximum number of bytes per topic+partition to request when " "fetching messages from the broker. " "If the client encounters a message larger than this value " "it will gradually try to increase it until the " "entire message can be fetched.", 1, 1000000000, 1024*1024 }, { _RK_GLOBAL|_RK_CONSUMER, "max.partition.fetch.bytes", _RK_C_ALIAS, .sdef = "fetch.message.max.bytes" }, { _RK_GLOBAL|_RK_CONSUMER, "fetch.max.bytes", _RK_C_INT, _RK(fetch_max_bytes), "Maximum amount of data the broker shall return for a Fetch request. " "Messages are fetched in batches by the consumer and if the first " "message batch in the first non-empty partition of the Fetch request " "is larger than this value, then the message batch will still be " "returned to ensure the consumer can make progress. " "The maximum message batch size accepted by the broker is defined " "via `message.max.bytes` (broker config) or " "`max.message.bytes` (broker topic config). " "`fetch.max.bytes` is automatically adjusted upwards to be " "at least `message.max.bytes` (consumer config).", 0, INT_MAX-512, 50*1024*1024 /* 50MB */ }, { _RK_GLOBAL|_RK_CONSUMER, "fetch.min.bytes", _RK_C_INT, _RK(fetch_min_bytes), "Minimum number of bytes the broker responds with. " "If fetch.wait.max.ms expires the accumulated data will " "be sent to the client regardless of this setting.", 1, 100000000, 1 }, { _RK_GLOBAL|_RK_CONSUMER, "fetch.error.backoff.ms", _RK_C_INT, _RK(fetch_error_backoff_ms), "How long to postpone the next fetch request for a " "topic+partition in case of a fetch error.", 0, 300*1000, 500 }, { _RK_GLOBAL|_RK_CONSUMER, "offset.store.method", _RK_C_S2I, _RK(offset_store_method), "Offset commit store method: " "'file' - local file store (offset.store.path, et.al), " "'broker' - broker commit store " "(requires Apache Kafka 0.8.2 or later on the broker).", .vdef = RD_KAFKA_OFFSET_METHOD_BROKER, .s2i = { { RD_KAFKA_OFFSET_METHOD_NONE, "none" }, { RD_KAFKA_OFFSET_METHOD_FILE, "file" }, { RD_KAFKA_OFFSET_METHOD_BROKER, "broker" } } }, { _RK_GLOBAL|_RK_CONSUMER, "consume_cb", _RK_C_PTR, _RK(consume_cb), "Message consume callback (set with rd_kafka_conf_set_consume_cb())"}, { _RK_GLOBAL|_RK_CONSUMER, "rebalance_cb", _RK_C_PTR, _RK(rebalance_cb), "Called after consumer group has been rebalanced " "(set with rd_kafka_conf_set_rebalance_cb())" }, { _RK_GLOBAL|_RK_CONSUMER, "offset_commit_cb", _RK_C_PTR, _RK(offset_commit_cb), "Offset commit result propagation callback. " "(set with rd_kafka_conf_set_offset_commit_cb())" }, { _RK_GLOBAL|_RK_CONSUMER, "enable.partition.eof", _RK_C_BOOL, _RK(enable_partition_eof), "Emit RD_KAFKA_RESP_ERR__PARTITION_EOF event whenever the " "consumer reaches the end of a partition.", 0, 1, 1 }, { _RK_GLOBAL|_RK_CONSUMER, "check.crcs", _RK_C_BOOL, _RK(check_crcs), "Verify CRC32 of consumed messages, ensuring no on-the-wire or " "on-disk corruption to the messages occurred. This check comes " "at slightly increased CPU usage.", 0, 1, 0 }, /* Global producer properties */ { _RK_GLOBAL|_RK_PRODUCER, "queue.buffering.max.messages", _RK_C_INT, _RK(queue_buffering_max_msgs), "Maximum number of messages allowed on the producer queue.", 1, 10000000, 100000 }, { _RK_GLOBAL|_RK_PRODUCER, "queue.buffering.max.kbytes", _RK_C_INT, _RK(queue_buffering_max_kbytes), "Maximum total message size sum allowed on the producer queue. " "This property has higher priority than queue.buffering.max.messages.", 1, INT_MAX/1024, 0x100000/*1GB*/ }, { _RK_GLOBAL|_RK_PRODUCER, "queue.buffering.max.ms", _RK_C_INT, _RK(buffering_max_ms), "Delay in milliseconds to wait for messages in the producer queue " "to accumulate before constructing message batches (MessageSets) to " "transmit to brokers. " "A higher value allows larger and more effective " "(less overhead, improved compression) batches of messages to " "accumulate at the expense of increased message delivery latency.", 0, 900*1000, 0 }, { _RK_GLOBAL|_RK_PRODUCER, "linger.ms", _RK_C_ALIAS, .sdef = "queue.buffering.max.ms" }, { _RK_GLOBAL|_RK_PRODUCER, "message.send.max.retries", _RK_C_INT, _RK(max_retries), "How many times to retry sending a failing MessageSet. " "**Note:** retrying may cause reordering.", 0, 10000000, 2 }, { _RK_GLOBAL | _RK_PRODUCER, "retries", _RK_C_ALIAS, .sdef = "message.send.max.retries" }, { _RK_GLOBAL|_RK_PRODUCER, "retry.backoff.ms", _RK_C_INT, _RK(retry_backoff_ms), "The backoff time in milliseconds before retrying a protocol request.", 1, 300*1000, 100 }, { _RK_GLOBAL|_RK_PRODUCER, "queue.buffering.backpressure.threshold", _RK_C_INT, _RK(queue_backpressure_thres), "The threshold of outstanding not yet transmitted requests " "needed to backpressure the producer's message accumulator. " "A lower number yields larger and more effective batches.", 0, 1000000, 10 }, { _RK_GLOBAL|_RK_PRODUCER, "compression.codec", _RK_C_S2I, _RK(compression_codec), "compression codec to use for compressing message sets. " "This is the default value for all topics, may be overriden by " "the topic configuration property `compression.codec`. ", .vdef = RD_KAFKA_COMPRESSION_NONE, .s2i = { { RD_KAFKA_COMPRESSION_NONE, "none" }, #if WITH_ZLIB { RD_KAFKA_COMPRESSION_GZIP, "gzip" }, #endif #if WITH_SNAPPY { RD_KAFKA_COMPRESSION_SNAPPY, "snappy" }, #endif { RD_KAFKA_COMPRESSION_LZ4, "lz4" }, { 0 } } }, { _RK_GLOBAL|_RK_PRODUCER, "compression.type", _RK_C_ALIAS, .sdef = "compression.codec" }, { _RK_GLOBAL|_RK_PRODUCER, "batch.num.messages", _RK_C_INT, _RK(batch_num_messages), "Maximum number of messages batched in one MessageSet. " "The total MessageSet size is also limited by message.max.bytes.", 1, 1000000, 10000 }, { _RK_GLOBAL|_RK_PRODUCER, "delivery.report.only.error", _RK_C_BOOL, _RK(dr_err_only), "Only provide delivery reports for failed messages.", 0, 1, 0 }, { _RK_GLOBAL|_RK_PRODUCER, "dr_cb", _RK_C_PTR, _RK(dr_cb), "Delivery report callback (set with rd_kafka_conf_set_dr_cb())" }, { _RK_GLOBAL|_RK_PRODUCER, "dr_msg_cb", _RK_C_PTR, _RK(dr_msg_cb), "Delivery report callback (set with rd_kafka_conf_set_dr_msg_cb())" }, /* * Topic properties */ /* Topic producer properties */ { _RK_TOPIC|_RK_PRODUCER, "request.required.acks", _RK_C_INT, _RKT(required_acks), "This field indicates how many acknowledgements the leader broker " "must receive from ISR brokers before responding to the request: " "*0*=Broker does not send any response/ack to client, " "*1*=Only the leader broker will need to ack the message, " "*-1* or *all*=broker will block until message is committed by all " "in sync replicas (ISRs) or broker's `min.insync.replicas` " "setting before sending response. ", -1, 1000, 1, .s2i = { { -1, "all" }, } }, { _RK_TOPIC | _RK_PRODUCER, "acks", _RK_C_ALIAS, .sdef = "request.required.acks" }, { _RK_TOPIC|_RK_PRODUCER, "request.timeout.ms", _RK_C_INT, _RKT(request_timeout_ms), "The ack timeout of the producer request in milliseconds. " "This value is only enforced by the broker and relies " "on `request.required.acks` being != 0.", 1, 900*1000, 5*1000 }, { _RK_TOPIC|_RK_PRODUCER, "message.timeout.ms", _RK_C_INT, _RKT(message_timeout_ms), "Local message timeout. " "This value is only enforced locally and limits the time a " "produced message waits for successful delivery. " "A time of 0 is infinite.", 0, 900*1000, 300*1000 }, { _RK_TOPIC|_RK_PRODUCER, "queuing.strategy", _RK_C_S2I, _RKT(queuing_strategy), "Producer queuing strategy. FIFO preserves produce ordering, " "while LIFO prioritizes new messages. " "WARNING: `lifo` is experimental and subject to change or removal.", .vdef = 0, .s2i = { { RD_KAFKA_QUEUE_FIFO, "fifo" }, { RD_KAFKA_QUEUE_LIFO, "lifo" } } }, { _RK_TOPIC|_RK_PRODUCER, "produce.offset.report", _RK_C_BOOL, _RKT(produce_offset_report), "Report offset of produced message back to application. " "The application must be use the `dr_msg_cb` to retrieve the offset " "from `rd_kafka_message_t.offset`.", 0, 1, 0 }, { _RK_TOPIC|_RK_PRODUCER, "partitioner", _RK_C_STR, _RKT(partitioner_str), "Partitioner: " "`random` - random distribution, " "`consistent` - CRC32 hash of key (Empty and NULL keys are mapped to single partition), " "`consistent_random` - CRC32 hash of key (Empty and NULL keys are randomly partitioned), " "`murmur2` - Java Producer compatible Murmur2 hash of key (NULL keys are mapped to single partition), " "`murmur2_random` - Java Producer compatible Murmur2 hash of key (NULL keys are randomly partitioned. This is functionally equivalent to the default partitioner in the Java Producer.).", .sdef = "consistent_random", .validate = rd_kafka_conf_validate_partitioner }, { _RK_TOPIC|_RK_PRODUCER, "partitioner_cb", _RK_C_PTR, _RKT(partitioner), "Custom partitioner callback " "(set with rd_kafka_topic_conf_set_partitioner_cb())" }, { _RK_TOPIC|_RK_PRODUCER, "msg_order_cmp", _RK_C_PTR, _RKT(msg_order_cmp), "Message queue ordering comparator " "(set with rd_kafka_topic_conf_set_msg_order_cmp()). " "Also see `queuing.strategy`." }, { _RK_TOPIC, "opaque", _RK_C_PTR, _RKT(opaque), "Application opaque (set with rd_kafka_topic_conf_set_opaque())" }, { _RK_TOPIC | _RK_PRODUCER, "compression.codec", _RK_C_S2I, _RKT(compression_codec), "Compression codec to use for compressing message sets. " "inherit = inherit global compression.codec configuration.", .vdef = RD_KAFKA_COMPRESSION_INHERIT, .s2i = { { RD_KAFKA_COMPRESSION_NONE, "none" }, #if WITH_ZLIB { RD_KAFKA_COMPRESSION_GZIP, "gzip" }, #endif #if WITH_SNAPPY { RD_KAFKA_COMPRESSION_SNAPPY, "snappy" }, #endif { RD_KAFKA_COMPRESSION_LZ4, "lz4" }, { RD_KAFKA_COMPRESSION_INHERIT, "inherit" }, { 0 } } }, { _RK_TOPIC | _RK_PRODUCER, "compression.type", _RK_C_ALIAS, .sdef = "compression.codec" }, /* Topic consumer properties */ { _RK_TOPIC|_RK_CONSUMER, "auto.commit.enable", _RK_C_BOOL, _RKT(auto_commit), "If true, periodically commit offset of the last message handed " "to the application. This committed offset will be used when the " "process restarts to pick up where it left off. " "If false, the application will have to call " "`rd_kafka_offset_store()` to store an offset (optional). " "**NOTE:** This property should only be used with the simple " "legacy consumer, when using the high-level KafkaConsumer the global " "`enable.auto.commit` property must be used instead. " "**NOTE:** There is currently no zookeeper integration, offsets " "will be written to broker or local file according to " "offset.store.method.", 0, 1, 1 }, { _RK_TOPIC|_RK_CONSUMER, "enable.auto.commit", _RK_C_ALIAS, .sdef = "auto.commit.enable" }, { _RK_TOPIC|_RK_CONSUMER, "auto.commit.interval.ms", _RK_C_INT, _RKT(auto_commit_interval_ms), "The frequency in milliseconds that the consumer offsets " "are committed (written) to offset storage. " "This setting is used by the low-level legacy consumer.", 10, 86400*1000, 60*1000 }, { _RK_TOPIC|_RK_CONSUMER, "auto.offset.reset", _RK_C_S2I, _RKT(auto_offset_reset), "Action to take when there is no initial offset in offset store " "or the desired offset is out of range: " "'smallest','earliest' - automatically reset the offset to the smallest offset, " "'largest','latest' - automatically reset the offset to the largest offset, " "'error' - trigger an error which is retrieved by consuming messages " "and checking 'message->err'.", .vdef = RD_KAFKA_OFFSET_END, .s2i = { { RD_KAFKA_OFFSET_BEGINNING, "smallest" }, { RD_KAFKA_OFFSET_BEGINNING, "earliest" }, { RD_KAFKA_OFFSET_BEGINNING, "beginning" }, { RD_KAFKA_OFFSET_END, "largest" }, { RD_KAFKA_OFFSET_END, "latest" }, { RD_KAFKA_OFFSET_END, "end" }, { RD_KAFKA_OFFSET_INVALID, "error" }, } }, { _RK_TOPIC|_RK_CONSUMER, "offset.store.path", _RK_C_STR, _RKT(offset_store_path), "Path to local file for storing offsets. If the path is a directory " "a filename will be automatically generated in that directory based " "on the topic and partition.", .sdef = "." }, { _RK_TOPIC|_RK_CONSUMER, "offset.store.sync.interval.ms", _RK_C_INT, _RKT(offset_store_sync_interval_ms), "fsync() interval for the offset file, in milliseconds. " "Use -1 to disable syncing, and 0 for immediate sync after " "each write.", -1, 86400*1000, -1 }, { _RK_TOPIC|_RK_CONSUMER, "offset.store.method", _RK_C_S2I, _RKT(offset_store_method), "Offset commit store method: " "'file' - local file store (offset.store.path, et.al), " "'broker' - broker commit store " "(requires \"group.id\" to be configured and " "Apache Kafka 0.8.2 or later on the broker.).", .vdef = RD_KAFKA_OFFSET_METHOD_BROKER, .s2i = { { RD_KAFKA_OFFSET_METHOD_FILE, "file" }, { RD_KAFKA_OFFSET_METHOD_BROKER, "broker" } } }, { _RK_TOPIC|_RK_CONSUMER, "consume.callback.max.messages", _RK_C_INT, _RKT(consume_callback_max_msgs), "Maximum number of messages to dispatch in " "one `rd_kafka_consume_callback*()` call (0 = unlimited)", 0, 1000000, 0 }, { 0, /* End */ } }; static rd_kafka_conf_res_t rd_kafka_anyconf_set_prop0 (int scope, void *conf, const struct rd_kafka_property *prop, const char *istr, int ival, rd_kafka_conf_set_mode_t set_mode, char *errstr, size_t errstr_size) { rd_kafka_conf_res_t res; #define _RK_PTR(TYPE,BASE,OFFSET) (TYPE)(void *)(((char *)(BASE))+(OFFSET)) /* Try interceptors first (only for GLOBAL config) */ if (scope & _RK_GLOBAL) { if (prop->type == _RK_C_PTR || prop->type == _RK_C_INTERNAL) res = RD_KAFKA_CONF_UNKNOWN; else res = rd_kafka_interceptors_on_conf_set(conf, prop->name, istr, errstr, errstr_size); if (res != RD_KAFKA_CONF_UNKNOWN) return res; } if (prop->set) { /* Custom setter */ rd_kafka_conf_res_t res; res = prop->set(scope, conf, prop->name, istr, _RK_PTR(void *, conf, prop->offset), set_mode, errstr, errstr_size); if (res != RD_KAFKA_CONF_OK) return res; /* FALLTHRU so that property value is set. */ } switch (prop->type) { case _RK_C_STR: { char **str = _RK_PTR(char **, conf, prop->offset); if (*str) rd_free(*str); if (istr) *str = rd_strdup(istr); else *str = prop->sdef ? rd_strdup(prop->sdef) : NULL; return RD_KAFKA_CONF_OK; } case _RK_C_KSTR: { rd_kafkap_str_t **kstr = _RK_PTR(rd_kafkap_str_t **, conf, prop->offset); if (*kstr) rd_kafkap_str_destroy(*kstr); if (istr) *kstr = rd_kafkap_str_new(istr, -1); else *kstr = prop->sdef ? rd_kafkap_str_new(prop->sdef, -1) : NULL; return RD_KAFKA_CONF_OK; } case _RK_C_PTR: *_RK_PTR(const void **, conf, prop->offset) = istr; return RD_KAFKA_CONF_OK; case _RK_C_BOOL: case _RK_C_INT: case _RK_C_S2I: case _RK_C_S2F: { int *val = _RK_PTR(int *, conf, prop->offset); if (prop->type == _RK_C_S2F) { switch (set_mode) { case _RK_CONF_PROP_SET_REPLACE: *val = ival; break; case _RK_CONF_PROP_SET_ADD: *val |= ival; break; case _RK_CONF_PROP_SET_DEL: *val &= ~ival; break; } } else { /* Single assignment */ *val = ival; } return RD_KAFKA_CONF_OK; } case _RK_C_PATLIST: { /* Split comma-separated list into individual regex expressions * that are verified and then append to the provided list. */ rd_kafka_pattern_list_t **plist; plist = _RK_PTR(rd_kafka_pattern_list_t **, conf, prop->offset); if (*plist) rd_kafka_pattern_list_destroy(*plist); if (istr) { if (!(*plist = rd_kafka_pattern_list_new(istr, errstr, (int)errstr_size))) return RD_KAFKA_CONF_INVALID; } else *plist = NULL; return RD_KAFKA_CONF_OK; } case _RK_C_INTERNAL: /* Probably handled by setter */ return RD_KAFKA_CONF_OK; default: rd_kafka_assert(NULL, !*"unknown conf type"); } /* unreachable */ return RD_KAFKA_CONF_INVALID; } /** * @brief Find s2i (string-to-int mapping) entry and return its array index, * or -1 on miss. */ static int rd_kafka_conf_s2i_find (const struct rd_kafka_property *prop, const char *value) { int j; for (j = 0 ; j < (int)RD_ARRAYSIZE(prop->s2i); j++) { if (prop->s2i[j].str && !rd_strcasecmp(prop->s2i[j].str, value)) return j; } return -1; } static rd_kafka_conf_res_t rd_kafka_anyconf_set_prop (int scope, void *conf, const struct rd_kafka_property *prop, const char *value, char *errstr, size_t errstr_size) { int ival; switch (prop->type) { case _RK_C_STR: case _RK_C_KSTR: if (prop->s2i[0].str) { int match; if (!value || (match = rd_kafka_conf_s2i_find(prop, value)) == -1){ rd_snprintf(errstr, errstr_size, "Invalid value for " "configuration property \"%s\": " "%s", prop->name, value); return RD_KAFKA_CONF_INVALID; } /* Replace value string with canonical form */ value = prop->s2i[match].str; } /* FALLTHRU */ case _RK_C_PATLIST: if (prop->validate && (!value || !prop->validate(prop, value, -1))) { rd_snprintf(errstr, errstr_size, "Invalid value for " "configuration property \"%s\": %s", prop->name, value); return RD_KAFKA_CONF_INVALID; } return rd_kafka_anyconf_set_prop0(scope, conf, prop, value, 0, _RK_CONF_PROP_SET_REPLACE, errstr, errstr_size); case _RK_C_PTR: rd_snprintf(errstr, errstr_size, "Property \"%s\" must be set through dedicated " ".._set_..() function", prop->name); return RD_KAFKA_CONF_INVALID; case _RK_C_BOOL: if (!value) { rd_snprintf(errstr, errstr_size, "Bool configuration property \"%s\" cannot " "be set to empty value", prop->name); return RD_KAFKA_CONF_INVALID; } if (!rd_strcasecmp(value, "true") || !rd_strcasecmp(value, "t") || !strcmp(value, "1")) ival = 1; else if (!rd_strcasecmp(value, "false") || !rd_strcasecmp(value, "f") || !strcmp(value, "0")) ival = 0; else { rd_snprintf(errstr, errstr_size, "Expected bool value for \"%s\": " "true or false", prop->name); return RD_KAFKA_CONF_INVALID; } rd_kafka_anyconf_set_prop0(scope, conf, prop, value, ival, _RK_CONF_PROP_SET_REPLACE, errstr, errstr_size); return RD_KAFKA_CONF_OK; case _RK_C_INT: { const char *end; if (!value) { rd_snprintf(errstr, errstr_size, "Integer configuration " "property \"%s\" cannot be set " "to empty value", prop->name); return RD_KAFKA_CONF_INVALID; } ival = (int)strtol(value, (char **)&end, 0); if (end == value) { /* Non numeric, check s2i for string mapping */ int match = rd_kafka_conf_s2i_find(prop, value); if (match == -1) { rd_snprintf(errstr, errstr_size, "Invalid value for " "configuration property \"%s\"", prop->name); return RD_KAFKA_CONF_INVALID; } ival = prop->s2i[match].val; } if (ival < prop->vmin || ival > prop->vmax) { rd_snprintf(errstr, errstr_size, "Configuration property \"%s\" value " "%i is outside allowed range %i..%i\n", prop->name, ival, prop->vmin, prop->vmax); return RD_KAFKA_CONF_INVALID; } rd_kafka_anyconf_set_prop0(scope, conf, prop, value, ival, _RK_CONF_PROP_SET_REPLACE, errstr, errstr_size); return RD_KAFKA_CONF_OK; } case _RK_C_S2I: case _RK_C_S2F: { int j; const char *next; if (!value) { rd_snprintf(errstr, errstr_size, "Configuration " "property \"%s\" cannot be set " "to empty value", prop->name); return RD_KAFKA_CONF_INVALID; } next = value; while (next && *next) { const char *s, *t; rd_kafka_conf_set_mode_t set_mode = _RK_CONF_PROP_SET_ADD; /* S2F */ s = next; if (prop->type == _RK_C_S2F && (t = strchr(s, ','))) { /* CSV flag field */ next = t+1; } else { /* Single string */ t = s+strlen(s); next = NULL; } /* Left trim */ while (s < t && isspace((int)*s)) s++; /* Right trim */ while (t > s && isspace((int)*t)) t--; /* S2F: +/- prefix */ if (prop->type == _RK_C_S2F) { if (*s == '+') { set_mode = _RK_CONF_PROP_SET_ADD; s++; } else if (*s == '-') { set_mode = _RK_CONF_PROP_SET_DEL; s++; } } /* Empty string? */ if (s == t) continue; /* Match string to s2i table entry */ for (j = 0 ; j < (int)RD_ARRAYSIZE(prop->s2i); j++) { int new_val; if (!prop->s2i[j].str) continue; if (strlen(prop->s2i[j].str) == (size_t)(t-s) && !rd_strncasecmp(prop->s2i[j].str, s, (int)(t-s))) new_val = prop->s2i[j].val; else continue; rd_kafka_anyconf_set_prop0(scope, conf, prop, value, new_val, set_mode, errstr, errstr_size); if (prop->type == _RK_C_S2F) { /* Flags: OR it in: do next */ break; } else { /* Single assignment */ return RD_KAFKA_CONF_OK; } } /* S2F: Good match: continue with next */ if (j < (int)RD_ARRAYSIZE(prop->s2i)) continue; /* No match */ rd_snprintf(errstr, errstr_size, "Invalid value for " "configuration property \"%s\"", prop->name); return RD_KAFKA_CONF_INVALID; } return RD_KAFKA_CONF_OK; } case _RK_C_INTERNAL: rd_snprintf(errstr, errstr_size, "Internal property \"%s\" not settable", prop->name); return RD_KAFKA_CONF_INVALID; case _RK_C_INVALID: rd_snprintf(errstr, errstr_size, "%s", prop->desc); return RD_KAFKA_CONF_INVALID; default: rd_kafka_assert(NULL, !*"unknown conf type"); } /* not reachable */ return RD_KAFKA_CONF_INVALID; } static void rd_kafka_defaultconf_set (int scope, void *conf) { const struct rd_kafka_property *prop; for (prop = rd_kafka_properties ; prop->name ; prop++) { if (!(prop->scope & scope)) continue; if (prop->type == _RK_C_ALIAS || prop->type == _RK_C_INVALID) continue; if (prop->ctor) prop->ctor(scope, conf); if (prop->sdef || prop->vdef || prop->pdef) rd_kafka_anyconf_set_prop0(scope, conf, prop, prop->sdef ? prop->sdef : prop->pdef, prop->vdef, _RK_CONF_PROP_SET_REPLACE, NULL, 0); } } rd_kafka_conf_t *rd_kafka_conf_new (void) { rd_kafka_conf_t *conf = rd_calloc(1, sizeof(*conf)); rd_kafka_defaultconf_set(_RK_GLOBAL, conf); return conf; } rd_kafka_topic_conf_t *rd_kafka_topic_conf_new (void) { rd_kafka_topic_conf_t *tconf = rd_calloc(1, sizeof(*tconf)); rd_kafka_defaultconf_set(_RK_TOPIC, tconf); return tconf; } static int rd_kafka_anyconf_set (int scope, void *conf, const char *name, const char *value, char *errstr, size_t errstr_size) { char estmp[1]; const struct rd_kafka_property *prop; rd_kafka_conf_res_t res; if (!errstr) { errstr = estmp; errstr_size = 0; } if (value && !*value) value = NULL; /* Try interceptors first (only for GLOBAL config for now) */ if (scope & _RK_GLOBAL) { res = rd_kafka_interceptors_on_conf_set( (rd_kafka_conf_t *)conf, name, value, errstr, errstr_size); /* Handled (successfully or not) by interceptor. */ if (res != RD_KAFKA_CONF_UNKNOWN) return res; } /* Then global config */ for (prop = rd_kafka_properties ; prop->name ; prop++) { if (!(prop->scope & scope)) continue; if (strcmp(prop->name, name)) continue; if (prop->type == _RK_C_ALIAS) return rd_kafka_anyconf_set(scope, conf, prop->sdef, value, errstr, errstr_size); return rd_kafka_anyconf_set_prop(scope, conf, prop, value, errstr, errstr_size); } rd_snprintf(errstr, errstr_size, "No such configuration property: \"%s\"", name); return RD_KAFKA_CONF_UNKNOWN; } rd_kafka_conf_res_t rd_kafka_conf_set (rd_kafka_conf_t *conf, const char *name, const char *value, char *errstr, size_t errstr_size) { rd_kafka_conf_res_t res; res = rd_kafka_anyconf_set(_RK_GLOBAL, conf, name, value, errstr, errstr_size); if (res != RD_KAFKA_CONF_UNKNOWN) return res; /* Fallthru: * If the global property was unknown, try setting it on the * default topic config. */ if (!conf->topic_conf) { /* Create topic config, might be over-written by application * later. */ conf->topic_conf = rd_kafka_topic_conf_new(); } return rd_kafka_topic_conf_set(conf->topic_conf, name, value, errstr, errstr_size); } rd_kafka_conf_res_t rd_kafka_topic_conf_set (rd_kafka_topic_conf_t *conf, const char *name, const char *value, char *errstr, size_t errstr_size) { if (!strncmp(name, "topic.", strlen("topic."))) name += strlen("topic."); return rd_kafka_anyconf_set(_RK_TOPIC, conf, name, value, errstr, errstr_size); } static void rd_kafka_anyconf_clear (int scope, void *conf, const struct rd_kafka_property *prop) { switch (prop->type) { case _RK_C_STR: { char **str = _RK_PTR(char **, conf, prop->offset); if (*str) { if (prop->set) { prop->set(scope, conf, prop->name, NULL, *str, _RK_CONF_PROP_SET_DEL, NULL, 0); /* FALLTHRU */ } rd_free(*str); *str = NULL; } } break; case _RK_C_KSTR: { rd_kafkap_str_t **kstr = _RK_PTR(rd_kafkap_str_t **, conf, prop->offset); if (*kstr) { rd_kafkap_str_destroy(*kstr); *kstr = NULL; } } break; case _RK_C_PATLIST: { rd_kafka_pattern_list_t **plist; plist = _RK_PTR(rd_kafka_pattern_list_t **, conf, prop->offset); if (*plist) { rd_kafka_pattern_list_destroy(*plist); *plist = NULL; } } break; case _RK_C_PTR: if (_RK_PTR(void *, conf, prop->offset) != NULL) { if (!strcmp(prop->name, "default_topic_conf")) { rd_kafka_topic_conf_t **tconf; tconf = _RK_PTR(rd_kafka_topic_conf_t **, conf, prop->offset); if (*tconf) { rd_kafka_topic_conf_destroy(*tconf); *tconf = NULL; } } } break; default: break; } if (prop->dtor) prop->dtor(scope, conf); } void rd_kafka_anyconf_destroy (int scope, void *conf) { const struct rd_kafka_property *prop; /* Call on_conf_destroy() interceptors */ if (scope == _RK_GLOBAL) rd_kafka_interceptors_on_conf_destroy(conf); for (prop = rd_kafka_properties; prop->name ; prop++) { if (!(prop->scope & scope)) continue; rd_kafka_anyconf_clear(scope, conf, prop); } } void rd_kafka_conf_destroy (rd_kafka_conf_t *conf) { rd_kafka_anyconf_destroy(_RK_GLOBAL, conf); //FIXME: partition_assignors rd_free(conf); } void rd_kafka_topic_conf_destroy (rd_kafka_topic_conf_t *topic_conf) { rd_kafka_anyconf_destroy(_RK_TOPIC, topic_conf); rd_free(topic_conf); } static void rd_kafka_anyconf_copy (int scope, void *dst, const void *src, size_t filter_cnt, const char **filter) { const struct rd_kafka_property *prop; for (prop = rd_kafka_properties ; prop->name ; prop++) { const char *val = NULL; int ival = 0; char *valstr; size_t valsz; size_t fi; size_t nlen; if (!(prop->scope & scope)) continue; if (prop->type == _RK_C_ALIAS || prop->type == _RK_C_INVALID) continue; /* Apply filter, if any. */ nlen = strlen(prop->name); for (fi = 0 ; fi < filter_cnt ; fi++) { size_t flen = strlen(filter[fi]); if (nlen >= flen && !strncmp(filter[fi], prop->name, flen)) break; } if (fi < filter_cnt) continue; /* Filter matched */ switch (prop->type) { case _RK_C_STR: case _RK_C_PTR: val = *_RK_PTR(const char **, src, prop->offset); if (!strcmp(prop->name, "default_topic_conf") && val) val = (void *)rd_kafka_topic_conf_dup( (const rd_kafka_topic_conf_t *) (void *)val); break; case _RK_C_KSTR: { rd_kafkap_str_t **kstr = _RK_PTR(rd_kafkap_str_t **, src, prop->offset); if (*kstr) val = (*kstr)->str; break; } case _RK_C_BOOL: case _RK_C_INT: case _RK_C_S2I: case _RK_C_S2F: ival = *_RK_PTR(const int *, src, prop->offset); /* Get string representation of configuration value. */ valsz = 0; rd_kafka_anyconf_get0(src, prop, NULL, &valsz); valstr = rd_alloca(valsz); rd_kafka_anyconf_get0(src, prop, valstr, &valsz); val = valstr; break; case _RK_C_PATLIST: { const rd_kafka_pattern_list_t **plist; plist = _RK_PTR(const rd_kafka_pattern_list_t **, src, prop->offset); if (*plist) val = (*plist)->rkpl_orig; break; } case _RK_C_INTERNAL: /* Handled by ->copy() below. */ break; default: continue; } if (prop->copy) prop->copy(scope, dst, src, _RK_PTR(void *, dst, prop->offset), _RK_PTR(const void *, src, prop->offset), filter_cnt, filter); rd_kafka_anyconf_set_prop0(scope, dst, prop, val, ival, _RK_CONF_PROP_SET_REPLACE, NULL, 0); } } rd_kafka_conf_t *rd_kafka_conf_dup (const rd_kafka_conf_t *conf) { rd_kafka_conf_t *new = rd_kafka_conf_new(); rd_kafka_interceptors_on_conf_dup(new, conf, 0, NULL); rd_kafka_anyconf_copy(_RK_GLOBAL, new, conf, 0, NULL); return new; } rd_kafka_conf_t *rd_kafka_conf_dup_filter (const rd_kafka_conf_t *conf, size_t filter_cnt, const char **filter) { rd_kafka_conf_t *new = rd_kafka_conf_new(); rd_kafka_interceptors_on_conf_dup(new, conf, filter_cnt, filter); rd_kafka_anyconf_copy(_RK_GLOBAL, new, conf, filter_cnt, filter); return new; } rd_kafka_topic_conf_t *rd_kafka_topic_conf_dup (const rd_kafka_topic_conf_t *conf) { rd_kafka_topic_conf_t *new = rd_kafka_topic_conf_new(); rd_kafka_anyconf_copy(_RK_TOPIC, new, conf, 0, NULL); return new; } rd_kafka_topic_conf_t *rd_kafka_default_topic_conf_dup (rd_kafka_t *rk) { if (rk->rk_conf.topic_conf) return rd_kafka_topic_conf_dup(rk->rk_conf.topic_conf); else return rd_kafka_topic_conf_new(); } void rd_kafka_conf_set_events (rd_kafka_conf_t *conf, int events) { conf->enabled_events = events; } void rd_kafka_conf_set_dr_cb (rd_kafka_conf_t *conf, void (*dr_cb) (rd_kafka_t *rk, void *payload, size_t len, rd_kafka_resp_err_t err, void *opaque, void *msg_opaque)) { conf->dr_cb = dr_cb; } void rd_kafka_conf_set_dr_msg_cb (rd_kafka_conf_t *conf, void (*dr_msg_cb) (rd_kafka_t *rk, const rd_kafka_message_t * rkmessage, void *opaque)) { conf->dr_msg_cb = dr_msg_cb; } void rd_kafka_conf_set_consume_cb (rd_kafka_conf_t *conf, void (*consume_cb) (rd_kafka_message_t * rkmessage, void *opaque)) { conf->consume_cb = consume_cb; } void rd_kafka_conf_set_rebalance_cb ( rd_kafka_conf_t *conf, void (*rebalance_cb) (rd_kafka_t *rk, rd_kafka_resp_err_t err, rd_kafka_topic_partition_list_t *partitions, void *opaque)) { conf->rebalance_cb = rebalance_cb; } void rd_kafka_conf_set_offset_commit_cb ( rd_kafka_conf_t *conf, void (*offset_commit_cb) (rd_kafka_t *rk, rd_kafka_resp_err_t err, rd_kafka_topic_partition_list_t *offsets, void *opaque)) { conf->offset_commit_cb = offset_commit_cb; } void rd_kafka_conf_set_error_cb (rd_kafka_conf_t *conf, void (*error_cb) (rd_kafka_t *rk, int err, const char *reason, void *opaque)) { conf->error_cb = error_cb; } void rd_kafka_conf_set_throttle_cb (rd_kafka_conf_t *conf, void (*throttle_cb) ( rd_kafka_t *rk, const char *broker_name, int32_t broker_id, int throttle_time_ms, void *opaque)) { conf->throttle_cb = throttle_cb; } void rd_kafka_conf_set_log_cb (rd_kafka_conf_t *conf, void (*log_cb) (const rd_kafka_t *rk, int level, const char *fac, const char *buf)) { conf->log_cb = log_cb; } void rd_kafka_conf_set_stats_cb (rd_kafka_conf_t *conf, int (*stats_cb) (rd_kafka_t *rk, char *json, size_t json_len, void *opaque)) { conf->stats_cb = stats_cb; } void rd_kafka_conf_set_socket_cb (rd_kafka_conf_t *conf, int (*socket_cb) (int domain, int type, int protocol, void *opaque)) { conf->socket_cb = socket_cb; } void rd_kafka_conf_set_connect_cb (rd_kafka_conf_t *conf, int (*connect_cb) (int sockfd, const struct sockaddr *addr, int addrlen, const char *id, void *opaque)) { conf->connect_cb = connect_cb; } void rd_kafka_conf_set_closesocket_cb (rd_kafka_conf_t *conf, int (*closesocket_cb) (int sockfd, void *opaque)) { conf->closesocket_cb = closesocket_cb; } #ifndef _MSC_VER void rd_kafka_conf_set_open_cb (rd_kafka_conf_t *conf, int (*open_cb) (const char *pathname, int flags, mode_t mode, void *opaque)) { conf->open_cb = open_cb; } #endif void rd_kafka_conf_set_opaque (rd_kafka_conf_t *conf, void *opaque) { conf->opaque = opaque; } void rd_kafka_conf_set_default_topic_conf (rd_kafka_conf_t *conf, rd_kafka_topic_conf_t *tconf) { if (conf->topic_conf) rd_kafka_topic_conf_destroy(conf->topic_conf); conf->topic_conf = tconf; } void rd_kafka_topic_conf_set_partitioner_cb (rd_kafka_topic_conf_t *topic_conf, int32_t (*partitioner) ( const rd_kafka_topic_t *rkt, const void *keydata, size_t keylen, int32_t partition_cnt, void *rkt_opaque, void *msg_opaque)) { topic_conf->partitioner = partitioner; } void rd_kafka_topic_conf_set_msg_order_cmp (rd_kafka_topic_conf_t *topic_conf, int (*msg_order_cmp) ( const rd_kafka_message_t *a, const rd_kafka_message_t *b)) { topic_conf->msg_order_cmp = (int (*)(const void *, const void *))msg_order_cmp; } void rd_kafka_topic_conf_set_opaque (rd_kafka_topic_conf_t *topic_conf, void *opaque) { topic_conf->opaque = opaque; } /** * @brief Convert flags \p ival to csv-string using S2F property \p prop. * * This function has two modes: size query and write. * To query for needed size call with dest==NULL, * to write to buffer of size dest_size call with dest!=NULL. * * An \p ival of -1 means all. * * @returns the number of bytes written to \p dest (if not NULL), else the * total number of bytes needed. * */ size_t rd_kafka_conf_flags2str (char *dest, size_t dest_size, const char *delim, const struct rd_kafka_property *prop, int ival) { size_t of = 0; int j; if (dest && dest_size > 0) *dest = '\0'; /* Phase 1: scan for set flags, accumulate needed size. * Phase 2: write to dest */ for (j = 0 ; prop->s2i[j].str ; j++) { if (prop->type == _RK_C_S2F && ival != -1 && (ival & prop->s2i[j].val) != prop->s2i[j].val) continue; else if (prop->type == _RK_C_S2I && ival != -1 && prop->s2i[j].val != ival) continue; if (!dest) of += strlen(prop->s2i[j].str) + (of > 0 ? 1 : 0); else { size_t r; r = rd_snprintf(dest+of, dest_size-of, "%s%s", of > 0 ? delim:"", prop->s2i[j].str); if (r > dest_size-of) { r = dest_size-of; break; } of += r; } } return of+1/*nul*/; } /** * Return "original"(re-created) configuration value string */ static rd_kafka_conf_res_t rd_kafka_anyconf_get0 (const void *conf, const struct rd_kafka_property *prop, char *dest, size_t *dest_size) { char tmp[22]; const char *val = NULL; size_t val_len = 0; int j; switch (prop->type) { case _RK_C_STR: val = *_RK_PTR(const char **, conf, prop->offset); break; case _RK_C_KSTR: { const rd_kafkap_str_t **kstr = _RK_PTR(const rd_kafkap_str_t **, conf, prop->offset); if (*kstr) val = (*kstr)->str; break; } case _RK_C_PTR: val = *_RK_PTR(const void **, conf, prop->offset); if (val) { rd_snprintf(tmp, sizeof(tmp), "%p", (void *)val); val = tmp; } break; case _RK_C_BOOL: val = (*_RK_PTR(int *, conf, prop->offset) ? "true" : "false"); break; case _RK_C_INT: rd_snprintf(tmp, sizeof(tmp), "%i", *_RK_PTR(int *, conf, prop->offset)); val = tmp; break; case _RK_C_S2I: for (j = 0 ; j < (int)RD_ARRAYSIZE(prop->s2i); j++) { if (prop->s2i[j].val == *_RK_PTR(int *, conf, prop->offset)) { val = prop->s2i[j].str; break; } } break; case _RK_C_S2F: { const int ival = *_RK_PTR(const int *, conf, prop->offset); val_len = rd_kafka_conf_flags2str(dest, *dest_size, ",", prop, ival); if (dest) { val_len = 0; val = dest; dest = NULL; } break; } case _RK_C_PATLIST: { const rd_kafka_pattern_list_t **plist; plist = _RK_PTR(const rd_kafka_pattern_list_t **, conf, prop->offset); if (*plist) val = (*plist)->rkpl_orig; break; } default: break; } if (val_len) { *dest_size = val_len+1; return RD_KAFKA_CONF_OK; } if (!val) return RD_KAFKA_CONF_INVALID; val_len = strlen(val); if (dest) { size_t use_len = RD_MIN(val_len, (*dest_size)-1); memcpy(dest, val, use_len); dest[use_len] = '\0'; } /* Return needed size */ *dest_size = val_len+1; return RD_KAFKA_CONF_OK; } static rd_kafka_conf_res_t rd_kafka_anyconf_get (int scope, const void *conf, const char *name, char *dest, size_t *dest_size){ const struct rd_kafka_property *prop; for (prop = rd_kafka_properties; prop->name ; prop++) { if (!(prop->scope & scope) || strcmp(prop->name, name)) continue; if (prop->type == _RK_C_ALIAS) return rd_kafka_anyconf_get(scope, conf, prop->sdef, dest, dest_size); if (rd_kafka_anyconf_get0(conf, prop, dest, dest_size) == RD_KAFKA_CONF_OK) return RD_KAFKA_CONF_OK; } return RD_KAFKA_CONF_UNKNOWN; } rd_kafka_conf_res_t rd_kafka_topic_conf_get (const rd_kafka_topic_conf_t *conf, const char *name, char *dest, size_t *dest_size) { return rd_kafka_anyconf_get(_RK_TOPIC, conf, name, dest, dest_size); } rd_kafka_conf_res_t rd_kafka_conf_get (const rd_kafka_conf_t *conf, const char *name, char *dest, size_t *dest_size) { rd_kafka_conf_res_t res; res = rd_kafka_anyconf_get(_RK_GLOBAL, conf, name, dest, dest_size); if (res != RD_KAFKA_CONF_UNKNOWN || !conf->topic_conf) return res; /* Fallthru: * If the global property was unknown, try getting it from the * default topic config, if any. */ return rd_kafka_topic_conf_get(conf->topic_conf, name, dest, dest_size); } static const char **rd_kafka_anyconf_dump (int scope, const void *conf, size_t *cntp) { const struct rd_kafka_property *prop; char **arr; int cnt = 0; arr = rd_calloc(sizeof(char *), RD_ARRAYSIZE(rd_kafka_properties)*2); for (prop = rd_kafka_properties; prop->name ; prop++) { char *val = NULL; size_t val_size; if (!(prop->scope & scope)) continue; /* Skip aliases, show original property instead. * Skip invalids. */ if (prop->type == _RK_C_ALIAS || prop->type == _RK_C_INVALID) continue; /* Query value size */ if (rd_kafka_anyconf_get0(conf, prop, NULL, &val_size) != RD_KAFKA_CONF_OK) continue; /* Get value */ val = malloc(val_size); rd_kafka_anyconf_get0(conf, prop, val, &val_size); arr[cnt++] = rd_strdup(prop->name); arr[cnt++] = val; } *cntp = cnt; return (const char **)arr; } const char **rd_kafka_conf_dump (rd_kafka_conf_t *conf, size_t *cntp) { return rd_kafka_anyconf_dump(_RK_GLOBAL, conf, cntp); } const char **rd_kafka_topic_conf_dump (rd_kafka_topic_conf_t *conf, size_t *cntp) { return rd_kafka_anyconf_dump(_RK_TOPIC, conf, cntp); } void rd_kafka_conf_dump_free (const char **arr, size_t cnt) { char **_arr = (char **)arr; unsigned int i; for (i = 0 ; i < cnt ; i++) if (_arr[i]) rd_free(_arr[i]); rd_free(_arr); } void rd_kafka_conf_properties_show (FILE *fp) { const struct rd_kafka_property *prop; int last = 0; int j; char tmp[512]; const char *dash80 = "----------------------------------------" "----------------------------------------"; for (prop = rd_kafka_properties; prop->name ; prop++) { const char *typeinfo = ""; /* Skip invalid properties. */ if (prop->type == _RK_C_INVALID) continue; if (!(prop->scope & last)) { fprintf(fp, "%s## %s configuration properties\n\n", last ? "\n\n":"", prop->scope == _RK_GLOBAL ? "Global": "Topic"); fprintf(fp, "%-40s | %3s | %-15s | %13s | %-25s\n" "%.*s-|-%.*s-|-%.*s-|-%.*s:|-%.*s\n", "Property", "C/P", "Range", "Default", "Description", 40, dash80, 3, dash80, 15, dash80, 13, dash80, 25, dash80); last = prop->scope & (_RK_GLOBAL|_RK_TOPIC); } fprintf(fp, "%-40s | %3s | ", prop->name, (!(prop->scope & _RK_PRODUCER) == !(prop->scope & _RK_CONSUMER) ? " * " : ((prop->scope & _RK_PRODUCER) ? " P " : (prop->scope & _RK_CONSUMER) ? " C " : ""))); switch (prop->type) { case _RK_C_STR: case _RK_C_KSTR: typeinfo = "string"; case _RK_C_PATLIST: if (prop->type == _RK_C_PATLIST) typeinfo = "pattern list"; if (prop->s2i[0].str) { rd_kafka_conf_flags2str(tmp, sizeof(tmp), ", ", prop, -1); fprintf(fp, "%-15s | %13s", tmp, prop->sdef ? prop->sdef : ""); } else { fprintf(fp, "%-15s | %13s", "", prop->sdef ? prop->sdef : ""); } break; case _RK_C_BOOL: typeinfo = "boolean"; fprintf(fp, "%-15s | %13s", "true, false", prop->vdef ? "true" : "false"); break; case _RK_C_INT: typeinfo = "integer"; rd_snprintf(tmp, sizeof(tmp), "%d .. %d", prop->vmin, prop->vmax); fprintf(fp, "%-15s | %13i", tmp, prop->vdef); break; case _RK_C_S2I: typeinfo = "enum value"; rd_kafka_conf_flags2str(tmp, sizeof(tmp), ", ", prop, -1); fprintf(fp, "%-15s | ", tmp); for (j = 0 ; j < (int)RD_ARRAYSIZE(prop->s2i); j++) { if (prop->s2i[j].val == prop->vdef) { fprintf(fp, "%13s", prop->s2i[j].str); break; } } if (j == RD_ARRAYSIZE(prop->s2i)) fprintf(fp, "%13s", " "); break; case _RK_C_S2F: typeinfo = "CSV flags"; /* Dont duplicate builtin.features value in * both Range and Default */ if (!strcmp(prop->name, "builtin.features")) *tmp = '\0'; else rd_kafka_conf_flags2str(tmp, sizeof(tmp), ", ", prop, -1); fprintf(fp, "%-15s | ", tmp); rd_kafka_conf_flags2str(tmp, sizeof(tmp), ", ", prop, prop->vdef); fprintf(fp, "%13s", tmp); break; case _RK_C_PTR: typeinfo = "pointer"; /* FALLTHRU */ default: fprintf(fp, "%-15s | %-13s", "", " "); break; } if (prop->type == _RK_C_ALIAS) fprintf(fp, " | Alias for `%s`\n", prop->sdef); else fprintf(fp, " | %s
*Type: %s*\n", prop->desc, typeinfo); } fprintf(fp, "\n"); fprintf(fp, "### C/P legend: C = Consumer, P = Producer, * = both\n"); }